From f647b8c0c9e15df61ae21aa76cb89560d7bc0719 Mon Sep 17 00:00:00 2001 From: Chetan-Kansal Date: Mon, 22 Dec 2025 02:01:34 +0530 Subject: [PATCH 1/4] FIX: clarify decimal parsing in bulk fetch and add regression test --- mssql_python/pybind/ddbc_bindings.cpp | 4 +- tests/test_decimal_separator.py | 70 +++++++++++++++++++++++++++ 2 files changed, 72 insertions(+), 2 deletions(-) create mode 100644 tests/test_decimal_separator.py diff --git a/mssql_python/pybind/ddbc_bindings.cpp b/mssql_python/pybind/ddbc_bindings.cpp index 63696f91..d9cfa4f3 100644 --- a/mssql_python/pybind/ddbc_bindings.cpp +++ b/mssql_python/pybind/ddbc_bindings.cpp @@ -2899,7 +2899,7 @@ SQLRETURN SQLGetData_wrap(SqlHandlePtr StatementHandle, SQLUSMALLINT colCount, p SQLHSTMT hStmt = StatementHandle->get(); // Cache decimal separator to avoid repeated system calls - std::string decimalSeparator = GetDecimalSeparator(); + for (SQLSMALLINT i = 1; i <= colCount; ++i) { SQLWCHAR columnName[256]; @@ -3622,7 +3622,7 @@ SQLRETURN FetchBatchData(SQLHSTMT hStmt, ColumnBuffers& buffers, py::list& colum columnInfos[col].processedColumnSize + 1; // +1 for null terminator } - std::string decimalSeparator = GetDecimalSeparator(); // Cache decimal separator + // Performance: Build function pointer dispatch table (once per batch) // This eliminates the switch statement from the hot loop - 10,000 rows × 10 diff --git a/tests/test_decimal_separator.py b/tests/test_decimal_separator.py new file mode 100644 index 00000000..54578b6f --- /dev/null +++ b/tests/test_decimal_separator.py @@ -0,0 +1,70 @@ +import pytest +import mssql_python +import os +from decimal import Decimal + +# Helper to get connection +def get_connection(): + server = os.getenv("TEST_SERVER", "localhost") + database = os.getenv("TEST_DATABASE", "master") + username = os.getenv("TEST_USERNAME", "sa") + password = os.getenv("TEST_PASSWORD", "yourStrong(!)Password") + driver = os.getenv("TEST_DRIVER", "ODBC Driver 17 for SQL Server") + + conn_str = f"DRIVER={{{driver}}};SERVER={server};DATABASE={database};UID={username};PWD={password};TrustServerCertificate=yes" + return mssql_python.connect(conn_str) + +@pytest.mark.skipif(not os.getenv("TEST_SERVER"), reason="TEST_SERVER not set") +def test_decimal_separator_bug(): + """ + Test that fetchall() dealing with DECIMALS works correctly even when + setDecimalSeparator is set to something other than '.' + """ + conn = get_connection() + cursor = conn.cursor() + + try: + # Create a temp table + cursor.execute("CREATE TABLE #TestDecimal (Val DECIMAL(10, 2))") + cursor.execute("INSERT INTO #TestDecimal VALUES (1234.56)") + cursor.execute("INSERT INTO #TestDecimal VALUES (78.90)") + conn.commit() + + # Set custom separator + mssql_python.setDecimalSeparator(",") + + # Test fetchall + cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") + rows = cursor.fetchall() + + # Verify fetchall results + assert len(rows) == 2, f"Expected 2 rows, got {len(rows)}" + assert isinstance(rows[0][0], Decimal), f"Expected Decimal, got {type(rows[0][0])}" + assert rows[0][0] == Decimal("78.90"), f"Expected 78.90, got {rows[0][0]}" + assert rows[1][0] == Decimal("1234.56"), f"Expected 1234.56, got {rows[1][0]}" + + # Verify fetchmany + cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") + batch = cursor.fetchmany(2) + assert len(batch) == 2 + assert batch[1][0] == Decimal("1234.56") + + # Verify fetchone behavior is consistent + cursor.execute("SELECT CAST(99.99 AS DECIMAL(10,2))") + val = cursor.fetchone()[0] + assert isinstance(val, Decimal) + assert val == Decimal("99.99") + + finally: + # Reset separator to default just in case + mssql_python.setDecimalSeparator(".") + cursor.execute("DROP TABLE #TestDecimal") + conn.close() + +if __name__ == "__main__": + # Allow running as a script too + try: + test_decimal_separator_bug() + print("Test PASSED") + except Exception as e: + print(f"Test FAILED: {e}") From 1aef4ec698527163958c901a8b77fd9753ded409 Mon Sep 17 00:00:00 2001 From: Chetan-Kansal Date: Fri, 26 Dec 2025 23:31:03 +0530 Subject: [PATCH 2/4] Fix CI: Refactor test_decimal_separator.py for linting, coverage, and security --- tests/test_decimal_separator.py | 51 +++++++++++++-------------------- 1 file changed, 20 insertions(+), 31 deletions(-) diff --git a/tests/test_decimal_separator.py b/tests/test_decimal_separator.py index 54578b6f..1839d9b7 100644 --- a/tests/test_decimal_separator.py +++ b/tests/test_decimal_separator.py @@ -1,29 +1,21 @@ -import pytest -import mssql_python import os from decimal import Decimal -# Helper to get connection -def get_connection(): - server = os.getenv("TEST_SERVER", "localhost") - database = os.getenv("TEST_DATABASE", "master") - username = os.getenv("TEST_USERNAME", "sa") - password = os.getenv("TEST_PASSWORD", "yourStrong(!)Password") - driver = os.getenv("TEST_DRIVER", "ODBC Driver 17 for SQL Server") - - conn_str = f"DRIVER={{{driver}}};SERVER={server};DATABASE={database};UID={username};PWD={password};TrustServerCertificate=yes" - return mssql_python.connect(conn_str) +import pytest +import mssql_python + -@pytest.mark.skipif(not os.getenv("TEST_SERVER"), reason="TEST_SERVER not set") -def test_decimal_separator_bug(): +@pytest.mark.skipif(not os.getenv("DB_CONNECTION_STRING"), reason="Requires DB_CONNECTION_STRING") +def test_decimal_separator_bug(db_connection): """ - Test that fetchall() dealing with DECIMALS works correctly even when + Test that fetchall() dealing with DECIMALS works correctly even when setDecimalSeparator is set to something other than '.' """ - conn = get_connection() - cursor = conn.cursor() - + conn = db_connection + cursor = None + try: + cursor = conn.cursor() # Create a temp table cursor.execute("CREATE TABLE #TestDecimal (Val DECIMAL(10, 2))") cursor.execute("INSERT INTO #TestDecimal VALUES (1234.56)") @@ -32,17 +24,17 @@ def test_decimal_separator_bug(): # Set custom separator mssql_python.setDecimalSeparator(",") - + # Test fetchall cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") rows = cursor.fetchall() - + # Verify fetchall results assert len(rows) == 2, f"Expected 2 rows, got {len(rows)}" assert isinstance(rows[0][0], Decimal), f"Expected Decimal, got {type(rows[0][0])}" assert rows[0][0] == Decimal("78.90"), f"Expected 78.90, got {rows[0][0]}" assert rows[1][0] == Decimal("1234.56"), f"Expected 1234.56, got {rows[1][0]}" - + # Verify fetchmany cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") batch = cursor.fetchmany(2) @@ -58,13 +50,10 @@ def test_decimal_separator_bug(): finally: # Reset separator to default just in case mssql_python.setDecimalSeparator(".") - cursor.execute("DROP TABLE #TestDecimal") - conn.close() - -if __name__ == "__main__": - # Allow running as a script too - try: - test_decimal_separator_bug() - print("Test PASSED") - except Exception as e: - print(f"Test FAILED: {e}") + if cursor: + try: + cursor.execute("DROP TABLE #TestDecimal") + conn.commit() + except Exception: + pass + cursor.close() From a768e44ce4c070c168ede323d9465eeea559eb94 Mon Sep 17 00:00:00 2001 From: Chetan-Kansal Date: Mon, 19 Jan 2026 16:13:53 +0530 Subject: [PATCH 3/4] Move decimal separator regression test into cursor test suite --- tests/test_004_cursor.py | 49 +++++++++++++++++++++++++++ tests/test_decimal_separator.py | 59 --------------------------------- 2 files changed, 49 insertions(+), 59 deletions(-) delete mode 100644 tests/test_decimal_separator.py diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index f63972f1..fc57bfb2 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -9,6 +9,7 @@ """ import pytest +import os from datetime import datetime, date, time, timedelta, timezone import time as time_module import decimal @@ -9097,6 +9098,54 @@ def test_decimal_separator_calculations(cursor, db_connection): db_connection.commit() +@pytest.mark.skipif(not os.getenv("DB_CONNECTION_STRING"), reason="Requires DB_CONNECTION_STRING") +def test_decimal_separator_fetch_regression(cursor, db_connection): + """ + Test that fetchall() dealing with DECIMALS works correctly even when + setDecimalSeparator is set to something other than '.' + """ + try: + # Create a temp table + cursor.execute("CREATE TABLE #TestDecimal (Val DECIMAL(10, 2))") + cursor.execute("INSERT INTO #TestDecimal VALUES (1234.56)") + cursor.execute("INSERT INTO #TestDecimal VALUES (78.90)") + db_connection.commit() + + # Set custom separator + mssql_python.setDecimalSeparator(",") + + # Test fetchall + cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") + rows = cursor.fetchall() + + # Verify fetchall results + assert len(rows) == 2, f"Expected 2 rows, got {len(rows)}" + assert isinstance(rows[0][0], decimal.Decimal), f"Expected Decimal, got {type(rows[0][0])}" + assert rows[0][0] == decimal.Decimal("78.90"), f"Expected 78.90, got {rows[0][0]}" + assert rows[1][0] == decimal.Decimal("1234.56"), f"Expected 1234.56, got {rows[1][0]}" + + # Verify fetchmany + cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") + batch = cursor.fetchmany(2) + assert len(batch) == 2 + assert batch[1][0] == decimal.Decimal("1234.56") + + # Verify fetchone behavior is consistent + cursor.execute("SELECT CAST(99.99 AS DECIMAL(10,2))") + val = cursor.fetchone()[0] + assert isinstance(val, decimal.Decimal) + assert val == decimal.Decimal("99.99") + + finally: + # Reset separator to default just in case + mssql_python.setDecimalSeparator(".") + try: + cursor.execute("DROP TABLE IF EXISTS #TestDecimal") + db_connection.commit() + except Exception: + pass + + def test_datetimeoffset_read_write(cursor, db_connection): """Test reading and writing timezone-aware DATETIMEOFFSET values.""" try: diff --git a/tests/test_decimal_separator.py b/tests/test_decimal_separator.py deleted file mode 100644 index 1839d9b7..00000000 --- a/tests/test_decimal_separator.py +++ /dev/null @@ -1,59 +0,0 @@ -import os -from decimal import Decimal - -import pytest -import mssql_python - - -@pytest.mark.skipif(not os.getenv("DB_CONNECTION_STRING"), reason="Requires DB_CONNECTION_STRING") -def test_decimal_separator_bug(db_connection): - """ - Test that fetchall() dealing with DECIMALS works correctly even when - setDecimalSeparator is set to something other than '.' - """ - conn = db_connection - cursor = None - - try: - cursor = conn.cursor() - # Create a temp table - cursor.execute("CREATE TABLE #TestDecimal (Val DECIMAL(10, 2))") - cursor.execute("INSERT INTO #TestDecimal VALUES (1234.56)") - cursor.execute("INSERT INTO #TestDecimal VALUES (78.90)") - conn.commit() - - # Set custom separator - mssql_python.setDecimalSeparator(",") - - # Test fetchall - cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") - rows = cursor.fetchall() - - # Verify fetchall results - assert len(rows) == 2, f"Expected 2 rows, got {len(rows)}" - assert isinstance(rows[0][0], Decimal), f"Expected Decimal, got {type(rows[0][0])}" - assert rows[0][0] == Decimal("78.90"), f"Expected 78.90, got {rows[0][0]}" - assert rows[1][0] == Decimal("1234.56"), f"Expected 1234.56, got {rows[1][0]}" - - # Verify fetchmany - cursor.execute("SELECT Val FROM #TestDecimal ORDER BY Val") - batch = cursor.fetchmany(2) - assert len(batch) == 2 - assert batch[1][0] == Decimal("1234.56") - - # Verify fetchone behavior is consistent - cursor.execute("SELECT CAST(99.99 AS DECIMAL(10,2))") - val = cursor.fetchone()[0] - assert isinstance(val, Decimal) - assert val == Decimal("99.99") - - finally: - # Reset separator to default just in case - mssql_python.setDecimalSeparator(".") - if cursor: - try: - cursor.execute("DROP TABLE #TestDecimal") - conn.commit() - except Exception: - pass - cursor.close() From 5220c975eb773d581c780120cf285e0aa5a81d17 Mon Sep 17 00:00:00 2001 From: Chetan-Kansal Date: Mon, 19 Jan 2026 16:28:51 +0530 Subject: [PATCH 4/4] STYLE: apply Black formatting to Python tests --- mssql_python/logging.py | 1 - tests/test_001_globals.py | 12 +- tests/test_003_connection.py | 12 +- tests/test_004_cursor.py | 919 +++++++--------------- tests/test_008_logging_integration.py | 1 - tests/test_011_performance_stress.py | 36 +- tests/test_013_SqlHandle_free_shutdown.py | 96 +-- tests/test_013_encoding_decoding.py | 108 +-- tests/test_cache_invalidation.py | 109 +-- 9 files changed, 430 insertions(+), 864 deletions(-) diff --git a/mssql_python/logging.py b/mssql_python/logging.py index 77db4a45..2cb9361f 100644 --- a/mssql_python/logging.py +++ b/mssql_python/logging.py @@ -17,7 +17,6 @@ import atexit from typing import Optional - # Single DEBUG level - all or nothing philosophy # If you need logging, you need to see everything DEBUG = logging.DEBUG # 10 diff --git a/tests/test_001_globals.py b/tests/test_001_globals.py index db8b0d2e..7c004a13 100644 --- a/tests/test_001_globals.py +++ b/tests/test_001_globals.py @@ -388,8 +388,7 @@ def test_decimal_separator_with_db_operations(db_connection): try: # Create a test table with decimal values cursor = db_connection.cursor() - cursor.execute( - """ + cursor.execute(""" DROP TABLE IF EXISTS #decimal_separator_test; CREATE TABLE #decimal_separator_test ( id INT, @@ -400,8 +399,7 @@ def test_decimal_separator_with_db_operations(db_connection): (2, 678.90), (3, 0.01), (4, 999.99); - """ - ) + """) cursor.close() # Test 1: Fetch with default separator @@ -469,8 +467,7 @@ def test_decimal_separator_batch_operations(db_connection): try: # Create test data cursor = db_connection.cursor() - cursor.execute( - """ + cursor.execute(""" DROP TABLE IF EXISTS #decimal_batch_test; CREATE TABLE #decimal_batch_test ( id INT, @@ -481,8 +478,7 @@ def test_decimal_separator_batch_operations(db_connection): (1, 123.456, 12345.67890), (2, 0.001, 0.00001), (3, 999.999, 9999.99999); - """ - ) + """) cursor.close() # Test 1: Fetch results with default separator diff --git a/tests/test_003_connection.py b/tests/test_003_connection.py index e2f0cce0..c6141ea7 100644 --- a/tests/test_003_connection.py +++ b/tests/test_003_connection.py @@ -992,16 +992,14 @@ def test_execute_with_large_parameters(db_connection, conn_str): pytest.skip("Skipping for Azure SQL - large parameter tests may cause timeouts") # Test with a temporary table for large data - cursor = db_connection.execute( - """ + cursor = db_connection.execute(""" DROP TABLE IF EXISTS #large_params_test; CREATE TABLE #large_params_test ( id INT, large_text NVARCHAR(MAX), large_binary VARBINARY(MAX) ) - """ - ) + """) cursor.close() try: @@ -2126,12 +2124,10 @@ def test_timeout_long_query(db_connection): while retry_count < max_retries: start_time = time.perf_counter() try: - cursor.execute( - """ + cursor.execute(""" SELECT COUNT(*) FROM sys.objects a, sys.objects b, sys.objects c WHERE a.object_id = b.object_id * c.object_id - """ - ) + """) cursor.fetchall() elapsed_time = time.perf_counter() - start_time break # Success, exit retry loop diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index fc57bfb2..57549629 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -19,7 +19,6 @@ import re from conftest import is_azure_sql_connection - # Setup test table TEST_TABLE = """ CREATE TABLE #pytest_all_data_types ( @@ -183,15 +182,13 @@ def test_mixed_empty_and_null_values(cursor, db_connection): try: # Create test table drop_table_if_exists(cursor, "#pytest_empty_vs_null") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_empty_vs_null ( id INT, text_col NVARCHAR(100), binary_col VARBINARY(100) ) - """ - ) + """) db_connection.commit() # Insert mix of empty and NULL values @@ -889,15 +886,13 @@ def test_rowcount(cursor, db_connection): cursor.execute("INSERT INTO #pytest_test_rowcount (name) VALUES ('JohnDoe3');") assert cursor.rowcount == 1, "Rowcount should be 1 after third insert" - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_test_rowcount (name) VALUES ('JohnDoe4'), ('JohnDoe5'), ('JohnDoe6'); - """ - ) + """) assert cursor.rowcount == 3, "Rowcount should be 3 after inserting multiple rows" cursor.execute("SELECT * FROM #pytest_test_rowcount;") @@ -993,14 +988,12 @@ def test_fetchmany_size_zero_lob(cursor, db_connection): """Test fetchmany with size=0 for LOB columns""" try: cursor.execute("DROP TABLE IF EXISTS #test_fetchmany_lob") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_fetchmany_lob ( id INT PRIMARY KEY, lob_data NVARCHAR(MAX) ) - """ - ) + """) # Insert test data test_data = [(1, "First LOB data"), (2, "Second LOB data"), (3, "Third LOB data")] @@ -1025,14 +1018,12 @@ def test_fetchmany_more_than_exist_lob(cursor, db_connection): """Test fetchmany requesting more rows than exist with LOB columns""" try: cursor.execute("DROP TABLE IF EXISTS #test_fetchmany_lob_more") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_fetchmany_lob_more ( id INT PRIMARY KEY, lob_data NVARCHAR(MAX) ) - """ - ) + """) # Insert only 3 rows test_data = [(1, "First LOB data"), (2, "Second LOB data"), (3, "Third LOB data")] @@ -1066,14 +1057,12 @@ def test_fetchmany_empty_result_lob(cursor, db_connection): """Test fetchmany on empty result set with LOB columns""" try: cursor.execute("DROP TABLE IF EXISTS #test_fetchmany_lob_empty") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_fetchmany_lob_empty ( id INT PRIMARY KEY, lob_data NVARCHAR(MAX) ) - """ - ) + """) db_connection.commit() # Query empty table @@ -1096,14 +1085,12 @@ def test_fetchmany_very_large_lob(cursor, db_connection): """Test fetchmany with very large LOB column data""" try: cursor.execute("DROP TABLE IF EXISTS #test_fetchmany_large_lob") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_fetchmany_large_lob ( id INT PRIMARY KEY, large_lob NVARCHAR(MAX) ) - """ - ) + """) # Create very large data (10000 characters) large_data = "x" * 10000 @@ -1153,14 +1140,12 @@ def test_fetchmany_mixed_lob_sizes(cursor, db_connection): """Test fetchmany with mixed LOB sizes including empty and NULL""" try: cursor.execute("DROP TABLE IF EXISTS #test_fetchmany_mixed_lob") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_fetchmany_mixed_lob ( id INT PRIMARY KEY, mixed_lob NVARCHAR(MAX) ) - """ - ) + """) # Mix of sizes: empty, NULL, small, medium, large test_data = [ @@ -1288,14 +1273,12 @@ def test_executemany_empty_strings(cursor, db_connection): """Test executemany with empty strings - regression test for Unix UTF-16 conversion issue""" try: # Create test table for empty string testing - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_empty_batch ( id INT, data NVARCHAR(50) ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_empty_batch") @@ -1336,8 +1319,7 @@ def test_executemany_empty_strings_various_types(cursor, db_connection): """Test executemany with empty strings in different column types""" try: # Create test table with different string types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_string_types ( id INT, varchar_col VARCHAR(50), @@ -1345,8 +1327,7 @@ def test_executemany_empty_strings_various_types(cursor, db_connection): text_col TEXT, ntext_col NTEXT ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_string_types") @@ -1387,14 +1368,12 @@ def test_executemany_unicode_and_empty_strings(cursor, db_connection): """Test executemany with mix of Unicode characters and empty strings""" try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_unicode_test ( id INT, data NVARCHAR(100) ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_unicode_test") @@ -1439,14 +1418,12 @@ def test_executemany_large_batch_with_empty_strings(cursor, db_connection): """Test executemany with large batch containing empty strings""" try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_large_batch ( id INT, data NVARCHAR(50) ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_large_batch") @@ -1499,14 +1476,12 @@ def test_executemany_compare_with_execute(cursor, db_connection): """Test that executemany produces same results as individual execute calls""" try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_compare_test ( id INT, data NVARCHAR(50) ) - """ - ) + """) # Test data with empty strings test_data = [ @@ -1559,15 +1534,13 @@ def test_executemany_edge_cases_empty_strings(cursor, db_connection): """Test executemany edge cases with empty strings and special characters""" try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_edge_cases ( id INT, varchar_data VARCHAR(100), nvarchar_data NVARCHAR(100) ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_edge_cases") @@ -1621,14 +1594,12 @@ def test_executemany_null_vs_empty_string(cursor, db_connection): """Test that executemany correctly distinguishes between NULL and empty string""" try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_null_vs_empty ( id INT, data NVARCHAR(50) ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_null_vs_empty") @@ -1693,14 +1664,12 @@ def test_executemany_binary_data_edge_cases(cursor, db_connection): """Test executemany with binary data and empty byte arrays""" try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_binary_test ( id INT, binary_data VARBINARY(100) ) - """ - ) + """) # Clear any existing data cursor.execute("DELETE FROM #pytest_binary_test") @@ -1862,8 +1831,7 @@ def test_executemany_mixed_null_and_typed_values(cursor, db_connection): """Test executemany with randomly mixed NULL and non-NULL values across multiple columns and rows (50 rows, 10 columns).""" try: # Create table with 10 columns of various types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_empty_params ( col1 INT, col2 VARCHAR(50), @@ -1876,8 +1844,7 @@ def test_executemany_mixed_null_and_typed_values(cursor, db_connection): col9 DATE, col10 REAL ) - """ - ) + """) # Generate 50 rows with randomly mixed NULL and non-NULL values across 10 columns data = [] @@ -1941,8 +1908,7 @@ def test_executemany_multi_column_null_arrays(cursor, db_connection): """Test executemany with multi-column NULL arrays (50 records, 8 columns).""" try: # Create table with 8 columns of various types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_null_arrays ( col1 INT, col2 VARCHAR(100), @@ -1953,8 +1919,7 @@ def test_executemany_multi_column_null_arrays(cursor, db_connection): col7 BIGINT, col8 DATE ) - """ - ) + """) # Generate 50 rows with all NULL values across 8 columns data = [(None, None, None, None, None, None, None, None) for _ in range(50)] @@ -1974,14 +1939,12 @@ def test_executemany_multi_column_null_arrays(cursor, db_connection): assert null_count == 50, f"Expected 50 NULLs in col{col_num}, got {null_count}" # Verify no non-NULL values exist - cursor.execute( - """ + cursor.execute(""" SELECT COUNT(*) FROM #pytest_null_arrays WHERE col1 IS NOT NULL OR col2 IS NOT NULL OR col3 IS NOT NULL OR col4 IS NOT NULL OR col5 IS NOT NULL OR col6 IS NOT NULL OR col7 IS NOT NULL OR col8 IS NOT NULL - """ - ) + """) non_null_count = cursor.fetchone()[0] assert non_null_count == 0, f"Expected 0 non-NULL values, got {non_null_count}" @@ -2020,8 +1983,7 @@ def test_executemany_concurrent_null_parameters(db_connection): # Create table with db_connection.cursor() as cursor: - cursor.execute( - f""" + cursor.execute(f""" IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name} @@ -2033,8 +1995,7 @@ def test_executemany_concurrent_null_parameters(db_connection): col3 FLOAT, col4 DATETIME ) - """ - ) + """) db_connection.commit() # Execute multiple sequential insert operations @@ -2289,14 +2250,12 @@ def test_insert_data_for_join(cursor, db_connection): def test_join_operations(cursor): """Test join operations""" try: - cursor.execute( - """ + cursor.execute(""" SELECT e.name, d.department_name, p.project_name FROM #pytest_employees e JOIN #pytest_departments d ON e.department_id = d.department_id JOIN #pytest_projects p ON e.employee_id = p.employee_id - """ - ) + """) rows = cursor.fetchall() assert len(rows) == 3, "Join operation returned incorrect number of rows" assert rows[0] == [ @@ -2386,12 +2345,10 @@ def test_execute_stored_procedure_with_parameters(cursor): def test_execute_stored_procedure_without_parameters(cursor): """Test executing stored procedure without parameters""" try: - cursor.execute( - """ + cursor.execute(""" DECLARE @EmployeeID INT = 2 EXEC dbo.GetEmployeeProjects @EmployeeID - """ - ) + """) rows = cursor.fetchall() assert ( len(rows) == 1 @@ -2611,25 +2568,21 @@ def test_row_attribute_access(cursor, db_connection): """Test accessing row values by column name as attributes""" try: # Create test table with multiple columns - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_row_attr_test ( id INT PRIMARY KEY, name VARCHAR(50), email VARCHAR(100), age INT ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_row_attr_test (id, name, email, age) VALUES (1, 'John Doe', 'john@example.com', 30) - """ - ) + """) db_connection.commit() # Test attribute access @@ -2725,15 +2678,13 @@ def test_row_comparison_with_list(cursor, db_connection): def test_row_string_representation(cursor, db_connection): """Test Row string and repr representations""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_row_test ( id INT PRIMARY KEY, text_col NVARCHAR(50), null_col INT ) - """ - ) + """) db_connection.commit() cursor.execute( @@ -2766,15 +2717,13 @@ def test_row_string_representation(cursor, db_connection): def test_row_column_mapping(cursor, db_connection): """Test Row column name mapping""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_row_test ( FirstColumn INT PRIMARY KEY, Second_Column NVARCHAR(50), [Complex Name!] INT ) - """ - ) + """) db_connection.commit() cursor.execute( @@ -3257,12 +3206,10 @@ def test_execute_rowcount_chaining(cursor, db_connection): assert count == 1, "INSERT should affect 1 row" # Test multiple INSERT rowcount chaining - count = cursor.execute( - """ + count = cursor.execute(""" INSERT INTO #test_chaining (id, value) VALUES (2, 'test2'), (3, 'test3'), (4, 'test4') - """ - ).rowcount + """).rowcount assert count == 3, "Multiple INSERT should affect 3 rows" # Test UPDATE rowcount chaining @@ -3497,8 +3444,7 @@ def test_cursor_next_with_different_data_types(cursor, db_connection): """Test next() functionality with various data types""" try: # Create test table with various data types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_next_types ( id INT, name NVARCHAR(50), @@ -3507,8 +3453,7 @@ def test_cursor_next_with_different_data_types(cursor, db_connection): created_date DATE, created_time DATETIME ) - """ - ) + """) db_connection.commit() # Insert test data with different types @@ -3700,16 +3645,14 @@ def test_execute_chaining_compatibility_examples(cursor, db_connection): """Test real-world chaining examples""" try: # Create users table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #users ( user_id INT IDENTITY(1,1) PRIMARY KEY, user_name NVARCHAR(50), last_logon DATETIME, status NVARCHAR(20) ) - """ - ) + """) db_connection.commit() # Insert test users @@ -4408,8 +4351,7 @@ def test_fetchval_different_data_types(cursor, db_connection): try: # Create test table with different data types drop_table_if_exists(cursor, "#pytest_fetchval_types") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_fetchval_types ( int_col INTEGER, float_col FLOAT, @@ -4421,17 +4363,14 @@ def test_fetchval_different_data_types(cursor, db_connection): date_col DATE, time_col TIME ) - """ - ) + """) # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_fetchval_types VALUES (123, 45.67, 89.12, 'ASCII text', N'Unicode text', 1, '2024-05-20 12:34:56', '2024-05-20', '12:34:56') - """ - ) + """) db_connection.commit() # Test different data types @@ -5729,25 +5668,21 @@ def test_cursor_rollback_data_consistency(cursor, db_connection): drop_table_if_exists(cursor, "#pytest_rollback_orders") drop_table_if_exists(cursor, "#pytest_rollback_customers") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_rollback_customers ( id INTEGER PRIMARY KEY, name VARCHAR(50) ) - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_rollback_orders ( id INTEGER PRIMARY KEY, customer_id INTEGER, amount DECIMAL(10,2), FOREIGN KEY (customer_id) REFERENCES #pytest_rollback_customers(id) ) - """ - ) + """) cursor.commit() # Insert initial data @@ -6229,32 +6164,26 @@ def test_tables_setup(cursor, db_connection): cursor.execute("DROP VIEW IF EXISTS pytest_tables_schema.test_view") # Create regular table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_tables_schema.regular_table ( id INT PRIMARY KEY, name VARCHAR(100) ) - """ - ) + """) # Create another table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_tables_schema.another_table ( id INT PRIMARY KEY, description VARCHAR(200) ) - """ - ) + """) # Create a view - cursor.execute( - """ + cursor.execute(""" CREATE VIEW pytest_tables_schema.test_view AS SELECT id, name FROM pytest_tables_schema.regular_table - """ - ) + """) db_connection.commit() except Exception as e: @@ -6606,14 +6535,12 @@ def test_emoji_round_trip(cursor, db_connection): "1🚀' OR '1'='1", ] - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_emoji_test ( id INT IDENTITY PRIMARY KEY, content NVARCHAR(MAX) ); - """ - ) + """) db_connection.commit() for text in test_inputs: @@ -6765,16 +6692,14 @@ def test_empty_values_fetchmany(cursor, db_connection): try: # Create comprehensive test table drop_table_if_exists(cursor, "#pytest_fetchmany_empty") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_fetchmany_empty ( id INT, varchar_col VARCHAR(50), nvarchar_col NVARCHAR(50), binary_col VARBINARY(50) ) - """ - ) + """) db_connection.commit() # Insert multiple rows with empty values @@ -6899,8 +6824,7 @@ def test_batch_fetch_empty_values_no_assertion_failure(cursor, db_connection): try: # Create comprehensive test table drop_table_if_exists(cursor, "#pytest_batch_empty_assertions") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_batch_empty_assertions ( id INT, empty_varchar VARCHAR(100), @@ -6910,29 +6834,24 @@ def test_batch_fetch_empty_values_no_assertion_failure(cursor, db_connection): null_nvarchar NVARCHAR(100), null_binary VARBINARY(100) ) - """ - ) + """) db_connection.commit() # Insert rows with mix of empty and NULL values - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_batch_empty_assertions VALUES (1, '', '', 0x, NULL, NULL, NULL), (2, '', '', 0x, NULL, NULL, NULL), (3, '', '', 0x, NULL, NULL, NULL) - """ - ) + """) db_connection.commit() # Test fetchall - should not trigger any assertions about dataLen - cursor.execute( - """ + cursor.execute(""" SELECT empty_varchar, empty_nvarchar, empty_binary, null_varchar, null_nvarchar, null_binary FROM #pytest_batch_empty_assertions ORDER BY id - """ - ) + """) rows = cursor.fetchall() assert len(rows) == 3, "Should return 3 rows" @@ -6949,12 +6868,10 @@ def test_batch_fetch_empty_values_no_assertion_failure(cursor, db_connection): assert row[5] is None, f"Row {i+1} null_binary should be None" # Test fetchmany - should also not trigger assertions - cursor.execute( - """ + cursor.execute(""" SELECT empty_nvarchar, empty_binary FROM #pytest_batch_empty_assertions ORDER BY id - """ - ) + """) # Fetch in batches first_batch = cursor.fetchmany(2) @@ -6994,15 +6911,13 @@ def test_executemany_utf16_length_validation(cursor, db_connection): try: # Create test table with small column size to trigger validation drop_table_if_exists(cursor, "#pytest_utf16_validation") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_utf16_validation ( id INT, short_text NVARCHAR(5), -- Small column to test length validation medium_text NVARCHAR(10) -- Medium column for edge cases ) - """ - ) + """) db_connection.commit() # Test 1: Valid strings that should work on all platforms @@ -7148,14 +7063,12 @@ def test_binary_data_over_8000_bytes(cursor, db_connection): try: # Create test table with VARBINARY(MAX) to handle large data drop_table_if_exists(cursor, "#pytest_small_binary") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_small_binary ( id INT, large_binary VARBINARY(MAX) ) - """ - ) + """) # Test data that fits within both parameter and fetch limits (< 4096 bytes) medium_data = b"B" * 3000 # 3,000 bytes - under both limits @@ -7189,14 +7102,12 @@ def test_varbinarymax_insert_fetch(cursor, db_connection): try: # Create test table drop_table_if_exists(cursor, "#pytest_varbinarymax") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_varbinarymax ( id INT, binary_data VARBINARY(MAX) ) - """ - ) + """) # Prepare test data - use moderate sizes to guarantee LOB fetch path (line 867-868) efficiently test_data = [ @@ -7263,14 +7174,12 @@ def test_all_empty_binaries(cursor, db_connection): try: # Create test table drop_table_if_exists(cursor, "#pytest_all_empty_binary") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_all_empty_binary ( id INT, empty_binary VARBINARY(100) ) - """ - ) + """) # Insert multiple rows with only empty binary data test_data = [ @@ -7309,14 +7218,12 @@ def test_mixed_bytes_and_bytearray_types(cursor, db_connection): try: # Create test table drop_table_if_exists(cursor, "#pytest_mixed_binary_types") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_mixed_binary_types ( id INT, binary_data VARBINARY(100) ) - """ - ) + """) # Test data mixing bytes and bytearray for the same column test_data = [ @@ -7371,14 +7278,12 @@ def test_binary_mostly_small_one_large(cursor, db_connection): try: # Create test table drop_table_if_exists(cursor, "#pytest_mixed_size_binary") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_mixed_size_binary ( id INT, binary_data VARBINARY(MAX) ) - """ - ) + """) # Create large binary value within both parameter and fetch limits (< 4096 bytes) large_binary = b"X" * 3500 # 3,500 bytes - under both limits @@ -7438,14 +7343,12 @@ def test_varbinarymax_insert_fetch_null(cursor, db_connection): """Test insertion and retrieval of NULL value in VARBINARY(MAX) column.""" try: drop_table_if_exists(cursor, "#pytest_varbinarymax_null") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_varbinarymax_null ( id INT, binary_data VARBINARY(MAX) ) - """ - ) + """) # Insert a row with NULL for binary_data cursor.execute( @@ -7475,15 +7378,13 @@ def test_sql_double_type(cursor, db_connection): """Test SQL_DOUBLE type (FLOAT(53)) to cover line 3213 in dispatcher.""" try: drop_table_if_exists(cursor, "#pytest_double_type") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_double_type ( id INT PRIMARY KEY, double_col FLOAT(53), float_col FLOAT ) - """ - ) + """) # Insert test data with various double precision values test_data = [ @@ -7531,15 +7432,13 @@ def test_null_guid_type(cursor, db_connection): """Test NULL UNIQUEIDENTIFIER (GUID) to cover lines 3376-3377.""" try: drop_table_if_exists(cursor, "#pytest_null_guid") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_null_guid ( id INT PRIMARY KEY, guid_col UNIQUEIDENTIFIER, guid_nullable UNIQUEIDENTIFIER NULL ) - """ - ) + """) # Insert test data with NULL and non-NULL GUIDs test_guid = uuid.uuid4() @@ -7591,14 +7490,12 @@ def test_only_null_and_empty_binary(cursor, db_connection): try: # Create test table drop_table_if_exists(cursor, "#pytest_null_empty_binary") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_null_empty_binary ( id INT, binary_data VARBINARY(100) ) - """ - ) + """) # Test data with only NULL and empty values test_data = [ @@ -7921,8 +7818,7 @@ def test_money_smallmoney_insert_fetch(cursor, db_connection): """Test inserting and retrieving valid MONEY and SMALLMONEY values including boundaries and typical data""" try: drop_table_if_exists(cursor, "#pytest_money_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, @@ -7930,8 +7826,7 @@ def test_money_smallmoney_insert_fetch(cursor, db_connection): d DECIMAL(19,4), n NUMERIC(10,4) ) - """ - ) + """) db_connection.commit() # Max values @@ -8021,15 +7916,13 @@ def test_money_smallmoney_insert_fetch(cursor, db_connection): def test_money_smallmoney_null_handling(cursor, db_connection): """Test that NULL values for MONEY and SMALLMONEY are stored and retrieved correctly""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, sm SMALLMONEY ) - """ - ) + """) db_connection.commit() # Row with both NULLs @@ -8079,15 +7972,13 @@ def test_money_smallmoney_null_handling(cursor, db_connection): def test_money_smallmoney_roundtrip(cursor, db_connection): """Test inserting and retrieving MONEY and SMALLMONEY using decimal.Decimal roundtrip""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, sm SMALLMONEY ) - """ - ) + """) db_connection.commit() values = (decimal.Decimal("12345.6789"), decimal.Decimal("987.6543")) @@ -8111,15 +8002,13 @@ def test_money_smallmoney_boundaries(cursor, db_connection): """Test boundary values for MONEY and SMALLMONEY types are handled correctly""" try: drop_table_if_exists(cursor, "#pytest_money_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, sm SMALLMONEY ) - """ - ) + """) db_connection.commit() # Insert max boundary @@ -8159,15 +8048,13 @@ def test_money_smallmoney_boundaries(cursor, db_connection): def test_money_smallmoney_invalid_values(cursor, db_connection): """Test that invalid or out-of-range MONEY and SMALLMONEY values raise errors""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, sm SMALLMONEY ) - """ - ) + """) db_connection.commit() # Out of range MONEY @@ -8198,15 +8085,13 @@ def test_money_smallmoney_invalid_values(cursor, db_connection): def test_money_smallmoney_roundtrip_executemany(cursor, db_connection): """Test inserting and retrieving MONEY and SMALLMONEY using executemany with decimal.Decimal""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, sm SMALLMONEY ) - """ - ) + """) db_connection.commit() test_data = [ @@ -8240,15 +8125,13 @@ def test_money_smallmoney_roundtrip_executemany(cursor, db_connection): def test_money_smallmoney_executemany_null_handling(cursor, db_connection): """Test inserting NULLs into MONEY and SMALLMONEY using executemany""" try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_money_test ( id INT IDENTITY PRIMARY KEY, m MONEY, sm SMALLMONEY ) - """ - ) + """) db_connection.commit() rows = [ @@ -8306,14 +8189,12 @@ def test_uuid_insert_and_select_none(cursor, db_connection): table_name = "#pytest_uuid_nullable" try: cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER, name NVARCHAR(50) ) - """ - ) + """) db_connection.commit() # Insert a row with None for the UUID @@ -8337,14 +8218,12 @@ def test_insert_multiple_uuids(cursor, db_connection): table_name = "#pytest_uuid_multiple" try: cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER PRIMARY KEY, description NVARCHAR(50) ) - """ - ) + """) db_connection.commit() # Prepare test data @@ -8380,14 +8259,12 @@ def test_fetchmany_uuids(cursor, db_connection): table_name = "#pytest_uuid_fetchmany" try: cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER PRIMARY KEY, description NVARCHAR(50) ) - """ - ) + """) db_connection.commit() uuids_to_insert = {f"Item {i}": uuid.uuid4() for i in range(10)} @@ -8423,14 +8300,12 @@ def test_uuid_insert_with_none(cursor, db_connection): table_name = "#pytest_uuid_none" try: cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER, name NVARCHAR(50) ) - """ - ) + """) db_connection.commit() cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Alice"]) @@ -8526,14 +8401,12 @@ def test_executemany_uuid_insert_and_select(cursor, db_connection): try: # Drop and create a temporary table for the test cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER PRIMARY KEY, description NVARCHAR(50) ) - """ - ) + """) db_connection.commit() # Generate data for insertion @@ -8583,14 +8456,12 @@ def test_executemany_uuid_roundtrip_fixed_value(cursor, db_connection): table_name = "#pytest_uuid_fixed" try: cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER, description NVARCHAR(50) ) - """ - ) + """) db_connection.commit() fixed_uuid = uuid.UUID("12345678-1234-5678-1234-567812345678") @@ -8631,8 +8502,7 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_multi_test ( id INT PRIMARY KEY, positive_value DECIMAL(10, 2), @@ -8640,16 +8510,13 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): zero_value DECIMAL(10, 2), small_value DECIMAL(10, 4) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_multi_test VALUES (1, 123.45, -67.89, 0.00, 0.0001) - """ - ) + """) db_connection.commit() # Test with default separator first @@ -8686,23 +8553,19 @@ def test_decimal_separator_calculations(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_calc_test ( id INT PRIMARY KEY, value1 DECIMAL(10, 2), value2 DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_calc_test VALUES (1, 10.25, 5.75) - """ - ) + """) db_connection.commit() # Test with default separator @@ -8741,14 +8604,12 @@ def test_decimal_separator_function(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_separator_test ( id INT PRIMARY KEY, decimal_value DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test values with default separator (.) @@ -8833,25 +8694,21 @@ def test_lowercase_attribute(cursor, db_connection): try: # Create a test table with mixed-case column names - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lowercase_test ( ID INT PRIMARY KEY, UserName VARCHAR(50), EMAIL_ADDRESS VARCHAR(100), PhoneNumber VARCHAR(20) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_lowercase_test (ID, UserName, EMAIL_ADDRESS, PhoneNumber) VALUES (1, 'JohnDoe', 'john@example.com', '555-1234') - """ - ) + """) db_connection.commit() # First test with lowercase=False (default) @@ -8906,14 +8763,12 @@ def test_decimal_separator_function(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_separator_test ( id INT PRIMARY KEY, decimal_value DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test values with default separator (.) @@ -8995,8 +8850,7 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_multi_test ( id INT PRIMARY KEY, positive_value DECIMAL(10, 2), @@ -9004,16 +8858,13 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): zero_value DECIMAL(10, 2), small_value DECIMAL(10, 4) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_multi_test VALUES (1, 123.45, -67.89, 0.00, 0.0001) - """ - ) + """) db_connection.commit() # Test with default separator first @@ -9050,23 +8901,19 @@ def test_decimal_separator_calculations(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_calc_test ( id INT PRIMARY KEY, value1 DECIMAL(10, 2), value2 DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_calc_test VALUES (1, 10.25, 5.75) - """ - ) + """) db_connection.commit() # Test with default separator @@ -9579,25 +9426,21 @@ def test_lowercase_attribute(cursor, db_connection): try: # Create a test table with mixed-case column names - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lowercase_test ( ID INT PRIMARY KEY, UserName VARCHAR(50), EMAIL_ADDRESS VARCHAR(100), PhoneNumber VARCHAR(20) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_lowercase_test (ID, UserName, EMAIL_ADDRESS, PhoneNumber) VALUES (1, 'JohnDoe', 'john@example.com', '555-1234') - """ - ) + """) db_connection.commit() # First test with lowercase=False (default) @@ -9652,14 +9495,12 @@ def test_decimal_separator_function(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_separator_test ( id INT PRIMARY KEY, decimal_value DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test values with default separator (.) @@ -9741,8 +9582,7 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_multi_test ( id INT PRIMARY KEY, positive_value DECIMAL(10, 2), @@ -9750,16 +9590,13 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): zero_value DECIMAL(10, 2), small_value DECIMAL(10, 4) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_multi_test VALUES (1, 123.45, -67.89, 0.00, 0.0001) - """ - ) + """) db_connection.commit() # Test with default separator first @@ -9796,23 +9633,19 @@ def test_decimal_separator_calculations(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_calc_test ( id INT PRIMARY KEY, value1 DECIMAL(10, 2), value2 DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_calc_test VALUES (1, 10.25, 5.75) - """ - ) + """) db_connection.commit() # Test with default separator @@ -9851,14 +9684,12 @@ def test_cursor_setinputsizes_basic(db_connection): # Create a test table cursor.execute("DROP TABLE IF EXISTS #test_inputsizes") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes ( string_col NVARCHAR(100), int_col INT ) - """ - ) + """) # Set input sizes for parameters cursor.setinputsizes([(mssql_python.SQL_WVARCHAR, 100, 0), (mssql_python.SQL_INTEGER, 0, 0)]) @@ -9884,15 +9715,13 @@ def test_cursor_setinputsizes_with_executemany_float(db_connection): # Create a test table cursor.execute("DROP TABLE IF EXISTS #test_inputsizes_float") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes_float ( id INT, name NVARCHAR(50), price REAL /* Use REAL instead of DECIMAL */ ) - """ - ) + """) # Prepare data with float values data = [(1, "Item 1", 10.99), (2, "Item 2", 20.50), (3, "Item 3", 30.75)] @@ -9929,14 +9758,12 @@ def test_cursor_setinputsizes_reset(db_connection): # Create a test table cursor.execute("DROP TABLE IF EXISTS #test_inputsizes_reset") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes_reset ( col1 NVARCHAR(100), col2 INT ) - """ - ) + """) # Set input sizes for parameters cursor.setinputsizes([(mssql_python.SQL_WVARCHAR, 100, 0), (mssql_python.SQL_INTEGER, 0, 0)]) @@ -9971,14 +9798,12 @@ def test_cursor_setinputsizes_override_inference(db_connection): # Create a test table with specific types cursor.execute("DROP TABLE IF EXISTS #test_inputsizes_override") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes_override ( small_int SMALLINT, big_text NVARCHAR(MAX) ) - """ - ) + """) # Set input sizes that override the default inference # For SMALLINT, use a valid precision value (5 is typical for SMALLINT) @@ -10034,15 +9859,13 @@ def test_setinputsizes_parameter_count_mismatch_fewer(db_connection): # Create a test table cursor.execute("DROP TABLE IF EXISTS #test_inputsizes_mismatch") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes_mismatch ( col1 INT, col2 NVARCHAR(100), col3 FLOAT ) - """ - ) + """) # Set fewer input sizes than parameters cursor.setinputsizes( @@ -10085,14 +9908,12 @@ def test_setinputsizes_parameter_count_mismatch_more(db_connection): # Create a test table cursor.execute("DROP TABLE IF EXISTS #test_inputsizes_mismatch") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes_mismatch ( col1 INT, col2 NVARCHAR(100) ) - """ - ) + """) # Set more input sizes than parameters cursor.setinputsizes( @@ -10127,8 +9948,7 @@ def test_setinputsizes_with_null_values(db_connection): # Create a test table with multiple data types cursor.execute("DROP TABLE IF EXISTS #test_inputsizes_null") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_inputsizes_null ( int_col INT, string_col NVARCHAR(100), @@ -10136,8 +9956,7 @@ def test_setinputsizes_with_null_values(db_connection): date_col DATE, binary_col VARBINARY(100) ) - """ - ) + """) # Set input sizes for all columns cursor.setinputsizes( @@ -10440,18 +10259,15 @@ def test_procedures_setup(cursor, db_connection): ) # Create test stored procedures - cursor.execute( - """ + cursor.execute(""" CREATE OR ALTER PROCEDURE pytest_proc_schema.test_proc1 AS BEGIN SELECT 1 AS result END - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" CREATE OR ALTER PROCEDURE pytest_proc_schema.test_proc2 @param1 INT, @param2 VARCHAR(50) OUTPUT @@ -10460,8 +10276,7 @@ def test_procedures_setup(cursor, db_connection): SELECT @param2 = 'Output ' + CAST(@param1 AS VARCHAR(10)) RETURN @param1 END - """ - ) + """) db_connection.commit() except Exception as e: @@ -10579,8 +10394,7 @@ def test_procedures_with_parameters(cursor, db_connection): """Test that procedures() correctly reports parameter information""" try: # Create a simpler procedure with basic parameters - cursor.execute( - """ + cursor.execute(""" CREATE OR ALTER PROCEDURE pytest_proc_schema.test_params_proc @in1 INT, @in2 VARCHAR(50) @@ -10588,8 +10402,7 @@ def test_procedures_with_parameters(cursor, db_connection): BEGIN SELECT @in1 AS value1, @in2 AS value2 END - """ - ) + """) db_connection.commit() # Get procedure info @@ -10623,28 +10436,23 @@ def test_procedures_result_set_info(cursor, db_connection): """Test that procedures() reports information about result sets""" try: # Create procedures with different result set patterns - cursor.execute( - """ + cursor.execute(""" CREATE OR ALTER PROCEDURE pytest_proc_schema.test_no_results AS BEGIN DECLARE @x INT = 1 END - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" CREATE OR ALTER PROCEDURE pytest_proc_schema.test_one_result AS BEGIN SELECT 1 AS col1, 'test' AS col2 END - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" CREATE OR ALTER PROCEDURE pytest_proc_schema.test_multiple_results AS BEGIN @@ -10652,8 +10460,7 @@ def test_procedures_result_set_info(cursor, db_connection): SELECT 'test' AS result2 SELECT GETDATE() AS result3 END - """ - ) + """) db_connection.commit() # Get procedure info for all test procedures @@ -10735,18 +10542,15 @@ def test_foreignkeys_setup(cursor, db_connection): cursor.execute("DROP TABLE IF EXISTS pytest_fk_schema.customers") # Create parent table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_fk_schema.customers ( customer_id INT PRIMARY KEY, customer_name VARCHAR(100) NOT NULL ) - """ - ) + """) # Create child table with foreign key - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_fk_schema.orders ( order_id INT PRIMARY KEY, order_date DATETIME NOT NULL, @@ -10755,23 +10559,18 @@ def test_foreignkeys_setup(cursor, db_connection): CONSTRAINT FK_Orders_Customers FOREIGN KEY (customer_id) REFERENCES pytest_fk_schema.customers (customer_id) ) - """ - ) + """) # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO pytest_fk_schema.customers (customer_id, customer_name) VALUES (1, 'Test Customer 1'), (2, 'Test Customer 2') - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" INSERT INTO pytest_fk_schema.orders (order_id, order_date, customer_id, total_amount) VALUES (101, GETDATE(), 1, 150.00), (102, GETDATE(), 2, 250.50) - """ - ) + """) db_connection.commit() except Exception as e: @@ -10999,20 +10798,17 @@ def test_foreignkeys_multiple_column_fk(cursor, db_connection): cursor.execute("DROP TABLE IF EXISTS pytest_fk_schema.product_variants") # Create parent table with composite primary key - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_fk_schema.product_variants ( product_id INT NOT NULL, variant_id INT NOT NULL, variant_name VARCHAR(100) NOT NULL, PRIMARY KEY (product_id, variant_id) ) - """ - ) + """) # Create child table with composite foreign key - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_fk_schema.order_details ( order_id INT NOT NULL, product_id INT NOT NULL, @@ -11022,8 +10818,7 @@ def test_foreignkeys_multiple_column_fk(cursor, db_connection): CONSTRAINT FK_OrderDetails_ProductVariants FOREIGN KEY (product_id, variant_id) REFERENCES pytest_fk_schema.product_variants (product_id, variant_id) ) - """ - ) + """) db_connection.commit() @@ -11088,27 +10883,23 @@ def test_primarykeys_setup(cursor, db_connection): cursor.execute("DROP TABLE IF EXISTS pytest_pk_schema.composite_pk_test") # Create table with simple primary key - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_pk_schema.single_pk_test ( id INT PRIMARY KEY, name VARCHAR(100) NOT NULL, description VARCHAR(200) NULL ) - """ - ) + """) # Create table with composite primary key - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_pk_schema.composite_pk_test ( dept_id INT NOT NULL, emp_id INT NOT NULL, hire_date DATE NOT NULL, CONSTRAINT PK_composite_test PRIMARY KEY (dept_id, emp_id) ) - """ - ) + """) db_connection.commit() except Exception as e: @@ -11419,15 +11210,13 @@ def test_rowcount(cursor, db_connection): cursor.execute("INSERT INTO #pytest_test_rowcount (name) VALUES ('JohnDoe3');") assert cursor.rowcount == 1, "Rowcount should be 1 after third insert" - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_test_rowcount (name) VALUES ('JohnDoe4'), ('JohnDoe5'), ('JohnDoe6'); - """ - ) + """) assert cursor.rowcount == 3, "Rowcount should be 3 after inserting multiple rows" cursor.execute("SELECT * FROM #pytest_test_rowcount;") @@ -11462,31 +11251,26 @@ def test_specialcolumns_setup(cursor, db_connection): cursor.execute("DROP TABLE IF EXISTS pytest_special_schema.identity_test") # Create table with primary key (for rowIdColumns) - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_special_schema.rowid_test ( id INT PRIMARY KEY, name NVARCHAR(100) NOT NULL, unique_col NVARCHAR(100) UNIQUE, non_unique_col NVARCHAR(100) ) - """ - ) + """) # Create table with rowversion column (for rowVerColumns) - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_special_schema.timestamp_test ( id INT PRIMARY KEY, name NVARCHAR(100) NOT NULL, last_updated ROWVERSION ) - """ - ) + """) # Create table with multiple unique identifiers - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_special_schema.multiple_unique_test ( id INT NOT NULL, code VARCHAR(10) NOT NULL, @@ -11494,19 +11278,16 @@ def test_specialcolumns_setup(cursor, db_connection): order_number VARCHAR(20) UNIQUE, CONSTRAINT PK_multiple_unique_test PRIMARY KEY (id, code) ) - """ - ) + """) # Create table with identity column - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_special_schema.identity_test ( id INT IDENTITY(1,1) PRIMARY KEY, name NVARCHAR(100) NOT NULL, last_modified DATETIME DEFAULT GETDATE() ) - """ - ) + """) db_connection.commit() except Exception as e: @@ -11625,14 +11406,12 @@ def test_rowid_columns_nullable(cursor, db_connection): """Test rowIdColumns with nullable parameter""" try: # First create a table with nullable unique column and non-nullable PK - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_special_schema.nullable_test ( id INT PRIMARY KEY, -- PK can't be nullable in SQL Server data NVARCHAR(100) NULL ) - """ - ) + """) db_connection.commit() # Test with nullable=True (default) @@ -11725,14 +11504,12 @@ def test_rowver_columns_nullable(cursor, db_connection): """Test rowVerColumns with nullable parameter (not expected to have effect)""" try: # First create a table with rowversion column - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_special_schema.nullable_rowver_test ( id INT PRIMARY KEY, ts ROWVERSION ) - """ - ) + """) db_connection.commit() # Test with nullable=True (default) @@ -11841,8 +11618,7 @@ def test_statistics_setup(cursor, db_connection): cursor.execute("DROP TABLE IF EXISTS pytest_stats_schema.empty_stats_test") # Create test table with various indexes - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_stats_schema.stats_test ( id INT PRIMARY KEY, name VARCHAR(100) NOT NULL, @@ -11851,32 +11627,25 @@ def test_statistics_setup(cursor, db_connection): salary DECIMAL(10, 2) NULL, hire_date DATE NOT NULL ) - """ - ) + """) # Create a non-unique index - cursor.execute( - """ + cursor.execute(""" CREATE INDEX IX_stats_test_dept_date ON pytest_stats_schema.stats_test (department, hire_date) - """ - ) + """) # Create a unique index on multiple columns - cursor.execute( - """ + cursor.execute(""" CREATE UNIQUE INDEX UX_stats_test_name_dept ON pytest_stats_schema.stats_test (name, department) - """ - ) + """) # Create an empty table for testing - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_stats_schema.empty_stats_test ( id INT PRIMARY KEY, data VARCHAR(100) NULL ) - """ - ) + """) db_connection.commit() except Exception as e: @@ -12141,8 +11910,7 @@ def test_columns_setup(cursor, db_connection): cursor.execute("DROP TABLE IF EXISTS pytest_cols_schema.columns_special_test") # Create test table with various column types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_cols_schema.columns_test ( id INT PRIMARY KEY, name NVARCHAR(100) NOT NULL, @@ -12154,12 +11922,10 @@ def test_columns_setup(cursor, db_connection): notes TEXT NULL, [computed_col] AS (name + ' - ' + CAST(id AS VARCHAR(10))) ) - """ - ) + """) # Create table with special column names and edge cases - fix the problematic column name - cursor.execute( - """ + cursor.execute(""" CREATE TABLE pytest_cols_schema.columns_special_test ( [ID] INT PRIMARY KEY, [User Name] NVARCHAR(100) NULL, @@ -12171,8 +11937,7 @@ def test_columns_setup(cursor, db_connection): [Column/With/Slashes] VARCHAR(20) NULL, [Column_With_Underscores] VARCHAR(20) NULL -- Changed from problematic nested brackets ) - """ - ) + """) db_connection.commit() except Exception as e: @@ -12636,25 +12401,21 @@ def test_lowercase_attribute(cursor, db_connection): try: # Create a test table with mixed-case column names - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lowercase_test ( ID INT PRIMARY KEY, UserName VARCHAR(50), EMAIL_ADDRESS VARCHAR(100), PhoneNumber VARCHAR(20) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_lowercase_test (ID, UserName, EMAIL_ADDRESS, PhoneNumber) VALUES (1, 'JohnDoe', 'john@example.com', '555-1234') - """ - ) + """) db_connection.commit() # First test with lowercase=False (default) @@ -12709,14 +12470,12 @@ def test_decimal_separator_function(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_separator_test ( id INT PRIMARY KEY, decimal_value DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test values with default separator (.) @@ -12798,8 +12557,7 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_multi_test ( id INT PRIMARY KEY, positive_value DECIMAL(10, 2), @@ -12807,16 +12565,13 @@ def test_decimal_separator_with_multiple_values(cursor, db_connection): zero_value DECIMAL(10, 2), small_value DECIMAL(10, 4) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_multi_test VALUES (1, 123.45, -67.89, 0.00, 0.0001) - """ - ) + """) db_connection.commit() # Test with default separator first @@ -12853,23 +12608,19 @@ def test_decimal_separator_calculations(cursor, db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_calc_test ( id INT PRIMARY KEY, value1 DECIMAL(10, 2), value2 DECIMAL(10, 2) ) - """ - ) + """) db_connection.commit() # Insert test data - cursor.execute( - """ + cursor.execute(""" INSERT INTO #pytest_decimal_calc_test VALUES (1, 10.25, 5.75) - """ - ) + """) db_connection.commit() # Test with default separator @@ -12906,14 +12657,12 @@ def test_executemany_with_uuids(cursor, db_connection): table_name = "#pytest_uuid_batch" try: cursor.execute(f"DROP TABLE IF EXISTS {table_name}") - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id UNIQUEIDENTIFIER, description NVARCHAR(50) ) - """ - ) + """) db_connection.commit() # Prepare test data: mix of UUIDs and None @@ -13061,13 +12810,11 @@ def test_date_string_parameter_binding(cursor, db_connection): table_name = "#pytest_date_string" try: drop_table_if_exists(cursor, table_name) - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( a_column VARCHAR(20) ) - """ - ) + """) cursor.execute(f"INSERT INTO {table_name} (a_column) VALUES ('string1'), ('string2')") db_connection.commit() @@ -13094,13 +12841,11 @@ def test_time_string_parameter_binding(cursor, db_connection): table_name = "#pytest_time_string" try: drop_table_if_exists(cursor, table_name) - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( time_col VARCHAR(22) ) - """ - ) + """) cursor.execute(f"INSERT INTO {table_name} (time_col) VALUES ('prefix_14:30:45_suffix')") db_connection.commit() @@ -13125,13 +12870,11 @@ def test_datetime_string_parameter_binding(cursor, db_connection): table_name = "#pytest_datetime_string" try: drop_table_if_exists(cursor, table_name) - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( datetime_col VARCHAR(33) ) - """ - ) + """) cursor.execute( f"INSERT INTO {table_name} (datetime_col) VALUES ('prefix_2025-08-12T14:30:45_suffix')" ) @@ -13995,14 +13738,12 @@ def test_column_metadata_error_handling(cursor): """Test column metadata retrieval error handling (Lines 1156-1167).""" # Execute a complex query that might stress metadata retrieval - cursor.execute( - """ + cursor.execute(""" SELECT CAST(1 as INT) as int_col, CAST('test' as NVARCHAR(100)) as nvarchar_col, CAST(NEWID() as UNIQUEIDENTIFIER) as guid_col - """ - ) + """) # This should exercise the metadata retrieval code paths # If there are any errors, they should be logged but not crash @@ -14118,14 +13859,12 @@ def test_row_uuid_processing_with_braces(cursor, db_connection): drop_table_if_exists(cursor, "#pytest_uuid_braces") # Create table with UNIQUEIDENTIFIER column - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_uuid_braces ( id INT IDENTITY(1,1), guid_col UNIQUEIDENTIFIER ) - """ - ) + """) # Insert a GUID with braces (this is how SQL Server often returns them) test_guid = "12345678-1234-5678-9ABC-123456789ABC" @@ -14169,14 +13908,12 @@ def test_row_uuid_processing_sql_guid_type(cursor, db_connection): drop_table_if_exists(cursor, "#pytest_sql_guid_type") # Create table with UNIQUEIDENTIFIER column - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_sql_guid_type ( id INT, guid_col UNIQUEIDENTIFIER ) - """ - ) + """) # Insert test data test_guid = "ABCDEF12-3456-7890-ABCD-1234567890AB" @@ -14222,14 +13959,12 @@ def test_row_output_converter_overflow_error(cursor, db_connection): try: # Create a table with integer column drop_table_if_exists(cursor, "#pytest_overflow_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_overflow_test ( id INT, small_int TINYINT -- TINYINT can only hold 0-255 ) - """ - ) + """) # Insert a valid value first cursor.execute("INSERT INTO #pytest_overflow_test (id, small_int) VALUES (?, ?)", [1, 100]) @@ -14279,14 +14014,12 @@ def test_row_output_converter_general_exception(cursor, db_connection): try: # Create a table with string column drop_table_if_exists(cursor, "#pytest_exception_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_exception_test ( id INT, text_col VARCHAR(50) ) - """ - ) + """) # Insert test data cursor.execute( @@ -14337,14 +14070,12 @@ def test_row_cursor_log_method_availability(cursor, db_connection): try: # Create test data drop_table_if_exists(cursor, "#pytest_log_check") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_log_check ( id INT, value_col INT ) - """ - ) + """) cursor.execute("INSERT INTO #pytest_log_check (id, value_col) VALUES (?, ?)", [1, 42]) db_connection.commit() @@ -14372,8 +14103,7 @@ def test_all_numeric_types_with_nulls(cursor, db_connection): """Test NULL handling for all numeric types to ensure processor functions handle NULLs correctly""" try: drop_table_if_exists(cursor, "#pytest_all_numeric_nulls") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_all_numeric_nulls ( int_col INT, bigint_col BIGINT, @@ -14383,8 +14113,7 @@ def test_all_numeric_types_with_nulls(cursor, db_connection): real_col REAL, float_col FLOAT ) - """ - ) + """) db_connection.commit() # Insert row with all NULLs @@ -14426,16 +14155,14 @@ def test_lob_data_types(cursor, db_connection): """Test LOB (Large Object) data types to ensure LOB fallback paths are exercised""" try: drop_table_if_exists(cursor, "#pytest_lob_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_test ( id INT, text_lob VARCHAR(MAX), ntext_lob NVARCHAR(MAX), binary_lob VARBINARY(MAX) ) - """ - ) + """) db_connection.commit() # Create large data that will trigger LOB handling @@ -14468,14 +14195,12 @@ def test_lob_char_column_types(cursor, db_connection): """Test LOB fetching specifically for CHAR/VARCHAR columns (covers lines 3313-3314)""" try: drop_table_if_exists(cursor, "#pytest_lob_char") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_char ( id INT, char_lob VARCHAR(MAX) ) - """ - ) + """) db_connection.commit() # Create data large enough to trigger LOB path (>8000 bytes) @@ -14502,14 +14227,12 @@ def test_lob_wchar_column_types(cursor, db_connection): """Test LOB fetching specifically for WCHAR/NVARCHAR columns (covers lines 3358-3359)""" try: drop_table_if_exists(cursor, "#pytest_lob_wchar") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_wchar ( id INT, wchar_lob NVARCHAR(MAX) ) - """ - ) + """) db_connection.commit() # Create unicode data large enough to trigger LOB path (>4000 characters for NVARCHAR) @@ -14536,14 +14259,12 @@ def test_lob_binary_column_types(cursor, db_connection): """Test LOB fetching specifically for BINARY/VARBINARY columns (covers lines 3384-3385)""" try: drop_table_if_exists(cursor, "#pytest_lob_binary") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_binary ( id INT, binary_lob VARBINARY(MAX) ) - """ - ) + """) db_connection.commit() # Create binary data large enough to trigger LOB path (>8000 bytes) @@ -14570,16 +14291,14 @@ def test_zero_length_complex_types(cursor, db_connection): """Test zero-length data for complex types (covers lines 3531-3533)""" try: drop_table_if_exists(cursor, "#pytest_zero_length") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_zero_length ( id INT, empty_varchar VARCHAR(100), empty_nvarchar NVARCHAR(100), empty_binary VARBINARY(100) ) - """ - ) + """) db_connection.commit() # Insert empty (non-NULL) values @@ -14607,14 +14326,12 @@ def test_guid_with_nulls(cursor, db_connection): """Test GUID type with NULL values""" try: drop_table_if_exists(cursor, "#pytest_guid_nulls") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_guid_nulls ( id INT, guid_col UNIQUEIDENTIFIER ) - """ - ) + """) db_connection.commit() # Insert NULL GUID @@ -14641,14 +14358,12 @@ def test_datetimeoffset_with_nulls(cursor, db_connection): """Test DATETIMEOFFSET type with NULL values""" try: drop_table_if_exists(cursor, "#pytest_dto_nulls") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_dto_nulls ( id INT, dto_col DATETIMEOFFSET ) - """ - ) + """) db_connection.commit() # Insert NULL DATETIMEOFFSET @@ -14675,14 +14390,12 @@ def test_decimal_conversion_edge_cases(cursor, db_connection): """Test DECIMAL/NUMERIC type conversion including edge cases""" try: drop_table_if_exists(cursor, "#pytest_decimal_edge") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_edge ( id INT, dec_col DECIMAL(18, 4) ) - """ - ) + """) db_connection.commit() # Insert various decimal values including edge cases @@ -14803,8 +14516,7 @@ def test_all_numeric_types_with_nulls(cursor, db_connection): """Test NULL handling for all numeric types to ensure processor functions handle NULLs correctly""" try: drop_table_if_exists(cursor, "#pytest_all_numeric_nulls") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_all_numeric_nulls ( int_col INT, bigint_col BIGINT, @@ -14814,8 +14526,7 @@ def test_all_numeric_types_with_nulls(cursor, db_connection): real_col REAL, float_col FLOAT ) - """ - ) + """) db_connection.commit() # Insert row with all NULLs @@ -14857,16 +14568,14 @@ def test_lob_data_types(cursor, db_connection): """Test LOB (Large Object) data types to ensure LOB fallback paths are exercised""" try: drop_table_if_exists(cursor, "#pytest_lob_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_test ( id INT, text_lob VARCHAR(MAX), ntext_lob NVARCHAR(MAX), binary_lob VARBINARY(MAX) ) - """ - ) + """) db_connection.commit() # Create large data that will trigger LOB handling @@ -14899,14 +14608,12 @@ def test_lob_char_column_types(cursor, db_connection): """Test LOB fetching specifically for CHAR/VARCHAR columns (covers lines 3313-3314)""" try: drop_table_if_exists(cursor, "#pytest_lob_char") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_char ( id INT, char_lob VARCHAR(MAX) ) - """ - ) + """) db_connection.commit() # Create data large enough to trigger LOB path (>8000 bytes) @@ -14933,14 +14640,12 @@ def test_lob_wchar_column_types(cursor, db_connection): """Test LOB fetching specifically for WCHAR/NVARCHAR columns (covers lines 3358-3359)""" try: drop_table_if_exists(cursor, "#pytest_lob_wchar") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_wchar ( id INT, wchar_lob NVARCHAR(MAX) ) - """ - ) + """) db_connection.commit() # Create unicode data large enough to trigger LOB path (>4000 characters for NVARCHAR) @@ -14967,14 +14672,12 @@ def test_lob_binary_column_types(cursor, db_connection): """Test LOB fetching specifically for BINARY/VARBINARY columns (covers lines 3384-3385)""" try: drop_table_if_exists(cursor, "#pytest_lob_binary") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_lob_binary ( id INT, binary_lob VARBINARY(MAX) ) - """ - ) + """) db_connection.commit() # Create binary data large enough to trigger LOB path (>8000 bytes) @@ -15001,16 +14704,14 @@ def test_zero_length_complex_types(cursor, db_connection): """Test zero-length data for complex types (covers lines 3531-3533)""" try: drop_table_if_exists(cursor, "#pytest_zero_length") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_zero_length ( id INT, empty_varchar VARCHAR(100), empty_nvarchar NVARCHAR(100), empty_binary VARBINARY(100) ) - """ - ) + """) db_connection.commit() # Insert empty (non-NULL) values @@ -15038,14 +14739,12 @@ def test_guid_with_nulls(cursor, db_connection): """Test GUID type with NULL values""" try: drop_table_if_exists(cursor, "#pytest_guid_nulls") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_guid_nulls ( id INT, guid_col UNIQUEIDENTIFIER ) - """ - ) + """) db_connection.commit() # Insert NULL GUID @@ -15072,14 +14771,12 @@ def test_datetimeoffset_with_nulls(cursor, db_connection): """Test DATETIMEOFFSET type with NULL values""" try: drop_table_if_exists(cursor, "#pytest_dto_nulls") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_dto_nulls ( id INT, dto_col DATETIMEOFFSET ) - """ - ) + """) db_connection.commit() # Insert NULL DATETIMEOFFSET @@ -15106,14 +14803,12 @@ def test_decimal_conversion_edge_cases(cursor, db_connection): """Test DECIMAL/NUMERIC type conversion including edge cases""" try: drop_table_if_exists(cursor, "#pytest_decimal_edge") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_decimal_edge ( id INT, dec_col DECIMAL(18, 4) ) - """ - ) + """) db_connection.commit() # Insert various decimal values including edge cases @@ -15234,16 +14929,14 @@ def test_fetchall_with_integrity_constraint(cursor, db_connection): try: # Setup table with unique constraint cursor.execute("DROP TABLE IF EXISTS #uniq_cons_test") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #uniq_cons_test ( id INTEGER NOT NULL IDENTITY, data VARCHAR(50) NULL, PRIMARY KEY (id), UNIQUE (data) ) - """ - ) + """) # Insert initial row - should work cursor.execute( diff --git a/tests/test_008_logging_integration.py b/tests/test_008_logging_integration.py index a745bcbc..220cbdc6 100644 --- a/tests/test_008_logging_integration.py +++ b/tests/test_008_logging_integration.py @@ -11,7 +11,6 @@ from mssql_python import connect from mssql_python.logging import setup_logging, logger - # Skip all tests if no database connection string available pytestmark = pytest.mark.skipif( not os.getenv("DB_CONNECTION_STRING"), reason="Database connection string not provided" diff --git a/tests/test_011_performance_stress.py b/tests/test_011_performance_stress.py index 9f963632..7750fee5 100644 --- a/tests/test_011_performance_stress.py +++ b/tests/test_011_performance_stress.py @@ -53,15 +53,13 @@ def test_exception_mid_batch_no_corrupt_data(cursor, db_connection): drop_table_if_exists(cursor, "#pytest_mid_batch_exception") # Create simple table to test batch processing integrity - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_mid_batch_exception ( id INT, value NVARCHAR(50), amount FLOAT ) - """ - ) + """) db_connection.commit() # Insert 1000 rows using individual inserts to avoid executemany complications @@ -121,16 +119,14 @@ def test_python_c_api_null_handling_memory_pressure(cursor, db_connection): drop_table_if_exists(cursor, "#pytest_memory_pressure") # Create table with various string types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_memory_pressure ( id INT, varchar_col VARCHAR(1000), nvarchar_col NVARCHAR(1000), varbinary_col VARBINARY(1000) ) - """ - ) + """) db_connection.commit() # Insert test data @@ -192,16 +188,14 @@ def test_thousands_of_empty_strings_allocation_stress(cursor, db_connection): try: drop_table_if_exists(cursor, "#pytest_empty_stress") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_empty_stress ( id INT, empty_varchar VARCHAR(100), empty_nvarchar NVARCHAR(100), empty_varbinary VARBINARY(100) ) - """ - ) + """) db_connection.commit() # Insert 10,000 rows with empty strings @@ -277,16 +271,14 @@ def test_large_result_set_100k_rows_no_overflow(cursor, db_connection): try: drop_table_if_exists(cursor, "#pytest_100k_rows") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_100k_rows ( id INT, varchar_col VARCHAR(50), nvarchar_col NVARCHAR(50), int_col INT ) - """ - ) + """) db_connection.commit() # Insert 100,000 rows with sequential IDs and predictable data @@ -375,16 +367,14 @@ def test_very_large_lob_10mb_data_integrity(cursor, db_connection): try: drop_table_if_exists(cursor, "#pytest_10mb_lob") - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #pytest_10mb_lob ( id INT, varchar_lob VARCHAR(MAX), nvarchar_lob NVARCHAR(MAX), varbinary_lob VARBINARY(MAX) ) - """ - ) + """) db_connection.commit() # Create 10MB+ data @@ -494,15 +484,13 @@ def worker_thread(thread_id: int, conn_str: str, results_list: List, errors_list table_name = f"#pytest_concurrent_t{thread_id}" drop_table_if_exists(cursor, table_name) - cursor.execute( - f""" + cursor.execute(f""" CREATE TABLE {table_name} ( id INT, thread_id INT, data VARCHAR(100) ) - """ - ) + """) conn.commit() # Insert thread-specific data diff --git a/tests/test_013_SqlHandle_free_shutdown.py b/tests/test_013_SqlHandle_free_shutdown.py index 7e426cfb..9944d898 100644 --- a/tests/test_013_SqlHandle_free_shutdown.py +++ b/tests/test_013_SqlHandle_free_shutdown.py @@ -52,8 +52,7 @@ def test_aggressive_dbc_segfault_reproduction(self, conn_str): Expected with CURRENT CODE: May segfault (this is the bug we're testing for) Expected with FIXED CODE: No segfault """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys import gc from mssql_python import connect @@ -83,8 +82,7 @@ def test_aggressive_dbc_segfault_reproduction(self, conn_str): # Force immediate exit - this triggers finalize_garbage sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -117,8 +115,7 @@ def test_dbc_handle_outlives_env_handle(self, conn_str): Expected with CURRENT CODE: Likely segfault Expected with FIXED CODE: No segfault """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys import atexit from mssql_python import connect @@ -141,8 +138,7 @@ def on_exit(): print("Python GC will finalize DBC during shutdown") print("If DBC cleanup isn't skipped, SQLFreeHandle will access freed ENV") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -167,8 +163,7 @@ def test_force_gc_finalization_order_issue(self, conn_str): Expected with CURRENT CODE: May segfault Expected with FIXED CODE: No segfault """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys import gc import weakref @@ -207,8 +202,7 @@ def test_force_gc_finalization_order_issue(self, conn_str): print("Exiting - finalize_garbage will be called") print("If DBC handles aren't protected, segfault in SQLFreeHandle") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -234,8 +228,7 @@ def test_stmt_handle_cleanup_at_shutdown(self, conn_str): Expected: No segfault, clean exit """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys from mssql_python import connect @@ -251,8 +244,7 @@ def test_stmt_handle_cleanup_at_shutdown(self, conn_str): # Type 3 (STMT) handle should be skipped when pythonShuttingDown=true print("STMT handle cleanup test: Exiting without explicit cleanup") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -275,8 +267,7 @@ def test_dbc_handle_cleanup_at_shutdown(self, conn_str): Expected: No segfault, clean exit """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys from mssql_python import connect @@ -296,8 +287,7 @@ def test_dbc_handle_cleanup_at_shutdown(self, conn_str): # Type 2 (DBC) handles should be skipped when pythonShuttingDown=true print("DBC handle cleanup test: Exiting without explicit connection cleanup") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -326,8 +316,7 @@ def test_env_handle_cleanup_at_shutdown(self, conn_str): Note: ENV handle is static and destructs via normal C++ mechanisms, not during Python GC. This test verifies the overall flow. """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys from mssql_python import connect @@ -346,8 +335,7 @@ def test_env_handle_cleanup_at_shutdown(self, conn_str): # It does NOT have pythonShuttingDown protection (Type 1 not in check) print("ENV handle cleanup test: All connections closed properly") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -374,8 +362,7 @@ def test_mixed_handle_cleanup_at_shutdown(self, conn_str): Expected: No segfault, clean exit This tests the real-world scenario where cleanup is partial """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys from mssql_python import connect @@ -420,8 +407,7 @@ def test_mixed_handle_cleanup_at_shutdown(self, conn_str): # - Type 1 (ENV) handle: normal C++ static destruction print("Mixed handle cleanup test: Exiting with partial cleanup") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -446,8 +432,7 @@ def test_rapid_connection_churn_with_shutdown(self, conn_str): Expected: No segfault, proper handle cleanup order """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys import gc from mssql_python import connect @@ -475,8 +460,7 @@ def test_rapid_connection_churn_with_shutdown(self, conn_str): # Their DBC and STMT handles will be skipped during shutdown print("Rapid churn test: Exiting with mixed cleanup") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -499,8 +483,7 @@ def test_exception_during_query_with_shutdown(self, conn_str): Expected: No segfault, graceful error handling """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys from mssql_python import connect, ProgrammingError @@ -516,8 +499,7 @@ def test_exception_during_query_with_shutdown(self, conn_str): print("Exception test: Exiting after exception without cleanup") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -539,8 +521,7 @@ def test_weakref_cleanup_at_shutdown(self, conn_str): Expected: No segfault, proper weakref finalization """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys import weakref from mssql_python import connect @@ -571,8 +552,7 @@ def callback(ref): print("Weakref test: Exiting with weakrefs active") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -593,8 +573,7 @@ def test_gc_during_shutdown_with_circular_refs(self, conn_str): Expected: No segfault, proper cycle breaking """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys import gc from mssql_python import connect @@ -631,8 +610,7 @@ def execute_query(self): print("Circular ref test: Exiting after GC with cycles") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -653,8 +631,7 @@ def test_all_handle_types_comprehensive(self, conn_str): Expected: Clean shutdown with no segfaults """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import sys from mssql_python import connect @@ -705,8 +682,7 @@ def test_all_handle_types_comprehensive(self, conn_str): print("- Type 1 (ENV) handle: Normal C++ static destruction") print("=== Exiting ===") sys.exit(0) - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 @@ -950,8 +926,7 @@ def test_cleanup_connections_scenarios(self, conn_str, scenario, test_code, expe - empty_list: No errors with empty set - mixed_scenario: Mixed connection states handled correctly """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import mssql_python # Verify cleanup infrastructure exists @@ -962,8 +937,7 @@ def test_cleanup_connections_scenarios(self, conn_str, scenario, test_code, expe {test_code} print("{expected_msg}") - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=3 @@ -983,8 +957,7 @@ def test_active_connections_thread_safety(self, conn_str): - Cleanup can safely iterate while threads are registering - Lock prevents data corruption in WeakSet """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import mssql_python import threading import time @@ -1053,8 +1026,7 @@ def register_connections(thread_id, count): assert conn._closed, f"Connection {{conn.conn_id}} was not closed" print("Thread safety test: PASSED") - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 @@ -1076,8 +1048,7 @@ def test_cleanup_connections_list_copy_isolation(self, conn_str): 3. WeakSet can be modified (e.g., connections removed by GC) without breaking iteration 4. The copy prevents "Set changed size during iteration" RuntimeError """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import mssql_python import weakref import gc @@ -1152,8 +1123,7 @@ def close(self): print("List copy isolation: PASSED") print("[OK] connections_to_close = list(_active_connections) properly tested") - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=3 @@ -1177,8 +1147,7 @@ def test_cleanup_connections_weakset_modification_during_iteration(self, conn_st 2. With the list copy, iteration is safe even if WeakSet shrinks due to GC 3. The pattern prevents "dictionary changed size during iteration" type errors """ - script = textwrap.dedent( - f""" + script = textwrap.dedent(f""" import mssql_python import weakref import gc @@ -1244,8 +1213,7 @@ def close(self): print("WeakSet modification during iteration: PASSED") print("[OK] list() copy prevents 'set changed size during iteration' errors") - """ - ) + """) result = subprocess.run( [sys.executable, "-c", script], capture_output=True, text=True, timeout=3 diff --git a/tests/test_013_encoding_decoding.py b/tests/test_013_encoding_decoding.py index 9010647d..a30c061c 100644 --- a/tests/test_013_encoding_decoding.py +++ b/tests/test_013_encoding_decoding.py @@ -1078,15 +1078,13 @@ def test_setdecoding_with_unicode_data(db_connection): try: # Create test table with NVARCHAR columns for Unicode support - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_decoding_unicode ( id INT IDENTITY(1,1), ascii_col VARCHAR(100), unicode_col NVARCHAR(100) ) - """ - ) + """) # Test ASCII strings in VARCHAR (safe) ascii_strings = [ @@ -1161,8 +1159,7 @@ def test_encoding_decoding_comprehensive_unicode_characters(db_connection): try: # Create test table with different column types - use NVARCHAR for better Unicode support - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_encoding_comprehensive ( id INT PRIMARY KEY, varchar_col VARCHAR(1000), @@ -1170,8 +1167,7 @@ def test_encoding_decoding_comprehensive_unicode_characters(db_connection): text_col TEXT, ntext_col NTEXT ) - """ - ) + """) # Test cases with different Unicode character categories test_cases = [ @@ -1333,8 +1329,7 @@ def test_encoding_decoding_edge_case_data_types(db_connection): try: # Create table with various data types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_encoding_datatypes ( id INT PRIMARY KEY, varchar_small VARCHAR(50), @@ -1346,8 +1341,7 @@ def test_encoding_decoding_edge_case_data_types(db_connection): text_type TEXT, ntext_type NTEXT ) - """ - ) + """) # Test different encoding configurations test_configs = [ @@ -1639,16 +1633,14 @@ def test_encoding_decoding_large_dataset_performance(db_connection): cursor = db_connection.cursor() try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_large_encoding ( id INT PRIMARY KEY, ascii_data VARCHAR(1000), unicode_data NVARCHAR(1000), mixed_data NVARCHAR(MAX) ) - """ - ) + """) # Generate test data - ensure it fits in column sizes ascii_text = "This is ASCII text with numbers 12345." * 10 # ~400 chars @@ -1817,15 +1809,13 @@ def test_encoding_decoding_metadata_columns(db_connection): try: # Create table with Unicode column names if supported - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_metadata ( [normal_col] NVARCHAR(100), [column_with_unicode_测试] NVARCHAR(100), [special_chars_ñáéíóú] INT ) - """ - ) + """) # Test metadata decoding configuration db_connection.setdecoding(mssql_python.SQL_WMETADATA, encoding="utf-16le", ctype=SQL_WCHAR) @@ -1899,8 +1889,7 @@ def test_encoding_decoding_stress_test_comprehensive(db_connection): cursor = db_connection.cursor() try: - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #stress_test_encoding ( id INT IDENTITY(1,1) PRIMARY KEY, ascii_text VARCHAR(500), @@ -1908,8 +1897,7 @@ def test_encoding_decoding_stress_test_comprehensive(db_connection): binary_data VARBINARY(500), mixed_content NVARCHAR(MAX) ) - """ - ) + """) # Generate diverse test data test_datasets = [] @@ -2030,15 +2018,13 @@ def test_encoding_decoding_sql_char_various_encodings(db_connection): try: # Create test table with VARCHAR columns (SQL_CHAR type) - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_sql_char_encodings ( id INT PRIMARY KEY, data_col VARCHAR(100), description VARCHAR(200) ) - """ - ) + """) # Define various encoding types to test with SQL_CHAR encoding_tests = [ @@ -2315,15 +2301,13 @@ def test_encoding_decoding_sql_char_with_unicode_fallback(db_connection): try: # Create test table with both VARCHAR and NVARCHAR - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_unicode_fallback ( id INT PRIMARY KEY, varchar_data VARCHAR(100), nvarchar_data NVARCHAR(100) ) - """ - ) + """) # Test Unicode data unicode_test_cases = [ @@ -2394,15 +2378,13 @@ def test_encoding_decoding_sql_char_native_character_sets(db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_native_chars ( id INT PRIMARY KEY, data VARCHAR(200), encoding_used VARCHAR(50) ) - """ - ) + """) # Test encoding-specific character sets that should work encoding_native_tests = [ @@ -2537,15 +2519,13 @@ def test_encoding_decoding_sql_char_boundary_encoding_cases(db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_encoding_boundaries ( id INT PRIMARY KEY, test_data VARCHAR(500), test_type VARCHAR(100) ) - """ - ) + """) # Test boundary cases for different encodings boundary_tests = [ @@ -2646,16 +2626,14 @@ def test_encoding_decoding_sql_char_unicode_issue_diagnosis(db_connection): try: # Create test table with both VARCHAR and NVARCHAR for comparison - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_unicode_issue ( id INT PRIMARY KEY, varchar_col VARCHAR(100), nvarchar_col NVARCHAR(100), encoding_used VARCHAR(50) ) - """ - ) + """) # Test Unicode strings that commonly cause issues test_strings = [ @@ -2701,11 +2679,9 @@ def test_encoding_decoding_sql_char_unicode_issue_diagnosis(db_connection): ) # Retrieve results - cursor.execute( - """ + cursor.execute(""" SELECT varchar_col, nvarchar_col FROM #test_unicode_issue WHERE id = 1 - """ - ) + """) result = cursor.fetchone() if result: @@ -2760,8 +2736,7 @@ def test_encoding_decoding_sql_char_best_practices_guide(db_connection): try: # Create test table demonstrating different column types - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_best_practices ( id INT PRIMARY KEY, -- ASCII-safe columns (VARCHAR with SQL_CHAR) @@ -2775,8 +2750,7 @@ def test_encoding_decoding_sql_char_best_practices_guide(db_connection): -- Mixed approach column safe_text VARCHAR(200) ) - """ - ) + """) # Configure optimal settings db_connection.setencoding(encoding="utf-8", ctype=SQL_CHAR) # For ASCII data @@ -4992,15 +4966,13 @@ def test_execute_executemany_encoding_consistency(db_connection): try: # Create test table that can handle both VARCHAR and NVARCHAR data - cursor.execute( - """ + cursor.execute(""" CREATE TABLE #test_encoding_consistency ( id INT IDENTITY(1,1) PRIMARY KEY, varchar_col VARCHAR(1000) COLLATE SQL_Latin1_General_CP1_CI_AS, nvarchar_col NVARCHAR(1000) ) - """ - ) + """) # Test data with various encoding challenges # Using ASCII-safe characters that work across different encodings @@ -5053,13 +5025,11 @@ def test_execute_executemany_encoding_consistency(db_connection): ) # Retrieve immediately to verify encoding worked - cursor.execute( - """ + cursor.execute(""" SELECT varchar_col, nvarchar_col FROM #test_encoding_consistency WHERE id = (SELECT MAX(id) FROM #test_encoding_consistency) - """ - ) + """) result = cursor.fetchone() execute_results.append((result[0], result[1])) @@ -5084,13 +5054,11 @@ def test_execute_executemany_encoding_consistency(db_connection): ) # Retrieve all results from executemany - cursor.execute( - """ + cursor.execute(""" SELECT varchar_col, nvarchar_col FROM #test_encoding_consistency ORDER BY id - """ - ) + """) executemany_results = cursor.fetchall() # Verify executemany results match execute results @@ -5127,13 +5095,11 @@ def test_execute_executemany_encoding_consistency(db_connection): test_string, ) - cursor.execute( - """ + cursor.execute(""" SELECT nvarchar_col FROM #test_encoding_consistency WHERE id = (SELECT MAX(id) FROM #test_encoding_consistency) - """ - ) + """) result = cursor.fetchone() unicode_execute_results.append(result[0]) @@ -5160,13 +5126,11 @@ def test_execute_executemany_encoding_consistency(db_connection): unicode_params, ) - cursor.execute( - """ + cursor.execute(""" SELECT nvarchar_col FROM #test_encoding_consistency ORDER BY id - """ - ) + """) unicode_executemany_results = cursor.fetchall() # Compare Unicode results diff --git a/tests/test_cache_invalidation.py b/tests/test_cache_invalidation.py index 59f81ccd..fa1d34e2 100644 --- a/tests/test_cache_invalidation.py +++ b/tests/test_cache_invalidation.py @@ -7,7 +7,6 @@ silent data corruption. """ - import pytest import mssql_python @@ -23,38 +22,29 @@ def test_cursor_cache_invalidation_different_column_orders(db_connection): try: # Setup test tables with different column orders and types - cursor.execute( - """ + cursor.execute(""" IF OBJECT_ID('tempdb..#test_cache_table1') IS NOT NULL DROP TABLE #test_cache_table1 - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" CREATE TABLE #test_cache_table1 ( id INT, name VARCHAR(50), age INT, salary DECIMAL(10,2) ) - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" INSERT INTO #test_cache_table1 VALUES (1, 'Alice', 30, 50000.00), (2, 'Bob', 25, 45000.00) - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" IF OBJECT_ID('tempdb..#test_cache_table2') IS NOT NULL DROP TABLE #test_cache_table2 - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" CREATE TABLE #test_cache_table2 ( salary DECIMAL(10,2), age INT, @@ -62,15 +52,12 @@ def test_cursor_cache_invalidation_different_column_orders(db_connection): name VARCHAR(50), bonus FLOAT ) - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" INSERT INTO #test_cache_table2 VALUES (60000.00, 35, 3, 'Charlie', 5000.5), (55000.00, 28, 4, 'Diana', 3000.75) - """ - ) + """) # Execute first query - columns: id, name, age, salary cursor.execute("SELECT id, name, age, salary FROM #test_cache_table1 ORDER BY id") @@ -141,13 +128,11 @@ def test_cursor_cache_invalidation_stored_procedure_multiple_resultsets(db_conne # the scenario where cached maps need to be invalidated between different queries # First result set: user info (3 columns) - cursor.execute( - """ + cursor.execute(""" SELECT 1 as user_id, 'John' as username, 'john@example.com' as email UNION ALL SELECT 2, 'Jane', 'jane@example.com' - """ - ) + """) # Validate first result set - user info assert len(cursor.description) == 3 @@ -162,13 +147,11 @@ def test_cursor_cache_invalidation_stored_procedure_multiple_resultsets(db_conne assert user_rows[0].email == "john@example.com" # Execute second query with completely different structure - cursor.execute( - """ + cursor.execute(""" SELECT 101 as product_id, 'Widget A' as product_name, 29.99 as price, 100 as stock_qty UNION ALL SELECT 102, 'Widget B', 39.99, 50 - """ - ) + """) # Validate second result set - product info (different structure) assert len(cursor.description) == 4 @@ -212,14 +195,11 @@ def test_cursor_cache_invalidation_metadata_then_select(db_connection): try: # Create test table - cursor.execute( - """ + cursor.execute(""" IF OBJECT_ID('tempdb..#test_metadata_table') IS NOT NULL DROP TABLE #test_metadata_table - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" CREATE TABLE #test_metadata_table ( meta_id INT PRIMARY KEY, meta_name VARCHAR(100), @@ -227,19 +207,15 @@ def test_cursor_cache_invalidation_metadata_then_select(db_connection): meta_date DATETIME, meta_flag BIT ) - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" INSERT INTO #test_metadata_table VALUES (1, 'Config1', 123.4567, '2023-01-15 10:30:00', 1), (2, 'Config2', 987.6543, '2023-02-20 14:45:00', 0) - """ - ) + """) # First: Execute a metadata-only query (no actual data rows) - cursor.execute( - """ + cursor.execute(""" SELECT COLUMN_NAME, DATA_TYPE, @@ -249,8 +225,7 @@ def test_cursor_cache_invalidation_metadata_then_select(db_connection): WHERE TABLE_NAME = 'test_metadata_table' AND TABLE_SCHEMA = 'tempdb' ORDER BY ORDINAL_POSITION - """ - ) + """) # Verify metadata result structure meta_description = cursor.description @@ -324,30 +299,24 @@ def test_cursor_cache_invalidation_fetch_methods_consistency(db_connection): try: # Create test data - cursor.execute( - """ + cursor.execute(""" IF OBJECT_ID('tempdb..#test_fetch_cache') IS NOT NULL DROP TABLE #test_fetch_cache - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" CREATE TABLE #test_fetch_cache ( first_col VARCHAR(20), second_col INT, third_col DECIMAL(8,2) ) - """ - ) - cursor.execute( - """ + """) + cursor.execute(""" INSERT INTO #test_fetch_cache VALUES ('Row1', 10, 100.50), ('Row2', 20, 200.75), ('Row3', 30, 300.25), ('Row4', 40, 400.00) - """ - ) + """) # Execute first query with specific column order cursor.execute( @@ -411,11 +380,9 @@ def test_cache_specific_close_cleanup_validation(db_connection): try: # Setup test data - cursor.execute( - """ + cursor.execute(""" SELECT 1 as cache_col1, 'test' as cache_col2, 99.99 as cache_col3 - """ - ) + """) # Verify cache is populated assert cursor.description is not None @@ -560,15 +527,12 @@ def test_real_stored_procedure_cache_validation(db_connection): try: # Create a temporary stored procedure with multiple result sets - cursor.execute( - """ + cursor.execute(""" IF OBJECT_ID('tempdb..#sp_test_cache') IS NOT NULL DROP PROCEDURE #sp_test_cache - """ - ) + """) - cursor.execute( - """ + cursor.execute(""" CREATE PROCEDURE #sp_test_cache AS BEGIN @@ -581,8 +545,7 @@ def test_real_stored_procedure_cache_validation(db_connection): -- Third result set: Summary (yet another structure) SELECT GETDATE() as report_date, 'Cache Test' as report_type, 1 as version_num; END - """ - ) + """) # Execute the stored procedure cursor.execute("EXEC #sp_test_cache")