@@ -1526,42 +1526,6 @@ def test_chaining_with_parameters(cursor, db_connection):
15261526 except :
15271527 pass
15281528
1529- def test_chaining_with_iteration (cursor , db_connection ):
1530- """Test method chaining with iteration (for loop)"""
1531- try :
1532- # Create test table
1533- cursor .execute ("CREATE TABLE #test_iteration (id INT, name NVARCHAR(50))" )
1534- db_connection .commit ()
1535-
1536- # Insert test data
1537- names = ["Alice" , "Bob" , "Charlie" , "Diana" ]
1538- for i , name in enumerate (names , 1 ):
1539- cursor .execute ("INSERT INTO #test_iteration VALUES (?, ?)" , i , name )
1540- db_connection .commit ()
1541-
1542- # Test iteration over execute() result (should work because cursor implements __iter__)
1543- results = []
1544- for row in cursor .execute ("SELECT id, name FROM #test_iteration ORDER BY id" ):
1545- results .append ((row [0 ], row [1 ]))
1546-
1547- expected = [(1 , "Alice" ), (2 , "Bob" ), (3 , "Charlie" ), (4 , "Diana" )]
1548- assert results == expected , f"Iteration results should match expected: { results } != { expected } "
1549-
1550- # Test iteration with WHERE clause
1551- results = []
1552- for row in cursor .execute ("SELECT name FROM #test_iteration WHERE id > ?" , 2 ):
1553- results .append (row [0 ])
1554-
1555- expected_names = ["Charlie" , "Diana" ]
1556- assert results == expected_names , f"Filtered iteration should return: { expected_names } , got: { results } "
1557-
1558- finally :
1559- try :
1560- cursor .execute ("DROP TABLE #test_iteration" )
1561- db_connection .commit ()
1562- except :
1563- pass
1564-
15651529def test_chaining_error_handling (cursor ):
15661530 """Test that chaining works properly even when errors occur"""
15671531 # Test that cursor is still chainable after an error
@@ -1613,66 +1577,6 @@ def test_chaining_performance_statement_reuse(cursor, db_connection):
16131577 except :
16141578 pass
16151579
1616- def test_execute_chaining_compatibility_examples (cursor , db_connection ):
1617- """Test real-world pyodbc-style chaining examples"""
1618- try :
1619- # Create users table
1620- cursor .execute ("""
1621- CREATE TABLE #users (
1622- user_id INT IDENTITY(1,1) PRIMARY KEY,
1623- user_name NVARCHAR(50),
1624- last_logon DATETIME,
1625- status NVARCHAR(20)
1626- )
1627- """ )
1628- db_connection .commit ()
1629-
1630- # Insert test users
1631- cursor .execute ("INSERT INTO #users (user_name, status) VALUES ('john_doe', 'active')" )
1632- cursor .execute ("INSERT INTO #users (user_name, status) VALUES ('jane_smith', 'inactive')" )
1633- db_connection .commit ()
1634-
1635- # Example 1: Iterate over results directly (pyodbc style)
1636- user_names = []
1637- for row in cursor .execute ("SELECT user_id, user_name FROM #users WHERE status = ?" , "active" ):
1638- user_names .append (f"{ row .user_id } : { row .user_name } " )
1639- assert len (user_names ) == 1 , "Should find 1 active user"
1640- assert "john_doe" in user_names [0 ], "Should contain john_doe"
1641-
1642- # Example 2: Single row fetch chaining
1643- user = cursor .execute ("SELECT user_name FROM #users WHERE user_id = ?" , 1 ).fetchone ()
1644- assert user [0 ] == "john_doe" , "Should return john_doe"
1645-
1646- # Example 3: All rows fetch chaining
1647- all_users = cursor .execute ("SELECT user_name FROM #users ORDER BY user_id" ).fetchall ()
1648- assert len (all_users ) == 2 , "Should return 2 users"
1649- assert all_users [0 ] == ["john_doe" ], "First user should be john_doe"
1650- assert all_users [1 ] == ["jane_smith" ], "Second user should be jane_smith"
1651-
1652- # Example 4: Update with rowcount chaining
1653- from datetime import datetime
1654- now = datetime .now ()
1655- updated_count = cursor .execute (
1656- "UPDATE #users SET last_logon = ? WHERE user_name = ?" ,
1657- now , "john_doe"
1658- ).rowcount
1659- assert updated_count == 1 , "Should update 1 user"
1660-
1661- # Example 5: Delete with rowcount chaining
1662- deleted_count = cursor .execute ("DELETE FROM #users WHERE status = ?" , "inactive" ).rowcount
1663- assert deleted_count == 1 , "Should delete 1 inactive user"
1664-
1665- # Verify final state
1666- remaining_users = cursor .execute ("SELECT COUNT(*) FROM #users" ).fetchone ()[0 ]
1667- assert remaining_users == 1 , "Should have 1 user remaining"
1668-
1669- finally :
1670- try :
1671- cursor .execute ("DROP TABLE #users" )
1672- db_connection .commit ()
1673- except :
1674- pass
1675-
16761580def test_close (db_connection ):
16771581 """Test closing the cursor"""
16781582 try :
0 commit comments