From a6b367a1650f4241e28cf57f8934c26ea5eff44e Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Tue, 24 Feb 2026 06:39:54 +0000 Subject: [PATCH 1/6] base fix --- mssql_python/cursor.py | 9 ++---- tests/test_019_bulkcopy.py | 66 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 7 deletions(-) diff --git a/mssql_python/cursor.py b/mssql_python/cursor.py index 3acbffcc..935cf269 100644 --- a/mssql_python/cursor.py +++ b/mssql_python/cursor.py @@ -2624,15 +2624,10 @@ def _bulkcopy( parser = _ConnectionStringParser(validate_keywords=False) params = parser._parse(self.connection.connection_str) - if not params.get("server"): + # Check for server parameter (accepts synonyms: server, addr, address) + if not (params.get("server") or params.get("addr") or params.get("address")): raise ValueError("SERVER parameter is required in connection string") - if not params.get("database"): - raise ValueError( - "DATABASE parameter is required in connection string for bulk copy. " - "Specify the target database explicitly to avoid accidentally writing to system databases." - ) - # Translate parsed connection string into the dict py-core expects. pycore_context = connstr_to_pycore_params(params) diff --git a/tests/test_019_bulkcopy.py b/tests/test_019_bulkcopy.py index a50e5d6c..dcbc3962 100644 --- a/tests/test_019_bulkcopy.py +++ b/tests/test_019_bulkcopy.py @@ -82,3 +82,69 @@ def test_bulkcopy_basic(cursor): # Cleanup cursor.execute(f"DROP TABLE {table_name}") + + +def test_bulkcopy_without_database_parameter(conn_str): + """Test bulkcopy operation works when DATABASE is not specified in connection string. + + The database keyword in connection string is optional. In its absence, + the client sends an empty database name and the server responds with + the default database the client was connected to. + """ + import re + from mssql_python import connect + + # Remove DATABASE parameter from connection string if present + # Handle various formats: Database=...; database=...; DATABASE=...; + conn_str_no_db = re.sub( + r";?\s*database\s*=\s*[^;]+(;|$)", r"\1", conn_str, flags=re.IGNORECASE + ).strip() + # Clean up any double semicolons + conn_str_no_db = re.sub(r";+", ";", conn_str_no_db) + # Remove leading/trailing semicolons + conn_str_no_db = re.sub(r"^;+|;+$", "", conn_str_no_db) + + # Create connection without DATABASE parameter + conn = connect(conn_str_no_db) + try: + cursor = conn.cursor() + + # Verify we're connected to a database (should be the default) + cursor.execute("SELECT DB_NAME() AS current_db") + current_db = cursor.fetchone()[0] + assert current_db is not None, "Should be connected to a database" + + # Create test table in the current database + table_name = "mssql_python_bulkcopy_no_db_test" + cursor.execute(f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name}") + cursor.execute(f"CREATE TABLE {table_name} (id INT, name VARCHAR(50), value FLOAT)") + conn.commit() + + # Prepare test data + data = [ + (1, "Alice", 100.5), + (2, "Bob", 200.75), + (3, "Charlie", 300.25), + ] + + # Perform bulkcopy - this should NOT raise ValueError about missing DATABASE + result = cursor._bulkcopy(table_name, data, timeout=60) + + # Verify result + assert result is not None + assert result["rows_copied"] == 3 + + # Verify data was inserted correctly + cursor.execute(f"SELECT id, name, value FROM {table_name} ORDER BY id") + rows = cursor.fetchall() + + assert len(rows) == 3 + assert rows[0][0] == 1 and rows[0][1] == "Alice" and abs(rows[0][2] - 100.5) < 0.01 + assert rows[1][0] == 2 and rows[1][1] == "Bob" and abs(rows[1][2] - 200.75) < 0.01 + assert rows[2][0] == 3 and rows[2][1] == "Charlie" and abs(rows[2][2] - 300.25) < 0.01 + + # Cleanup + cursor.execute(f"DROP TABLE {table_name}") + cursor.close() + finally: + conn.close() From 0e6be2aecb0c2e90d8d783e897c81ac581ab8be3 Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Tue, 24 Feb 2026 07:11:00 +0000 Subject: [PATCH 2/6] some more test for server parameter --- tests/test_019_bulkcopy.py | 109 +++++++++++++++++++++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/tests/test_019_bulkcopy.py b/tests/test_019_bulkcopy.py index dcbc3962..b48b320a 100644 --- a/tests/test_019_bulkcopy.py +++ b/tests/test_019_bulkcopy.py @@ -148,3 +148,112 @@ def test_bulkcopy_without_database_parameter(conn_str): cursor.close() finally: conn.close() + + +def test_bulkcopy_with_server_synonyms(conn_str): + """Test that bulkcopy works with all SERVER parameter synonyms: server, addr, address.""" + import re + from mssql_python import connect + + # Test with 'Addr' synonym + conn_string_addr = re.sub(r"(?i)\bserver\s*=", "Addr=", conn_str, count=1) + conn = connect(conn_string_addr) + try: + cursor = conn.cursor() + table_name = "test_bulkcopy_addr_synonym" + + # Create table + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + cursor.execute(f""" + CREATE TABLE {table_name} ( + id INT, + name NVARCHAR(50), + value FLOAT + ) + """) + conn.commit() + + # Test data + test_data = [(1, "Test1", 1.5), (2, "Test2", 2.5), (3, "Test3", 3.5)] + + # Perform bulkcopy with connection using Addr parameter + result = cursor._bulkcopy(table_name, test_data) + + # Verify result + assert result is not None + assert "rows_copied" in result + assert result["rows_copied"] == 3 + + # Verify data + cursor.execute(f"SELECT COUNT(*) FROM {table_name}") + count = cursor.fetchone()[0] + assert count == 3 + + # Cleanup + cursor.execute(f"DROP TABLE {table_name}") + cursor.close() + finally: + conn.close() + + # Test with 'Address' synonym + conn_string_address = re.sub(r"(?i)\bserver\s*=", "Address=", conn_str, count=1) + conn = connect(conn_string_address) + try: + cursor = conn.cursor() + table_name = "test_bulkcopy_address_synonym" + + # Create table + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + cursor.execute(f""" + CREATE TABLE {table_name} ( + id INT, + name NVARCHAR(50), + value FLOAT + ) + """) + conn.commit() + + # Test data + test_data = [(1, "Test1", 1.5), (2, "Test2", 2.5), (3, "Test3", 3.5)] + + # Perform bulkcopy with connection using Address parameter + result = cursor._bulkcopy(table_name, test_data) + + # Verify result + assert result is not None + assert "rows_copied" in result + assert result["rows_copied"] == 3 + + # Verify data + cursor.execute(f"SELECT COUNT(*) FROM {table_name}") + count = cursor.fetchone()[0] + assert count == 3 + + # Cleanup + cursor.execute(f"DROP TABLE {table_name}") + cursor.close() + finally: + conn.close() + + # Test that bulkcopy fails when SERVER parameter is missing entirely + conn_string_no_server = re.sub(r"(?i);?\s*(server|addr|address)\s*=\s*[^;]+;?", ";", conn_str) + # Ensure we have a valid connection string for the main connection + conn = connect(conn_str) + try: + cursor = conn.cursor() + # Manually override the connection string to one without server + cursor.connection.connection_str = conn_string_no_server + + table_name = "test_bulkcopy_no_server" + test_data = [(1, "Test1", 1.5)] + + # This should raise ValueError due to missing SERVER parameter + try: + cursor._bulkcopy(table_name, test_data) + assert False, "Expected ValueError for missing SERVER parameter" + except ValueError as e: + assert "SERVER parameter is required" in str(e) + + cursor.close() + finally: + conn.close() From cdc2a5c36f4b3a2703ed87a603547784f6ea5c7e Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Tue, 24 Feb 2026 07:27:51 +0000 Subject: [PATCH 3/6] copilot review fix --- tests/test_019_bulkcopy.py | 59 +++++++++++++++++++++++++++----------- 1 file changed, 43 insertions(+), 16 deletions(-) diff --git a/tests/test_019_bulkcopy.py b/tests/test_019_bulkcopy.py index b48b320a..57ca5002 100644 --- a/tests/test_019_bulkcopy.py +++ b/tests/test_019_bulkcopy.py @@ -91,18 +91,20 @@ def test_bulkcopy_without_database_parameter(conn_str): the client sends an empty database name and the server responds with the default database the client was connected to. """ - import re from mssql_python import connect + from mssql_python.connection_string_parser import _ConnectionStringParser + from mssql_python.connection_string_builder import _ConnectionStringBuilder - # Remove DATABASE parameter from connection string if present - # Handle various formats: Database=...; database=...; DATABASE=...; - conn_str_no_db = re.sub( - r";?\s*database\s*=\s*[^;]+(;|$)", r"\1", conn_str, flags=re.IGNORECASE - ).strip() - # Clean up any double semicolons - conn_str_no_db = re.sub(r";+", ";", conn_str_no_db) - # Remove leading/trailing semicolons - conn_str_no_db = re.sub(r"^;+|;+$", "", conn_str_no_db) + # Parse the connection string using the proper parser + parser = _ConnectionStringParser(validate_keywords=False) + params = parser._parse(conn_str) + + # Remove DATABASE parameter if present (case-insensitive, handles all synonyms) + params.pop("database", None) + + # Rebuild the connection string using the builder to preserve braced values + builder = _ConnectionStringBuilder(params) + conn_str_no_db = builder.build() # Create connection without DATABASE parameter conn = connect(conn_str_no_db) @@ -152,11 +154,22 @@ def test_bulkcopy_without_database_parameter(conn_str): def test_bulkcopy_with_server_synonyms(conn_str): """Test that bulkcopy works with all SERVER parameter synonyms: server, addr, address.""" - import re from mssql_python import connect + from mssql_python.connection_string_parser import _ConnectionStringParser + from mssql_python.connection_string_builder import _ConnectionStringBuilder + + # Parse the connection string using the proper parser + parser = _ConnectionStringParser(validate_keywords=False) + params = parser._parse(conn_str) + + # Test with 'Addr' synonym - replace 'server' with 'addr' + server_value = ( + params.pop("server", None) or params.pop("addr", None) or params.pop("address", None) + ) + params["addr"] = server_value + builder = _ConnectionStringBuilder(params) + conn_string_addr = builder.build() - # Test with 'Addr' synonym - conn_string_addr = re.sub(r"(?i)\bserver\s*=", "Addr=", conn_str, count=1) conn = connect(conn_string_addr) try: cursor = conn.cursor() @@ -195,8 +208,15 @@ def test_bulkcopy_with_server_synonyms(conn_str): finally: conn.close() - # Test with 'Address' synonym - conn_string_address = re.sub(r"(?i)\bserver\s*=", "Address=", conn_str, count=1) + # Test with 'Address' synonym - replace with 'address' + params = parser._parse(conn_str) + server_value = ( + params.pop("server", None) or params.pop("addr", None) or params.pop("address", None) + ) + params["address"] = server_value + builder = _ConnectionStringBuilder(params) + conn_string_address = builder.build() + conn = connect(conn_string_address) try: cursor = conn.cursor() @@ -236,7 +256,14 @@ def test_bulkcopy_with_server_synonyms(conn_str): conn.close() # Test that bulkcopy fails when SERVER parameter is missing entirely - conn_string_no_server = re.sub(r"(?i);?\s*(server|addr|address)\s*=\s*[^;]+;?", ";", conn_str) + params = parser._parse(conn_str) + # Remove all server synonyms + params.pop("server", None) + params.pop("addr", None) + params.pop("address", None) + builder = _ConnectionStringBuilder(params) + conn_string_no_server = builder.build() + # Ensure we have a valid connection string for the main connection conn = connect(conn_str) try: From edbb65a68f709bb6c4867cbc2a95c8afb349b0bb Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Tue, 24 Feb 2026 09:14:33 +0000 Subject: [PATCH 4/6] CI/CD build failure fix --- tests/test_019_bulkcopy.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/test_019_bulkcopy.py b/tests/test_019_bulkcopy.py index 57ca5002..d1e899c6 100644 --- a/tests/test_019_bulkcopy.py +++ b/tests/test_019_bulkcopy.py @@ -99,6 +99,9 @@ def test_bulkcopy_without_database_parameter(conn_str): parser = _ConnectionStringParser(validate_keywords=False) params = parser._parse(conn_str) + # Save the original database name to use it explicitly in our operations + original_database = params.get("database") + # Remove DATABASE parameter if present (case-insensitive, handles all synonyms) params.pop("database", None) @@ -116,6 +119,10 @@ def test_bulkcopy_without_database_parameter(conn_str): current_db = cursor.fetchone()[0] assert current_db is not None, "Should be connected to a database" + # If original database was specified, switch to it to ensure we have permissions + if original_database: + cursor.execute(f"USE [{original_database}]") + # Create test table in the current database table_name = "mssql_python_bulkcopy_no_db_test" cursor.execute(f"IF OBJECT_ID('{table_name}', 'U') IS NOT NULL DROP TABLE {table_name}") From a1ff8c597aa0929f1fdbb4a4e2b2e14d77951efa Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Tue, 24 Feb 2026 10:38:56 +0000 Subject: [PATCH 5/6] test failure fix --- tests/test_019_bulkcopy.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/tests/test_019_bulkcopy.py b/tests/test_019_bulkcopy.py index d1e899c6..e9ab9662 100644 --- a/tests/test_019_bulkcopy.py +++ b/tests/test_019_bulkcopy.py @@ -137,7 +137,12 @@ def test_bulkcopy_without_database_parameter(conn_str): ] # Perform bulkcopy - this should NOT raise ValueError about missing DATABASE - result = cursor._bulkcopy(table_name, data, timeout=60) + # Note: bulkcopy creates its own connection, so we need to use fully qualified table name + # if we had a database in the original connection string + bulkcopy_table_name = ( + f"[{original_database}].[dbo].{table_name}" if original_database else table_name + ) + result = cursor._bulkcopy(bulkcopy_table_name, data, timeout=60) # Verify result assert result is not None From ff44427c1344aa76c65a824024470bf3ea67f7fa Mon Sep 17 00:00:00 2001 From: Subrata Paitandi Date: Wed, 25 Feb 2026 09:11:08 +0000 Subject: [PATCH 6/6] fixing test due to bulkcopy API name change --- tests/test_019_bulkcopy.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/tests/test_019_bulkcopy.py b/tests/test_019_bulkcopy.py index bb4dc2d3..7542139d 100644 --- a/tests/test_019_bulkcopy.py +++ b/tests/test_019_bulkcopy.py @@ -142,7 +142,7 @@ def test_bulkcopy_without_database_parameter(conn_str): bulkcopy_table_name = ( f"[{original_database}].[dbo].{table_name}" if original_database else table_name ) - result = cursor._bulkcopy(bulkcopy_table_name, data, timeout=60) + result = cursor.bulkcopy(bulkcopy_table_name, data, timeout=60) # Verify result assert result is not None @@ -202,7 +202,7 @@ def test_bulkcopy_with_server_synonyms(conn_str): test_data = [(1, "Test1", 1.5), (2, "Test2", 2.5), (3, "Test3", 3.5)] # Perform bulkcopy with connection using Addr parameter - result = cursor._bulkcopy(table_name, test_data) + result = cursor.bulkcopy(table_name, test_data) # Verify result assert result is not None @@ -249,7 +249,7 @@ def test_bulkcopy_with_server_synonyms(conn_str): test_data = [(1, "Test1", 1.5), (2, "Test2", 2.5), (3, "Test3", 3.5)] # Perform bulkcopy with connection using Address parameter - result = cursor._bulkcopy(table_name, test_data) + result = cursor.bulkcopy(table_name, test_data) # Verify result assert result is not None @@ -288,7 +288,7 @@ def test_bulkcopy_with_server_synonyms(conn_str): # This should raise ValueError due to missing SERVER parameter try: - cursor._bulkcopy(table_name, test_data) + cursor.bulkcopy(table_name, test_data) assert False, "Expected ValueError for missing SERVER parameter" except ValueError as e: assert "SERVER parameter is required" in str(e)