`cursor.bulkcopy()` fails,RuntimeError: Failed to connect to SQL Server: Protocol Error: Failed to receive token during login response parsing

-- coding:utf-8 --

"""
Test script for cursor.bulkcopy() failure

This script isolates the bulkcopy() issue without GUI interference.
Result: bulkcopy() ALWAYS fails with:
"RuntimeError: Failed to connect to SQL Server: Protocol Error: Failed to receive token during login response parsing"
while the normal ODBC connection via mssql-python.connect() works perfectly fine.

Environment:

• Python: 3.11.x
• MSSQL-Python: 1.7.1 (latest pip version, includes PR #439 which passes all connection string params to py-core)
• MSSQL-Py-Core: bundled with mssql-python (Rust .pyd binary, cannot be upgraded separately from pip)
• ODBC Driver: ODBC Driver 18 for SQL Server (auto-managed by mssql-python, no manual DSN setup needed)
• SQL Server: Microsoft SQL Server 2016 (13.x)
• Target table: temp_TEST_BATCHCOPY (56 columns, varchar/nvarchar only)
• Auth: SQL Login (UID/PWD), with TrustServerCertificate=yes
• OS: Windows 11
  """

import sys
import platform
import mssql_python

============================================================

0. Environment info

============================================================

print("=" * 60)
print("Environment")
print("=" * 60)
print(f"Python version: {sys.version}")
print(f"Platform: {platform.platform()}")
print(f"mssql-python version: {mssql_python.version}")

--- CONFIGURE THESE FOR YOUR ENVIRONMENT ---

CONN_STR = 'SERVER=192.168.0.18;DATABASE=AAB;UID=fa_app;PWD=qazwsx!@;TrustServerCertificate=yes;'
TABLE = 'temp_TEST_BATCHCOPY'

--------------------------------------------

try:
conn = mssql_python.connect(CONN_STR)
c = conn.cursor()
c.execute("SELECT @@VERSION")
sql_ver = c.fetchone()[0]
print(f"SQL Server: {sql_ver.split(chr(10))[0].strip()}")
conn.close()
except Exception as e:
print(f"Cannot get SQL Server version (will continue): {e}")

============================================================

1. Normal connection + metadata query (always works)

============================================================

print()
print("=" * 60)
print("1. Normal connection + metadata query")
print("=" * 60)

conn = mssql_python.connect(CONN_STR)
c = conn.cursor()

c.execute("IF OBJECT_ID(N'temp_TEST_BATCHCOPY', N'U') IS NULL SELECT 0 ELSE SELECT 1")
exists = c.fetchone()[0]
print(f"Table exists: {exists}")

c.execute("""SELECT COLUMN_NAME FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_NAME = ? ORDER BY ORDINAL_POSITION""", (TABLE,))
table_columns = [r[0] for r in c.fetchall()]
print(f"Number of columns: {len(table_columns)}")
print(f"Column names (first 5): {table_columns[:5]}...")

conn.close()

============================================================

2. Test cursor.bulkcopy()

============================================================

print()
print("=" * 60)
print("2. Test cursor.bulkcopy()")
print("=" * 60)

Build 5 rows of test data matching the table column count

test_data = [tuple(f"test_row{i}_col{j}" for j in range(len(table_columns))) for i in range(5)]
print(f"Test data rows: {len(test_data)}")
print(f"Values per row: {len(test_data[0])}")

fresh_conn = mssql_python.connect(CONN_STR)
fresh_c = fresh_conn.cursor()

try:
result = fresh_c.bulkcopy(
TABLE,
test_data,
batch_size=0,
timeout=30,
column_mappings=table_columns,
)
print(f"bulkcopy() SUCCEEDED: {result}")
except Exception as e:
print(f"bulkcopy() FAILED!")
print(f" Exception type: {type(e).name}")
print(f" Error message: {str(e)}")
finally:
fresh_conn.close()

============================================================

3. Verify if any data was actually inserted

============================================================

print()
print("=" * 60)
print("3. Verify row count after test")
print("=" * 60)

try:
vc = mssql_python.connect(CONN_STR)
cr = vc.cursor()
cr.execute(f"SELECT COUNT(*) FROM [{TABLE}]")
print(f"Table '{TABLE}' row count: {cr.fetchone()[0]}")
vc.close()
except Exception as e:
print(f"Verification failed (may be fine): {e}")

============================================================

4. Connection string variant matrix

============================================================

print()
print("=" * 60)
print("4. Connection string variant matrix")
print("=" * 60)

BASE = 'SERVER=192.168.0.18;DATABASE=AAB;UID=fa_app;PWD=qazwsx!@'
data2 = [tuple(f"v{i}_c{j}" for j in range(len(table_columns))) for i in range(2)]

variants = [
(f'{BASE};TrustServerCertificate=yes;', 'TrustServerCertificate=yes'),
(f'{BASE};Encrypt=yes;TrustServerCertificate=yes;','Encrypt=yes;TrustSC=yes'),
(f'{BASE};Encrypt=no;TrustServerCertificate=yes;', 'Encrypt=no;TrustSC=yes'),
(f'{BASE};Encrypt=no;', 'Encrypt=no (no Trust)'),
(f'{BASE};encrypt=no;TrustServerCertificate=yes;', 'lowercase encrypt=no'),
]

for cs, label in variants:
odbc_ok = False
try:
t = mssql_python.connect(cs)
t.close()
odbc_ok = True
except Exception:
pass

try:
    b = mssql_python.connect(cs)
    bc = b.cursor()
    bc.bulkcopy(TABLE, data2, batch_size=0, timeout=30, column_mappings=table_columns)
    b.close()
    print(f"[{label:35s}] ODBC={'OK ' if odbc_ok else 'ERR'} | BULKCOPY=OK")
except Exception as e:
    err = str(e).replace('\n', ' ')[:80]
    print(f"[{label:35s}] ODBC={'OK ' if odbc_ok else 'ERR'} | BULKCOPY=FAIL: {err}")