diff --git a/mssql_python/connection.py b/mssql_python/connection.py index 2a7af146..cc1701fb 100644 --- a/mssql_python/connection.py +++ b/mssql_python/connection.py @@ -18,6 +18,7 @@ from mssql_python.pooling import PoolingManager from mssql_python.exceptions import InterfaceError from mssql_python.auth import process_connection_string +from mssql_python.constants import GetInfoConstants class Connection: @@ -159,6 +160,30 @@ def setautocommit(self, value: bool = False) -> None: """ self._conn.set_autocommit(value) + @property + def searchescape(self): + """ + The ODBC search pattern escape character, as returned by + SQLGetInfo(SQL_SEARCH_PATTERN_ESCAPE), used to escape special characters + such as '%' and '_' in LIKE clauses. These are driver specific. + + Returns: + str: The search pattern escape character (usually '\' or another character) + """ + if not hasattr(self, '_searchescape'): + try: + escape_char = self.getinfo(GetInfoConstants.SQL_SEARCH_PATTERN_ESCAPE.value) + # Some drivers might return this as an integer memory address + # or other non-string format, so ensure we have a string + if not isinstance(escape_char, str): + escape_char = '\\' # Default to backslash if not a string + self._searchescape = escape_char + except Exception as e: + # Log the exception for debugging, but do not expose sensitive info + log('warning', f"Failed to retrieve search escape character, using default '\\'. Exception: {type(e).__name__}") + self._searchescape = '\\' + return self._searchescape + def cursor(self) -> Cursor: """ Return a new Cursor object using the connection. diff --git a/tests/test_003_connection.py b/tests/test_003_connection.py index 8ddb33ef..4937f685 100644 --- a/tests/test_003_connection.py +++ b/tests/test_003_connection.py @@ -698,4 +698,167 @@ def test_getinfo_standard_types(db_connection): except Exception as e: # Log but don't fail - some drivers might not support all info types - print(f"Info type {info_type} failed: {e}") \ No newline at end of file + print(f"Info type {info_type} failed: {e}") + +def test_connection_searchescape_basic(db_connection): + """Test the basic functionality of the searchescape property.""" + # Get the search escape character + escape_char = db_connection.searchescape + + # Verify it's not None + assert escape_char is not None, "Search escape character should not be None" + print(f"Search pattern escape character: '{escape_char}'") + + # Test property caching - calling it twice should return the same value + escape_char2 = db_connection.searchescape + assert escape_char == escape_char2, "Search escape character should be consistent" + +def test_connection_searchescape_with_percent(db_connection): + """Test using the searchescape property with percent wildcard.""" + escape_char = db_connection.searchescape + + # Skip test if we got a non-string or empty escape character + if not isinstance(escape_char, str) or not escape_char: + pytest.skip("No valid escape character available for testing") + + cursor = db_connection.cursor() + try: + # Create a temporary table with data containing % character + cursor.execute("CREATE TABLE #test_escape_percent (id INT, text VARCHAR(50))") + cursor.execute("INSERT INTO #test_escape_percent VALUES (1, 'abc%def')") + cursor.execute("INSERT INTO #test_escape_percent VALUES (2, 'abc_def')") + cursor.execute("INSERT INTO #test_escape_percent VALUES (3, 'abcdef')") + + # Use the escape character to find the exact % character + query = f"SELECT * FROM #test_escape_percent WHERE text LIKE 'abc{escape_char}%def' ESCAPE '{escape_char}'" + cursor.execute(query) + results = cursor.fetchall() + + # Should match only the row with the % character + assert len(results) == 1, f"Escaped LIKE query for % matched {len(results)} rows instead of 1" + if results: + assert 'abc%def' in results[0][1], "Escaped LIKE query did not match correct row" + + except Exception as e: + print(f"Note: LIKE escape test with % failed: {e}") + # Don't fail the test as some drivers might handle escaping differently + finally: + cursor.execute("DROP TABLE #test_escape_percent") + +def test_connection_searchescape_with_underscore(db_connection): + """Test using the searchescape property with underscore wildcard.""" + escape_char = db_connection.searchescape + + # Skip test if we got a non-string or empty escape character + if not isinstance(escape_char, str) or not escape_char: + pytest.skip("No valid escape character available for testing") + + cursor = db_connection.cursor() + try: + # Create a temporary table with data containing _ character + cursor.execute("CREATE TABLE #test_escape_underscore (id INT, text VARCHAR(50))") + cursor.execute("INSERT INTO #test_escape_underscore VALUES (1, 'abc_def')") + cursor.execute("INSERT INTO #test_escape_underscore VALUES (2, 'abcXdef')") # 'X' could match '_' + cursor.execute("INSERT INTO #test_escape_underscore VALUES (3, 'abcdef')") # No match + + # Use the escape character to find the exact _ character + query = f"SELECT * FROM #test_escape_underscore WHERE text LIKE 'abc{escape_char}_def' ESCAPE '{escape_char}'" + cursor.execute(query) + results = cursor.fetchall() + + # Should match only the row with the _ character + assert len(results) == 1, f"Escaped LIKE query for _ matched {len(results)} rows instead of 1" + if results: + assert 'abc_def' in results[0][1], "Escaped LIKE query did not match correct row" + + except Exception as e: + print(f"Note: LIKE escape test with _ failed: {e}") + # Don't fail the test as some drivers might handle escaping differently + finally: + cursor.execute("DROP TABLE #test_escape_underscore") + +def test_connection_searchescape_with_brackets(db_connection): + """Test using the searchescape property with bracket wildcards.""" + escape_char = db_connection.searchescape + + # Skip test if we got a non-string or empty escape character + if not isinstance(escape_char, str) or not escape_char: + pytest.skip("No valid escape character available for testing") + + cursor = db_connection.cursor() + try: + # Create a temporary table with data containing [ character + cursor.execute("CREATE TABLE #test_escape_brackets (id INT, text VARCHAR(50))") + cursor.execute("INSERT INTO #test_escape_brackets VALUES (1, 'abc[x]def')") + cursor.execute("INSERT INTO #test_escape_brackets VALUES (2, 'abcxdef')") + + # Use the escape character to find the exact [ character + # Note: This might not work on all drivers as bracket escaping varies + query = f"SELECT * FROM #test_escape_brackets WHERE text LIKE 'abc{escape_char}[x{escape_char}]def' ESCAPE '{escape_char}'" + cursor.execute(query) + results = cursor.fetchall() + + # Just check we got some kind of result without asserting specific behavior + print(f"Bracket escaping test returned {len(results)} rows") + + except Exception as e: + print(f"Note: LIKE escape test with brackets failed: {e}") + # Don't fail the test as bracket escaping varies significantly between drivers + finally: + cursor.execute("DROP TABLE #test_escape_brackets") + +def test_connection_searchescape_multiple_escapes(db_connection): + """Test using the searchescape property with multiple escape sequences.""" + escape_char = db_connection.searchescape + + # Skip test if we got a non-string or empty escape character + if not isinstance(escape_char, str) or not escape_char: + pytest.skip("No valid escape character available for testing") + + cursor = db_connection.cursor() + try: + # Create a temporary table with data containing multiple special chars + cursor.execute("CREATE TABLE #test_multiple_escapes (id INT, text VARCHAR(50))") + cursor.execute("INSERT INTO #test_multiple_escapes VALUES (1, 'abc%def_ghi')") + cursor.execute("INSERT INTO #test_multiple_escapes VALUES (2, 'abc%defXghi')") # Wouldn't match the pattern + cursor.execute("INSERT INTO #test_multiple_escapes VALUES (3, 'abcXdef_ghi')") # Wouldn't match the pattern + + # Use escape character for both % and _ + query = f""" + SELECT * FROM #test_multiple_escapes + WHERE text LIKE 'abc{escape_char}%def{escape_char}_ghi' ESCAPE '{escape_char}' + """ + cursor.execute(query) + results = cursor.fetchall() + + # Should match only the row with both % and _ + assert len(results) <= 1, f"Multiple escapes query matched {len(results)} rows instead of at most 1" + if len(results) == 1: + assert 'abc%def_ghi' in results[0][1], "Multiple escapes query matched incorrect row" + + except Exception as e: + print(f"Note: Multiple escapes test failed: {e}") + # Don't fail the test as escaping behavior varies + finally: + cursor.execute("DROP TABLE #test_multiple_escapes") + +def test_connection_searchescape_consistency(db_connection): + """Test that the searchescape property is cached and consistent.""" + # Call the property multiple times + escape1 = db_connection.searchescape + escape2 = db_connection.searchescape + escape3 = db_connection.searchescape + + # All calls should return the same value + assert escape1 == escape2 == escape3, "Searchescape property should be consistent" + + # Create a new connection and verify it returns the same escape character + # (assuming the same driver and connection settings) + if 'conn_str' in globals(): + try: + new_conn = connect(conn_str) + new_escape = new_conn.searchescape + assert new_escape == escape1, "Searchescape should be consistent across connections" + new_conn.close() + except Exception as e: + print(f"Note: New connection comparison failed: {e}") \ No newline at end of file