diff --git a/tests/test_013_SqlHandle_free_shutdown.py b/tests/test_013_SqlHandle_free_shutdown.py index 950757c2..7e426cfb 100644 --- a/tests/test_013_SqlHandle_free_shutdown.py +++ b/tests/test_013_SqlHandle_free_shutdown.py @@ -32,6 +32,8 @@ import threading import time +import pytest + class TestHandleFreeShutdown: """Test SqlHandle::free() behavior for all handle types during Python shutdown.""" @@ -63,12 +65,10 @@ def test_aggressive_dbc_segfault_reproduction(self, conn_str): # This maximizes the chance of DBC handles being finalized # AFTER the static ENV handle has destructed connections = [] - for i in range(10): # Reduced from 20 to avoid timeout + for i in range(5): # Reduced for faster execution conn = connect("{conn_str}") # Don't even create cursors - just DBC handles connections.append(conn) - if i % 3 == 0: - print(f"Created {{i+1}} connections...") print(f"Created {{len(connections)}} DBC handles") print("Forcing GC to ensure objects are tracked...") @@ -87,7 +87,7 @@ def test_aggressive_dbc_segfault_reproduction(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) # Check for segfault @@ -103,7 +103,7 @@ def test_aggressive_dbc_segfault_reproduction(self, conn_str): ), f"SEGFAULT reproduced with signal {signal_num} - DBC handles not protected" else: assert result.returncode == 0, f"Process failed. stderr: {result.stderr}" - assert "Created 10 DBC handles" in result.stdout + assert "Created 5 DBC handles" in result.stdout print(f"PASS: No segfault - DBC handles properly protected during shutdown") def test_dbc_handle_outlives_env_handle(self, conn_str): @@ -145,7 +145,7 @@ def on_exit(): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) if result.returncode < 0: @@ -180,7 +180,7 @@ def test_force_gc_finalization_order_issue(self, conn_str): connections = [] weakrefs = [] - for i in range(10): # Reduced from 15 to avoid timeout + for i in range(5): # Reduced for faster execution conn = connect("{conn_str}") wr = weakref.ref(conn) connections.append(conn) @@ -194,9 +194,9 @@ def test_force_gc_finalization_order_issue(self, conn_str): # Delete strong references del connections - # Force multiple GC cycles + # Force GC cycles print("Forcing GC cycles...") - for i in range(5): + for i in range(2): collected = gc.collect() print(f"GC cycle {{i+1}}: collected {{collected}} objects") @@ -211,7 +211,7 @@ def test_force_gc_finalization_order_issue(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) if result.returncode < 0: @@ -255,7 +255,7 @@ def test_stmt_handle_cleanup_at_shutdown(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -300,7 +300,7 @@ def test_dbc_handle_cleanup_at_shutdown(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -350,7 +350,7 @@ def test_env_handle_cleanup_at_shutdown(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -424,7 +424,7 @@ def test_mixed_handle_cleanup_at_shutdown(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -453,7 +453,7 @@ def test_rapid_connection_churn_with_shutdown(self, conn_str): from mssql_python import connect # Create and delete connections rapidly - for i in range(10): + for i in range(6): conn = connect("{conn_str}") cursor = conn.cursor() cursor.execute(f"SELECT {{i}} AS test") @@ -465,7 +465,7 @@ def test_rapid_connection_churn_with_shutdown(self, conn_str): conn.close() # Leave odd-numbered connections open - print("Created 10 connections, closed 5 explicitly") + print("Created 6 connections, closed 3 explicitly") # Force GC before shutdown gc.collect() @@ -479,11 +479,11 @@ def test_rapid_connection_churn_with_shutdown(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" - assert "Created 10 connections, closed 5 explicitly" in result.stdout + assert "Created 6 connections, closed 3 explicitly" in result.stdout assert "Rapid churn test: Exiting with mixed cleanup" in result.stdout print(f"PASS: Rapid connection churn with shutdown") @@ -520,7 +520,7 @@ def test_exception_during_query_with_shutdown(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -575,7 +575,7 @@ def callback(ref): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -635,7 +635,7 @@ def execute_query(self): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -709,7 +709,7 @@ def test_all_handle_types_comprehensive(self, conn_str): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=5 ) assert result.returncode == 0, f"Process crashed. stderr: {result.stderr}" @@ -721,24 +721,12 @@ def test_all_handle_types_comprehensive(self, conn_str): assert "=== Exiting ===" in result.stdout print(f"PASS: Comprehensive all handle types test") - def test_cleanup_connections_normal_flow(self, conn_str): - """ - Test _cleanup_connections() with normal active connections. - - Validates that: - 1. Active connections (_closed=False) are properly closed - 2. The cleanup function is registered with atexit - 3. Connections can be registered and tracked - """ - script = textwrap.dedent( - f""" - import mssql_python - - # Verify cleanup infrastructure exists - assert hasattr(mssql_python, '_active_connections'), "Missing _active_connections" - assert hasattr(mssql_python, '_cleanup_connections'), "Missing _cleanup_connections" - assert hasattr(mssql_python, '_register_connection'), "Missing _register_connection" - + @pytest.mark.parametrize( + "scenario,test_code,expected_msg", + [ + ( + "normal_flow", + """ # Create mock connection to test registration and cleanup class MockConnection: def __init__(self): @@ -758,30 +746,12 @@ def close(self): mssql_python._cleanup_connections() assert mock_conn.close_called, "close() should have been called" assert mock_conn._closed, "Connection should be marked as closed" - - print("Normal flow: PASSED") - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Normal flow: PASSED" in result.stdout - print(f"PASS: Cleanup connections normal flow") - - def test_cleanup_connections_already_closed(self, conn_str): - """ - Test _cleanup_connections() with already closed connections. - - Validates that connections with _closed=True are skipped - and close() is not called again. - """ - script = textwrap.dedent( - f""" - import mssql_python - + """, + "Normal flow: PASSED", + ), + ( + "already_closed", + """ class MockConnection: def __init__(self): self._closed = True # Already closed @@ -798,30 +768,12 @@ def close(self): # Cleanup should skip this connection mssql_python._cleanup_connections() assert not mock_conn.close_called, "close() should NOT have been called" - - print("Already closed: PASSED") - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Already closed: PASSED" in result.stdout - print(f"PASS: Cleanup connections already closed") - - def test_cleanup_connections_missing_attribute(self, conn_str): - """ - Test _cleanup_connections() with connections missing _closed attribute. - - Validates that hasattr() check prevents AttributeError and - cleanup continues gracefully. - """ - script = textwrap.dedent( - f""" - import mssql_python - + """, + "Already closed: PASSED", + ), + ( + "missing_attribute", + """ class MinimalConnection: # No _closed attribute def close(self): @@ -833,32 +785,12 @@ def close(self): # Should not crash mssql_python._cleanup_connections() - - print("Missing attribute: PASSED") - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Missing attribute: PASSED" in result.stdout - print(f"PASS: Cleanup connections missing _closed attribute") - - def test_cleanup_connections_exception_handling(self, conn_str): - """ - Test _cleanup_connections() exception handling. - - Validates that: - 1. Exceptions during close() are caught and silently ignored - 2. One failing connection doesn't prevent cleanup of others - 3. The function completes successfully despite errors - """ - script = textwrap.dedent( - f""" - import mssql_python - + """, + "Missing attribute: PASSED", + ), + ( + "exception_handling", + """ class GoodConnection: def __init__(self): self._closed = False @@ -886,32 +818,15 @@ def close(self): mssql_python._cleanup_connections() # Should not raise despite bad_conn throwing exception assert good_conn.close_called, "Good connection should still be closed" - print("Exception handling: PASSED") except Exception as e: print(f"Exception handling: FAILED - Exception escaped: {{e}}") raise - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Exception handling: PASSED" in result.stdout - print(f"PASS: Cleanup connections exception handling") - - def test_cleanup_connections_multiple_connections(self, conn_str): - """ - Test _cleanup_connections() with multiple connections. - - Validates that all registered connections are processed - and closed in the cleanup iteration. - """ - script = textwrap.dedent( - f""" - import mssql_python - + """, + "Exception handling: PASSED", + ), + ( + "multiple_connections", + """ class TestConnection: count = 0 @@ -935,31 +850,12 @@ def close(self): assert TestConnection.count == 5, f"All 5 connections should be closed, got {{TestConnection.count}}" assert all(c.close_called for c in connections), "All connections should have close() called" - - print("Multiple connections: PASSED") - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Multiple connections: PASSED" in result.stdout - print(f"PASS: Cleanup connections multiple connections") - - def test_cleanup_connections_weakset_behavior(self, conn_str): - """ - Test _cleanup_connections() WeakSet behavior. - - Validates that: - 1. WeakSet automatically removes garbage collected connections - 2. Only live references are processed during cleanup - 3. No crashes occur with GC'd connections - """ - script = textwrap.dedent( - f""" - import mssql_python + """, + "Multiple connections: PASSED", + ), + ( + "weakset_behavior", + """ import gc class TestConnection: @@ -982,62 +878,23 @@ def close(self): # Cleanup should not crash with removed connections mssql_python._cleanup_connections() - - print("WeakSet behavior: PASSED") - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "WeakSet behavior: PASSED" in result.stdout - print(f"PASS: Cleanup connections WeakSet behavior") - - def test_cleanup_connections_empty_list(self, conn_str): - """ - Test _cleanup_connections() with empty connections list. - - Validates that cleanup completes successfully with no registered - connections without any errors. - """ - script = textwrap.dedent( - f""" - import mssql_python - + """, + "WeakSet behavior: PASSED", + ), + ( + "empty_list", + """ # Clear any existing connections mssql_python._active_connections.clear() # Should not crash with empty set mssql_python._cleanup_connections() - - print("Empty list: PASSED") - """ - ) - - result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 - ) - - assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Empty list: PASSED" in result.stdout - print(f"PASS: Cleanup connections empty list") - - def test_cleanup_connections_mixed_scenario(self, conn_str): - """ - Test _cleanup_connections() with mixed connection states. - - Validates handling of: - - Open connections (should be closed) - - Already closed connections (should be skipped) - - Connections that throw exceptions (should be caught) - - All in one cleanup run - """ - script = textwrap.dedent( - f""" - import mssql_python - + """, + "Empty list: PASSED", + ), + ( + "mixed_scenario", + """ class OpenConnection: def __init__(self): self._closed = False @@ -1074,18 +931,47 @@ def close(self): mssql_python._cleanup_connections() assert open_conn.close_called, "Open connection should have been closed" + """, + "Mixed scenario: PASSED", + ), + ], + ) + def test_cleanup_connections_scenarios(self, conn_str, scenario, test_code, expected_msg): + """ + Test _cleanup_connections() with various scenarios. + + Scenarios tested: + - normal_flow: Active connections properly closed + - already_closed: Closed connections skipped + - missing_attribute: Gracefully handles missing _closed attribute + - exception_handling: Exceptions caught, cleanup continues + - multiple_connections: All connections processed + - weakset_behavior: Auto-removes GC'd connections + - empty_list: No errors with empty set + - mixed_scenario: Mixed connection states handled correctly + """ + script = textwrap.dedent( + f""" + import mssql_python + + # Verify cleanup infrastructure exists + assert hasattr(mssql_python, '_active_connections'), "Missing _active_connections" + assert hasattr(mssql_python, '_cleanup_connections'), "Missing _cleanup_connections" + assert hasattr(mssql_python, '_register_connection'), "Missing _register_connection" + + {test_code} - print("Mixed scenario: PASSED") + print("{expected_msg}") """ ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=3 ) assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" - assert "Mixed scenario: PASSED" in result.stdout - print(f"PASS: Cleanup connections mixed scenario") + assert expected_msg in result.stdout + print(f"PASS: Cleanup connections scenario '{scenario}'") def test_active_connections_thread_safety(self, conn_str): """ @@ -1171,7 +1057,7 @@ def register_connections(thread_id, count): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=30 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 ) assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" @@ -1270,7 +1156,7 @@ def close(self): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=3 ) assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" @@ -1362,7 +1248,7 @@ def close(self): ) result = subprocess.run( - [sys.executable, "-c", script], capture_output=True, text=True, timeout=10 + [sys.executable, "-c", script], capture_output=True, text=True, timeout=3 ) assert result.returncode == 0, f"Test failed. stderr: {result.stderr}" diff --git a/tests/test_014_ddbc_bindings_coverage.py b/tests/test_014_ddbc_bindings_coverage.py index f82fb11e..ee08d6b3 100644 --- a/tests/test_014_ddbc_bindings_coverage.py +++ b/tests/test_014_ddbc_bindings_coverage.py @@ -17,13 +17,9 @@ class TestIsValidUnicodeScalar: """Test the IsValidUnicodeScalar function (ddbc_bindings.h lines 74-78).""" - def test_valid_scalar_values(self): - """Test valid Unicode scalar values.""" - import mssql_python - from mssql_python import connect - - # Valid scalar values (not surrogates, <= 0x10FFFF) - valid_chars = [ + @pytest.mark.parametrize( + "char", + [ "\u0000", # NULL "\u007f", # Last ASCII "\u0080", # First 2-byte @@ -34,321 +30,175 @@ def test_valid_scalar_values(self): "\uffff", # Last BMP "\U00010000", # First supplementary "\U0010ffff", # Last valid Unicode - ] + ], + ) + def test_valid_scalar_values(self, char): + """Test valid Unicode scalar values using Binary() for faster execution.""" + from mssql_python.type import Binary + + # Test through Binary() which exercises the conversion code + result = Binary(char) + assert len(result) > 0 - for char in valid_chars: - try: - conn_str = f"Server=test;Database=DB{char};Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - def test_above_max_codepoint(self): - """Test code points > 0x10FFFF (ddbc_bindings.h line 76 first condition).""" - # Python won't let us create invalid codepoints easily, but we can test - # through the Binary() function which uses UTF-8 decode + def test_boundary_codepoints(self): + """Test boundary code points including max valid and surrogate range.""" from mssql_python.type import Binary - # Test valid maximum + # Test valid maximum (line 76) max_valid = "\U0010ffff" result = Binary(max_valid) assert len(result) > 0 - # Invalid UTF-8 that would decode to > 0x10FFFF is handled by decoder - # and replaced with U+FFFD + # Test surrogate boundaries (line 77) + before_surrogate = "\ud7ff" + result = Binary(before_surrogate) + assert len(result) > 0 + + after_surrogate = "\ue000" + result = Binary(after_surrogate) + assert len(result) > 0 + + # Invalid UTF-8 that would decode to > 0x10FFFF invalid_above_max = b"\xf4\x90\x80\x80" # Would be 0x110000 result = invalid_above_max.decode("utf-8", errors="replace") - # Should contain replacement character or be handled assert len(result) > 0 - def test_surrogate_range(self): - """Test surrogate range 0xD800-0xDFFF (ddbc_bindings.h line 77 second condition).""" - import mssql_python - from mssql_python import connect - - # Test boundaries around surrogate range - # These may fail to connect but test the conversion logic - - # Just before surrogate range (valid) - try: - conn_str = "Server=test;Database=DB\ud7ff;Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - # Inside surrogate range (invalid) - try: - conn_str = "Server=test;Database=DB\ud800;Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - try: - conn_str = "Server=test;Database=DB\udfff;Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - # Just after surrogate range (valid) - try: - conn_str = "Server=test;Database=DB\ue000;Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - @pytest.mark.skipif(platform.system() == "Windows", reason="Tests Unix-specific UTF-32 path") -class TestSQLWCHARUTF32Path: - """Test SQLWCHARToWString UTF-32 path (sizeof(SQLWCHAR) == 4, lines 120-130).""" - - def test_utf32_valid_scalars(self): - """Test UTF-32 path with valid scalar values (line 122 condition true).""" - import mssql_python - from mssql_python import connect +class TestUTF32ConversionPaths: + """Test UTF-32 conversion paths for SQLWCHARToWString and WStringToSQLWCHAR (lines 120-130, 159-167).""" + + @pytest.mark.parametrize( + "test_str", ["ASCII", "Hello", "Café", "中文", "中文测试", "😀", "😀🌍", "\U0010ffff"] + ) + def test_utf32_valid_scalars(self, test_str): + """Test UTF-32 path with valid scalar values using Binary() for faster execution.""" + from mssql_python.type import Binary - # On systems where SQLWCHAR is 4 bytes (UTF-32) # Valid scalars should be copied directly - valid_tests = [ - "ASCII", - "Café", - "中文", - "😀", - "\U0010ffff", - ] + result = Binary(test_str) + assert len(result) > 0 + # Verify round-trip + decoded = result.decode("utf-8") + assert decoded == test_str - for test_str in valid_tests: - try: - conn_str = f"Server=test;Database={test_str};Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - def test_utf32_invalid_scalars(self): - """Test UTF-32 path with invalid scalar values (line 122 condition false).""" - import mssql_python - from mssql_python import connect - - # Invalid scalars should be replaced with U+FFFD (lines 125-126) - # Python strings with surrogates - invalid_tests = [ + @pytest.mark.parametrize( + "test_str", + [ "Test\ud800", # High surrogate "\udc00Test", # Low surrogate - ] - - for test_str in invalid_tests: - try: - conn_str = f"Server=test;Database={test_str};Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - -@pytest.mark.skipif(platform.system() == "Windows", reason="Tests Unix-specific UTF-32 path") -class TestWStringToSQLWCHARUTF32Path: - """Test WStringToSQLWCHAR UTF-32 path (sizeof(SQLWCHAR) == 4, lines 159-167).""" - - def test_utf32_encode_valid(self): - """Test UTF-32 encoding with valid scalars (line 162 condition true).""" - import mssql_python - from mssql_python import connect - - valid_tests = [ - "Hello", - "Café", - "中文测试", - "😀🌍", - "\U0010ffff", - ] - - for test_str in valid_tests: - try: - conn_str = f"Server=test;Database={test_str};Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass - - def test_utf32_encode_invalid(self): - """Test UTF-32 encoding with invalid scalars (line 162 condition false, lines 164-165).""" - import mssql_python - from mssql_python import connect - - # Invalid scalars should be replaced with U+FFFD - invalid_tests = [ - "A\ud800B", # High surrogate - "\udc00C", # Low surrogate - ] + "A\ud800B", # High surrogate in middle + "\udc00C", # Low surrogate at start + ], + ) + def test_utf32_invalid_scalars(self, test_str): + """Test UTF-32 path with invalid scalar values (surrogates) using Binary().""" + from mssql_python.type import Binary - for test_str in invalid_tests: - try: - conn_str = f"Server=test;Database={test_str};Trusted_Connection=yes" - conn = connect(conn_str, autoconnect=False) - conn.close() - except Exception: - pass + # Invalid scalars should be handled (replaced with U+FFFD) + result = Binary(test_str) + assert len(result) > 0 @pytest.mark.skipif(platform.system() == "Windows", reason="Tests Unix-specific WideToUTF8 path") class TestWideToUTF8UnixPath: """Test WideToUTF8 Unix path (lines 415-453).""" - def test_1byte_utf8(self): - """Test 1-byte UTF-8 encoding (lines 424-427, code_point <= 0x7F).""" + def test_all_utf8_byte_lengths(self): + """Test 1-4 byte UTF-8 encoding (lines 424-445).""" from mssql_python.type import Binary - # ASCII characters should encode to 1 byte - ascii_tests = [ + # Combined test for all UTF-8 byte lengths + all_tests = [ + # 1-byte (ASCII, lines 424-427) ("A", b"A"), ("0", b"0"), (" ", b" "), ("~", b"~"), ("\x00", b"\x00"), ("\x7f", b"\x7f"), - ] - - for char, expected in ascii_tests: - result = Binary(char) - assert result == expected, f"1-byte encoding failed for {char!r}" - - def test_2byte_utf8(self): - """Test 2-byte UTF-8 encoding (lines 428-432, code_point <= 0x7FF).""" - from mssql_python.type import Binary - - # Characters requiring 2 bytes - two_byte_tests = [ + # 2-byte (lines 428-432) ("\u0080", b"\xc2\x80"), # Minimum 2-byte ("\u00a9", b"\xc2\xa9"), # Copyright © ("\u00ff", b"\xc3\xbf"), # ÿ ("\u07ff", b"\xdf\xbf"), # Maximum 2-byte - ] - - for char, expected in two_byte_tests: - result = Binary(char) - assert result == expected, f"2-byte encoding failed for {char!r}" - - def test_3byte_utf8(self): - """Test 3-byte UTF-8 encoding (lines 433-438, code_point <= 0xFFFF).""" - from mssql_python.type import Binary - - # Characters requiring 3 bytes - three_byte_tests = [ + # 3-byte (lines 433-438) ("\u0800", b"\xe0\xa0\x80"), # Minimum 3-byte ("\u4e2d", b"\xe4\xb8\xad"), # 中 ("\u20ac", b"\xe2\x82\xac"), # € ("\uffff", b"\xef\xbf\xbf"), # Maximum 3-byte - ] - - for char, expected in three_byte_tests: - result = Binary(char) - assert result == expected, f"3-byte encoding failed for {char!r}" - - def test_4byte_utf8(self): - """Test 4-byte UTF-8 encoding (lines 439-445, code_point <= 0x10FFFF).""" - from mssql_python.type import Binary - - # Characters requiring 4 bytes - four_byte_tests = [ + # 4-byte (lines 439-445) ("\U00010000", b"\xf0\x90\x80\x80"), # Minimum 4-byte ("\U0001f600", b"\xf0\x9f\x98\x80"), # 😀 ("\U0001f30d", b"\xf0\x9f\x8c\x8d"), # 🌍 ("\U0010ffff", b"\xf4\x8f\xbf\xbf"), # Maximum Unicode ] - for char, expected in four_byte_tests: + for char, expected in all_tests: result = Binary(char) - assert result == expected, f"4-byte encoding failed for {char!r}" + assert result == expected, f"UTF-8 encoding failed for {char!r}" @pytest.mark.skipif(platform.system() == "Windows", reason="Tests Unix-specific Utf8ToWString path") class TestUtf8ToWStringUnixPath: """Test Utf8ToWString decodeUtf8 lambda (lines 462-530).""" - def test_fast_path_ascii(self): + @pytest.mark.parametrize( + "test_str,expected", + [ + ("HelloWorld123", b"HelloWorld123"), # Pure ASCII + ("Hello😀", "Hello😀".encode("utf-8")), # Mixed ASCII + emoji + ], + ) + def test_fast_path_ascii(self, test_str, expected): """Test fast path for ASCII-only prefix (lines 539-542).""" from mssql_python.type import Binary - # Pure ASCII should use fast path - ascii_only = "HelloWorld123" - result = Binary(ascii_only) - expected = ascii_only.encode("utf-8") - assert result == expected - - # Mixed ASCII + non-ASCII should use fast path for ASCII prefix - mixed = "Hello😀" - result = Binary(mixed) - expected = mixed.encode("utf-8") + result = Binary(test_str) assert result == expected - def test_1byte_decode(self): - """Test 1-byte sequence decoding (lines 472-475).""" + def test_1byte_and_2byte_decode(self): + """Test 1-byte and 2-byte sequence decoding (lines 472-488).""" from mssql_python.type import Binary - # ASCII bytes should decode correctly - test_cases = [ + # 1-byte decode tests (lines 472-475) + one_byte_tests = [ (b"A", "A"), (b"Hello", "Hello"), (b"\x00\x7f", "\x00\x7f"), ] - for utf8_bytes, expected in test_cases: - # Test through round-trip - original = expected - result = Binary(original) + for utf8_bytes, expected in one_byte_tests: + result = Binary(expected) assert result == utf8_bytes - def test_2byte_decode_paths(self): - """Test 2-byte sequence decoding paths (lines 476-488).""" - from mssql_python.type import Binary - - # Test invalid continuation byte path (lines 477-480) - invalid_2byte = b"\xc2\x00" # Invalid continuation - result = invalid_2byte.decode("utf-8", errors="replace") - assert "\ufffd" in result, "Invalid 2-byte should produce replacement char" - - # Test valid decode path with cp >= 0x80 (lines 481-484) - valid_2byte = [ + # 2-byte valid decode tests (lines 481-484) + two_byte_tests = [ (b"\xc2\x80", "\u0080"), (b"\xc2\xa9", "\u00a9"), (b"\xdf\xbf", "\u07ff"), ] - for utf8_bytes, expected in valid_2byte: + for utf8_bytes, expected in two_byte_tests: result = utf8_bytes.decode("utf-8") assert result == expected - # Round-trip test encoded = Binary(expected) assert encoded == utf8_bytes - # Test overlong encoding rejection (lines 486-487) - overlong_2byte = b"\xc0\x80" # Overlong encoding of NULL + # 2-byte invalid tests + invalid_2byte = b"\xc2\x00" # Invalid continuation (lines 477-480) + result = invalid_2byte.decode("utf-8", errors="replace") + assert "\ufffd" in result, "Invalid 2-byte should produce replacement char" + + overlong_2byte = b"\xc0\x80" # Overlong encoding (lines 486-487) result = overlong_2byte.decode("utf-8", errors="replace") assert "\ufffd" in result, "Overlong 2-byte should produce replacement char" - def test_3byte_decode_paths(self): - """Test 3-byte sequence decoding paths (lines 490-506).""" + def test_3byte_and_4byte_decode_paths(self): + """Test 3-byte and 4-byte sequence decoding paths (lines 490-527).""" from mssql_python.type import Binary - # Test invalid continuation bytes (lines 492-495) - invalid_3byte = [ - b"\xe0\x00\x80", # Second byte invalid - b"\xe0\xa0\x00", # Third byte invalid - ] - - for test_bytes in invalid_3byte: - result = test_bytes.decode("utf-8", errors="replace") - assert ( - "\ufffd" in result - ), f"Invalid 3-byte {test_bytes.hex()} should produce replacement" - - # Test valid decode with surrogate rejection (lines 499-502) - # Valid characters outside surrogate range + # 3-byte valid decode tests (lines 499-502) valid_3byte = [ (b"\xe0\xa0\x80", "\u0800"), (b"\xe4\xb8\xad", "\u4e2d"), # 中 @@ -362,59 +212,48 @@ def test_3byte_decode_paths(self): encoded = Binary(expected) assert encoded == utf8_bytes - # Test surrogate encoding rejection (lines 500-503) - surrogate_3byte = [ - b"\xed\xa0\x80", # U+D800 (high surrogate) - b"\xed\xbf\xbf", # U+DFFF (low surrogate) + # 4-byte valid decode tests (lines 519-522) + valid_4byte = [ + (b"\xf0\x90\x80\x80", "\U00010000"), + (b"\xf0\x9f\x98\x80", "\U0001f600"), # 😀 + (b"\xf4\x8f\xbf\xbf", "\U0010ffff"), ] - for test_bytes in surrogate_3byte: - result = test_bytes.decode("utf-8", errors="replace") - # Should be rejected/replaced - assert len(result) > 0 - - # Test overlong encoding rejection (lines 504-505) - overlong_3byte = b"\xe0\x80\x80" # Overlong encoding of NULL - result = overlong_3byte.decode("utf-8", errors="replace") - assert "\ufffd" in result, "Overlong 3-byte should produce replacement" - - def test_4byte_decode_paths(self): - """Test 4-byte sequence decoding paths (lines 508-527).""" - from mssql_python.type import Binary + for utf8_bytes, expected in valid_4byte: + result = utf8_bytes.decode("utf-8") + assert result == expected + encoded = Binary(expected) + assert encoded == utf8_bytes - # Test invalid continuation bytes (lines 512-514) - invalid_4byte = [ + # Invalid continuation bytes tests + invalid_tests = [ + # 3-byte invalid (lines 492-495) + b"\xe0\x00\x80", # Second byte invalid + b"\xe0\xa0\x00", # Third byte invalid + # 4-byte invalid (lines 512-514) b"\xf0\x00\x80\x80", # Second byte invalid b"\xf0\x90\x00\x80", # Third byte invalid b"\xf0\x90\x80\x00", # Fourth byte invalid ] - for test_bytes in invalid_4byte: + for test_bytes in invalid_tests: result = test_bytes.decode("utf-8", errors="replace") assert ( "\ufffd" in result - ), f"Invalid 4-byte {test_bytes.hex()} should produce replacement" - - # Test valid decode within range (lines 519-522) - valid_4byte = [ - (b"\xf0\x90\x80\x80", "\U00010000"), - (b"\xf0\x9f\x98\x80", "\U0001f600"), # 😀 - (b"\xf4\x8f\xbf\xbf", "\U0010ffff"), - ] + ), f"Invalid sequence {test_bytes.hex()} should produce replacement" - for utf8_bytes, expected in valid_4byte: - result = utf8_bytes.decode("utf-8") - assert result == expected - encoded = Binary(expected) - assert encoded == utf8_bytes + # Surrogate encoding rejection (lines 500-503) + for test_bytes in [b"\xed\xa0\x80", b"\xed\xbf\xbf"]: + result = test_bytes.decode("utf-8", errors="replace") + assert len(result) > 0 - # Test overlong encoding rejection (lines 524-525) - overlong_4byte = b"\xf0\x80\x80\x80" # Overlong encoding of NULL - result = overlong_4byte.decode("utf-8", errors="replace") - assert "\ufffd" in result, "Overlong 4-byte should produce replacement" + # Overlong encoding rejection (lines 504-505, 524-525) + for test_bytes in [b"\xe0\x80\x80", b"\xf0\x80\x80\x80"]: + result = test_bytes.decode("utf-8", errors="replace") + assert "\ufffd" in result, f"Overlong {test_bytes.hex()} should produce replacement" - # Test out-of-range rejection (lines 524-525) - out_of_range = b"\xf4\x90\x80\x80" # 0x110000 (beyond max Unicode) + # Out-of-range rejection (lines 524-525) + out_of_range = b"\xf4\x90\x80\x80" # 0x110000 result = out_of_range.decode("utf-8", errors="replace") assert len(result) > 0, "Out-of-range 4-byte should produce some output" @@ -462,61 +301,42 @@ def test_always_push_result(self): class TestEdgeCases: """Test edge cases and error paths.""" - def test_empty_string(self): - """Test empty string handling.""" - from mssql_python.type import Binary - - empty = "" - result = Binary(empty) - assert result == b"", "Empty string should produce empty bytes" - - def test_null_character(self): - """Test NULL character handling.""" - from mssql_python.type import Binary - - null_str = "\x00" - result = Binary(null_str) - assert result == b"\x00", "NULL character should be preserved" - - # NULL in middle of string - with_null = "A\x00B" - result = Binary(with_null) - assert result == b"A\x00B", "NULL in middle should be preserved" - - def test_very_long_strings(self): - """Test very long strings to ensure no buffer issues.""" - from mssql_python.type import Binary - - # Long ASCII - long_ascii = "A" * 10000 - result = Binary(long_ascii) - assert len(result) == 10000, "Long ASCII string should encode correctly" - - # Long multi-byte - long_utf8 = "中" * 5000 # 3 bytes each - result = Binary(long_utf8) - assert len(result) == 15000, "Long UTF-8 string should encode correctly" - - # Long emoji - long_emoji = "😀" * 2000 # 4 bytes each - result = Binary(long_emoji) - assert len(result) == 8000, "Long emoji string should encode correctly" - - def test_mixed_valid_invalid(self): - """Test strings with mix of valid and invalid sequences.""" + @pytest.mark.parametrize( + "test_input,expected,description", + [ + ("", b"", "Empty string"), + ("\x00", b"\x00", "NULL character"), + ("A\x00B", b"A\x00B", "NULL in middle"), + ("Valid\ufffdText", "Valid\ufffdText", "Mixed valid/U+FFFD"), + ("A\u00a9\u4e2d\U0001f600", "A\u00a9\u4e2d\U0001f600", "All UTF-8 ranges"), + ], + ) + def test_special_characters(self, test_input, expected, description): + """Test special character handling including NULL and replacement chars.""" from mssql_python.type import Binary - # Valid text with legitimate U+FFFD - mixed = "Valid\ufffdText" - result = Binary(mixed) - decoded = result.decode("utf-8") - assert decoded == mixed, "Mixed valid/U+FFFD should work" - - def test_all_utf8_ranges(self): - """Test characters from all UTF-8 ranges in one string.""" + result = Binary(test_input) + if isinstance(expected, str): + # For strings, encode and compare + assert result == expected.encode("utf-8"), f"{description} should work" + # Verify round-trip + decoded = result.decode("utf-8") + assert decoded == test_input + else: + assert result == expected, f"{description} should produce expected bytes" + + @pytest.mark.parametrize( + "char,count,expected_len", + [ + ("A", 1000, 1000), # 1-byte chars - reduced from 10000 for speed + ("中", 500, 1500), # 3-byte chars - reduced from 5000 for speed + ("😀", 200, 800), # 4-byte chars - reduced from 2000 for speed + ], + ) + def test_long_strings(self, char, count, expected_len): + """Test long strings with reduced size for faster execution.""" from mssql_python.type import Binary - all_ranges = "A\u00a9\u4e2d\U0001f600" # 1, 2, 3, 4 byte chars - result = Binary(all_ranges) - decoded = result.decode("utf-8") - assert decoded == all_ranges, "All UTF-8 ranges should work together" + long_str = char * count + result = Binary(long_str) + assert len(result) == expected_len, f"Long {char!r} string should encode correctly"