test added

microsoft · subrata-ms · Jan 7, 2026 · Jan 7, 2026 · Jan 7, 2026 · Jan 7, 2026
commit 2800534530bf0b02dc69cd42c6bf36db8712534b
diff --git a/main.py b/main.py
@@ -1,18 +1,118 @@
 from mssql_python import connect
 from mssql_python.logging import setup_logging
 import os
+from datetime import datetime
 
 # Clean one-liner: set level and output mode together
 setup_logging(output="both")
 
-conn_str = os.getenv("DB_CONNECTION_STRING")
+print("=" * 70)
+print("SQL Server - Bulk Copy Demo")
+print("=" * 70)
+
+# Use local SQL Server or environment variable
+conn_str = os.getenv("DB_CONNECTION_STRING", 
+                      "Server=localhost,1433;Database=master;UID=sa;PWD=uvFvisUxK4En7AAV;TrustServerCertificate=yes;")
+
+print("\n[1] Connecting to database...")
 conn = connect(conn_str)
 cursor = conn.cursor()
+
+# Query databases
+print("[2] Querying sys.databases...")
 cursor.execute("SELECT database_id, name from sys.databases;")
 rows = cursor.fetchall()
 
 for row in rows:
-    print(f"Database ID: {row[0]}, Name: {row[1]}")
+    print(f"  Database ID: {row[0]}, Name: {row[1]}")
+
+print(f"\n  Total databases: {len(rows)}")
+
+# Demonstrate bulk copy functionality
+print("\n" + "=" * 70)
+print("Bulk Copy with Rust Bindings")
+print("=" * 70)
+
+try:
+    import mssql_rust_bindings as rust
+    import mssql_core_tds
+
+    print("\n[3] Creating temporary table for bulk copy...")
+    cursor.execute("""
+        CREATE TABLE #bulk_copy_demo (
+            id INT,
+            name NVARCHAR(50),
+            value DECIMAL(10, 2),
+            created_date DATETIME
+        )
+    """)
+    conn.commit()
+    print("  ✓ Table created")
+
+    # Generate 100 rows of test data
+    print("\n[4] Generating 100 rows of test data...")
+    test_data = []
+    for i in range(1, 101):
+        test_data.append([
+            i,
+            f"TestItem_{i}",
+            float(i * 10.5),
+            datetime.now()
+        ])
+    print(f"  ✓ Generated {len(test_data)} rows")
+
+    # Create mssql_core_tds connection
+    print("\n[5] Creating mssql_core_tds connection...")
+    conn_dict = {
+        'server': 'localhost',
+        'database': 'master',
+        'user_name': 'sa',
+        'password': 'uvFvisUxK4En7AAV',
+        'trust_server_certificate': 'yes'
+    }
+    core_conn = mssql_core_tds.DdbcConnection(conn_dict)
+    print("  ✓ Connection created")
+
+    # Check if bulk_copy is available
+    if not hasattr(core_conn, 'bulk_copy'):
+        print("\n  ⚠ bulk_copy method not yet implemented in mssql_core_tds")
+        print("  Skipping bulk copy demo")
+    else:
+        # Create BulkCopyWrapper and perform bulk copy
+        print("\n[6] Creating BulkCopyWrapper...")
+        bulk_wrapper = rust.BulkCopyWrapper(core_conn)
+        print("  ✓ Wrapper created")
+
+        print("\n[7] Performing bulk copy...")
+        result = bulk_wrapper.bulk_copy('#bulk_copy_demo', test_data)
+        print(f"  ✓ Bulk copy completed: {result}")
+
+        # Verify the data
+        print("\n[8] Verifying bulk copy results...")
+        cursor.execute("SELECT COUNT(*) FROM #bulk_copy_demo")
+        count = cursor.fetchone()[0]
+        print(f"  ✓ Total rows copied: {count}")
+
+        # Show sample data
+        cursor.execute("SELECT TOP 5 id, name, value FROM #bulk_copy_demo ORDER BY id")
+        sample_rows = cursor.fetchall()
+        print("\n  Sample data:")
+        for row in sample_rows:
+            print(f"    ID: {row[0]}, Name: {row[1]}, Value: {row[2]}")
+
+        core_conn.close()
+        print("\n✓ Bulk copy demo completed successfully!")
+
+    # Cleanup
+    cursor.execute("DROP TABLE IF EXISTS #bulk_copy_demo")
+    conn.commit()
+
+except ImportError as e:
+    print(f"\n✗ Rust bindings or mssql_core_tds not available: {e}")
+except Exception as e:
+    print(f"\n✗ Bulk copy failed: {e}")
 
