Skip to content
Merged
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
1 change: 1 addition & 0 deletions mssql_python/cursor.py
Original file line number Diff line number Diff line change
Expand Up @@ -1613,6 +1613,7 @@ def executemany(self, operation: str, seq_of_parameters: list) -> None:
# Use auto-detection for columns without explicit types
column = [row[col_index] for row in seq_of_parameters] if hasattr(seq_of_parameters, '__getitem__') else []
sample_value, min_val, max_val = self._compute_column_type(column)

dummy_row = list(sample_row)
paraminfo = self._create_parameter_types_list(
sample_value, param_info, dummy_row, col_index, min_val=min_val, max_val=max_val
Expand Down
112 changes: 112 additions & 0 deletions tests/test_004_cursor.py
Original file line number Diff line number Diff line change
Expand Up @@ -1262,6 +1262,118 @@ def test_executemany_binary_data_edge_cases(cursor, db_connection):
cursor.execute("DROP TABLE IF EXISTS #pytest_binary_test")
db_connection.commit()

def test_executemany_mixed_ints(cursor, db_connection):
"""Test executemany with mixed positive and negative integers."""
try:
cursor.execute("CREATE TABLE #pytest_mixed_ints (val INT)")
data = [(1,), (-5,), (3,)]
cursor.executemany("INSERT INTO #pytest_mixed_ints VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT val FROM #pytest_mixed_ints ORDER BY val")
results = [row[0] for row in cursor.fetchall()]
assert sorted(results) == [-5, 1, 3]
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_mixed_ints")
db_connection.commit()


def test_executemany_int_edge_cases(cursor, db_connection):
"""Test executemany with very large and very small integers."""
try:
cursor.execute("CREATE TABLE #pytest_int_edges (val BIGINT)")
data = [(0,), (2**31-1,), (-2**31,), (2**63-1,), (-2**63,)]
cursor.executemany("INSERT INTO #pytest_int_edges VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT val FROM #pytest_int_edges ORDER BY val")
results = [row[0] for row in cursor.fetchall()]
assert results == sorted([0, 2**31-1, -2**31, 2**63-1, -2**63])
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_int_edges")
db_connection.commit()


def test_executemany_bools_and_ints(cursor, db_connection):
"""Test executemany with mix of booleans and integers."""
try:
cursor.execute("CREATE TABLE #pytest_bool_int (val INT)")
data = [(True,), (False,), (2,)]
cursor.executemany("INSERT INTO #pytest_bool_int VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT val FROM #pytest_bool_int ORDER BY val")
results = [row[0] for row in cursor.fetchall()]
# True -> 1, False -> 0
assert results == [0, 1, 2]
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_bool_int")
db_connection.commit()


def test_executemany_ints_with_none(cursor, db_connection):
"""Test executemany with integers and None values."""
try:
cursor.execute("CREATE TABLE #pytest_int_none (val INT)")
data = [(1,), (None,), (3,)]
cursor.executemany("INSERT INTO #pytest_int_none VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT val FROM #pytest_int_none ORDER BY val")
results = [row[0] for row in cursor.fetchall()]
assert results.count(None) == 1
assert results.count(1) == 1
assert results.count(3) == 1
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_int_none")
db_connection.commit()


def test_executemany_strings_of_various_lengths(cursor, db_connection):
"""Test executemany with strings of different lengths."""
try:
cursor.execute("CREATE TABLE #pytest_varied_strings (val NVARCHAR(50))")
data = [("a",), ("abcd",), ("abc",)]
cursor.executemany("INSERT INTO #pytest_varied_strings VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT val FROM #pytest_varied_strings ORDER BY val")
results = [row[0] for row in cursor.fetchall()]
assert sorted(results) == ["a", "abc", "abcd"]
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_varied_strings")
db_connection.commit()


def test_executemany_bytes_values(cursor, db_connection):
"""Test executemany with bytes values."""
try:
cursor.execute("CREATE TABLE #pytest_bytes (val VARBINARY(50))")
data = [(b"a",), (b"abcdef",)]
cursor.executemany("INSERT INTO #pytest_bytes VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT val FROM #pytest_bytes ORDER BY val")
results = [row[0] for row in cursor.fetchall()]
assert results == [b"a", b"abcdef"]
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_bytes")
db_connection.commit()

def test_executemany_empty_parameter_list(cursor, db_connection):
"""Test executemany with an empty parameter list."""
try:
cursor.execute("CREATE TABLE #pytest_empty_params (val INT)")
data = []
cursor.executemany("INSERT INTO #pytest_empty_params VALUES (?)", data)
db_connection.commit()

cursor.execute("SELECT COUNT(*) FROM #pytest_empty_params")
count = cursor.fetchone()[0]
assert count == 0
finally:
cursor.execute("DROP TABLE IF EXISTS #pytest_empty_params")
db_connection.commit()

def test_nextset(cursor):
"""Test nextset"""
Expand Down