Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .github/workflows/devskim.yml
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,10 @@ jobs:

- name: Run DevSkim scanner
uses: microsoft/DevSkim-Action@v1
with:
ignore_globs: "tests/**,**/test_*.py,benchmarks/**"
# DS176209: TODO comments — not a security concern
Comment thread
bewithgaurav marked this conversation as resolved.
Dismissed
ignore_rules_list: "DS176209"

- name: Upload DevSkim scan results to GitHub Security tab
uses: github/codeql-action/upload-sarif@v3
Expand Down
41 changes: 25 additions & 16 deletions tests/test_019_bulkcopy.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,9 +99,6 @@ 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)

Expand All @@ -119,15 +116,32 @@ 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}]")
# Skip on Azure SQL Database (EngineEdition 5): bulkcopy internally
# opens a second connection via mssql_py_core using the stored
# connection string. Without a DATABASE keyword that second
# connection cannot resolve the target table on Azure SQL.
cursor.execute("SELECT CAST(SERVERPROPERTY('EngineEdition') AS INT)")
engine_edition = cursor.fetchone()[0]
if engine_edition == 5:
pytest.skip(
"bulkcopy uses a separate internal connection; without "
"DATABASE in the connection string it cannot resolve "
"table metadata on Azure SQL"
)

# Create test table in the current database
# Use unqualified table names — the table is created in whatever
# database we connected to. Three-part names ([db].[schema].[table])
# are NOT supported on Azure SQL, and the default database (often
# master) may deny CREATE TABLE on other CI environments, so we
# skip gracefully when the current database doesn't allow DDL.
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()

try:
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()
except Exception as e:
pytest.skip(f"Cannot create table in default database '{current_db}': {e}")

# Prepare test data
data = [
Expand All @@ -137,12 +151,7 @@ def test_bulkcopy_without_database_parameter(conn_str):
]

# Perform bulkcopy - this should NOT raise ValueError about missing DATABASE
# 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)
result = cursor.bulkcopy(table_name, data, timeout=60)

# Verify result
assert result is not None
Expand Down
Loading