+print("\n" + "=" * 70)
 cursor.close()
-conn.close()
+conn.close()
+print("✓ Connection closed")
diff --git a/tests/test_016_bulk_copy_rust.py b/tests/test_016_bulk_copy_rust.py
@@ -0,0 +1,89 @@
+"""
+Test bulk copy functionality using Rust bindings with mssql_core_tds.
+
+This test validates the BulkCopyWrapper from mssql_rust_bindings module
+by copying 100 rows of test data into a temporary table using bulk_copy API.
+"""
+
+import pytest
+import mssql_python
+from datetime import datetime
+
+
+def test_bulk_copy_100_rows(db_connection, cursor):
+    """Test bulk copy with 100 rows of data"""
+    try:
+        import mssql_rust_bindings as rust
+        import mssql_core_tds
+    except ImportError as e:
+        pytest.skip(f"Rust bindings or mssql_core_tds not available: {e}")
+
+    # Check if bulk_copy is implemented in mssql_core_tds
+    conn_dict = {
+        'server': 'localhost',
+        'database': 'master',
+        'user_name': 'sa',
+        'password': 'uvFvisUxK4En7AAV',
+        'trust_server_certificate': 'yes'
+    }
+    test_conn = mssql_core_tds.DdbcConnection(conn_dict)
+    if not hasattr(test_conn, 'bulk_copy'):
+        test_conn.close()
+        pytest.skip("bulk_copy method not yet implemented in mssql_core_tds")
+    test_conn.close()
+
+    # Create a temporary test table
+    table_name = "#bulk_copy_test_100"
+    cursor.execute(f"""
+        CREATE TABLE {table_name} (
+            id INT,
+            name NVARCHAR(50),
+            value DECIMAL(10, 2),
+            created_date DATETIME
+        )
+    """)
+    db_connection.commit()
+
+    # Generate 100 rows of test data
+    test_data = []
+    for i in range(1, 101):
+        test_data.append([
+            i,
+            f"TestName_{i}",
+            float(i * 10.5),
+            datetime.now()
+        ])
+
+    # Create mssql_core_tds connection and BulkCopyWrapper
+    try:
+        core_conn = mssql_core_tds.DdbcConnection(conn_dict)
+        bulk_wrapper = rust.BulkCopyWrapper(core_conn)
+
+        # Perform bulk copy
+        result = bulk_wrapper.bulk_copy(table_name, test_data)
+
+        # Verify the copy count
+        cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
+        count = cursor.fetchone()[0]
+
+        assert count == 100, f"Expected 100 rows, but found {count}"
+
+        # Verify some sample data
+        cursor.execute(f"SELECT id, name, value FROM {table_name} WHERE id IN (1, 50, 100) ORDER BY id")
+        rows = cursor.fetchall()
+
+        assert len(rows) == 3, f"Expected 3 sample rows, but found {len(rows)}"
+        assert rows[0][0] == 1 and rows[0][1] == "TestName_1"
+        assert rows[1][0] == 50 and rows[1][1] == "TestName_50"
+        assert rows[2][0] == 100 and rows[2][1] == "TestName_100"
+
+        print(f"✓ Successfully copied and validated 100 rows using bulk_copy")
+
+        core_conn.close()
+
+    except Exception as e:
+        pytest.skip(f"Bulk copy operation not supported or failed: {e}")
+
+    # Cleanup
+    cursor.execute(f"DROP TABLE IF EXISTS {table_name}")
+    db_connection.commit()