diff --git a/tests/test_004_cursor.py b/tests/test_004_cursor.py index fa291c64..dbc573ec 100644 --- a/tests/test_004_cursor.py +++ b/tests/test_004_cursor.py @@ -7836,376 +7836,6 @@ def test_nvarcharmax_large(cursor, db_connection): db_connection.commit() -def test_money_smallmoney_insert_fetch(cursor, db_connection): - """Test inserting and retrieving valid MONEY and SMALLMONEY values including boundaries and typical data""" - try: - drop_table_if_exists(cursor, "#pytest_money_test") - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY, - d DECIMAL(19,4), - n NUMERIC(10,4) - ) - """) - db_connection.commit() - - # Max values - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm, d, n) VALUES (?, ?, ?, ?)", - ( - decimal.Decimal("922337203685477.5807"), - decimal.Decimal("214748.3647"), - decimal.Decimal("9999999999999.9999"), - decimal.Decimal("1234.5678"), - ), - ) - - # Min values - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm, d, n) VALUES (?, ?, ?, ?)", - ( - decimal.Decimal("-922337203685477.5808"), - decimal.Decimal("-214748.3648"), - decimal.Decimal("-9999999999999.9999"), - decimal.Decimal("-1234.5678"), - ), - ) - - # Typical values - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm, d, n) VALUES (?, ?, ?, ?)", - ( - decimal.Decimal("1234567.8901"), - decimal.Decimal("12345.6789"), - decimal.Decimal("42.4242"), - decimal.Decimal("3.1415"), - ), - ) - - # NULL values - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm, d, n) VALUES (?, ?, ?, ?)", - (None, None, None, None), - ) - - db_connection.commit() - - cursor.execute("SELECT m, sm, d, n FROM #pytest_money_test ORDER BY id") - results = cursor.fetchall() - assert len(results) == 4, f"Expected 4 rows, got {len(results)}" - - expected = [ - ( - decimal.Decimal("922337203685477.5807"), - decimal.Decimal("214748.3647"), - decimal.Decimal("9999999999999.9999"), - decimal.Decimal("1234.5678"), - ), - ( - decimal.Decimal("-922337203685477.5808"), - decimal.Decimal("-214748.3648"), - decimal.Decimal("-9999999999999.9999"), - decimal.Decimal("-1234.5678"), - ), - ( - decimal.Decimal("1234567.8901"), - decimal.Decimal("12345.6789"), - decimal.Decimal("42.4242"), - decimal.Decimal("3.1415"), - ), - (None, None, None, None), - ] - - for i, (row, exp) in enumerate(zip(results, expected)): - for j, (val, exp_val) in enumerate(zip(row, exp), 1): - if exp_val is None: - assert val is None, f"Row {i+1} col{j}: expected None, got {val}" - else: - assert val == exp_val, f"Row {i+1} col{j}: expected {exp_val}, got {val}" - assert isinstance( - val, decimal.Decimal - ), f"Row {i+1} col{j}: expected Decimal, got {type(val)}" - - except Exception as e: - pytest.fail(f"MONEY and SMALLMONEY insert/fetch test failed: {e}") - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_null_handling(cursor, db_connection): - """Test that NULL values for MONEY and SMALLMONEY are stored and retrieved correctly""" - try: - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY - ) - """) - db_connection.commit() - - # Row with both NULLs - cursor.execute("INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", (None, None)) - - # Row with m filled, sm NULL - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", - (decimal.Decimal("123.4500"), None), - ) - - # Row with m NULL, sm filled - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", - (None, decimal.Decimal("67.8900")), - ) - - db_connection.commit() - - cursor.execute("SELECT m, sm FROM #pytest_money_test ORDER BY id") - results = cursor.fetchall() - assert len(results) == 3, f"Expected 3 rows, got {len(results)}" - - expected = [ - (None, None), - (decimal.Decimal("123.4500"), None), - (None, decimal.Decimal("67.8900")), - ] - - for i, (row, exp) in enumerate(zip(results, expected)): - for j, (val, exp_val) in enumerate(zip(row, exp), 1): - if exp_val is None: - assert val is None, f"Row {i+1} col{j}: expected None, got {val}" - else: - assert val == exp_val, f"Row {i+1} col{j}: expected {exp_val}, got {val}" - assert isinstance( - val, decimal.Decimal - ), f"Row {i+1} col{j}: expected Decimal, got {type(val)}" - - except Exception as e: - pytest.fail(f"MONEY and SMALLMONEY NULL handling test failed: {e}") - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_roundtrip(cursor, db_connection): - """Test inserting and retrieving MONEY and SMALLMONEY using decimal.Decimal roundtrip""" - try: - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY - ) - """) - db_connection.commit() - - values = (decimal.Decimal("12345.6789"), decimal.Decimal("987.6543")) - cursor.execute("INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", values) - db_connection.commit() - - cursor.execute("SELECT m, sm FROM #pytest_money_test ORDER BY id DESC") - row = cursor.fetchone() - for i, (val, exp_val) in enumerate(zip(row, values), 1): - assert val == exp_val, f"col{i} roundtrip mismatch, got {val}, expected {exp_val}" - assert isinstance(val, decimal.Decimal), f"col{i} should be Decimal, got {type(val)}" - - except Exception as e: - pytest.fail(f"MONEY and SMALLMONEY roundtrip test failed: {e}") - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_boundaries(cursor, db_connection): - """Test boundary values for MONEY and SMALLMONEY types are handled correctly""" - try: - drop_table_if_exists(cursor, "#pytest_money_test") - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY - ) - """) - db_connection.commit() - - # Insert max boundary - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", - (decimal.Decimal("922337203685477.5807"), decimal.Decimal("214748.3647")), - ) - - # Insert min boundary - cursor.execute( - "INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", - (decimal.Decimal("-922337203685477.5808"), decimal.Decimal("-214748.3648")), - ) - - db_connection.commit() - - cursor.execute("SELECT m, sm FROM #pytest_money_test ORDER BY id DESC") - results = cursor.fetchall() - expected = [ - (decimal.Decimal("-922337203685477.5808"), decimal.Decimal("-214748.3648")), - (decimal.Decimal("922337203685477.5807"), decimal.Decimal("214748.3647")), - ] - for i, (row, exp_row) in enumerate(zip(results, expected), 1): - for j, (val, exp_val) in enumerate(zip(row, exp_row), 1): - assert val == exp_val, f"Row {i} col{j} mismatch, got {val}, expected {exp_val}" - assert isinstance( - val, decimal.Decimal - ), f"Row {i} col{j} should be Decimal, got {type(val)}" - - except Exception as e: - pytest.fail(f"MONEY and SMALLMONEY boundary values test failed: {e}") - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_invalid_values(cursor, db_connection): - """Test that invalid or out-of-range MONEY and SMALLMONEY values raise errors""" - try: - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY - ) - """) - db_connection.commit() - - # Out of range MONEY - with pytest.raises(Exception): - cursor.execute( - "INSERT INTO #pytest_money_test (m) VALUES (?)", - (decimal.Decimal("922337203685477.5808"),), - ) - - # Out of range SMALLMONEY - with pytest.raises(Exception): - cursor.execute( - "INSERT INTO #pytest_money_test (sm) VALUES (?)", - (decimal.Decimal("214748.3648"),), - ) - - # Invalid string - with pytest.raises(Exception): - cursor.execute("INSERT INTO #pytest_money_test (m) VALUES (?)", ("invalid_string",)) - - except Exception as e: - pytest.fail(f"MONEY and SMALLMONEY invalid values test failed: {e}") - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_roundtrip_executemany(cursor, db_connection): - """Test inserting and retrieving MONEY and SMALLMONEY using executemany with decimal.Decimal""" - try: - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY - ) - """) - db_connection.commit() - - test_data = [ - (decimal.Decimal("12345.6789"), decimal.Decimal("987.6543")), - (decimal.Decimal("0.0001"), decimal.Decimal("0.01")), - (None, decimal.Decimal("42.42")), - (decimal.Decimal("-1000.99"), None), - ] - - # Insert using executemany directly with Decimals - cursor.executemany("INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", test_data) - db_connection.commit() - - cursor.execute("SELECT m, sm FROM #pytest_money_test ORDER BY id") - results = cursor.fetchall() - assert len(results) == len(test_data) - - for i, (row, expected) in enumerate(zip(results, test_data), 1): - for j, (val, exp_val) in enumerate(zip(row, expected), 1): - if exp_val is None: - assert val is None - else: - assert val == exp_val - assert isinstance(val, decimal.Decimal) - - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_executemany_null_handling(cursor, db_connection): - """Test inserting NULLs into MONEY and SMALLMONEY using executemany""" - try: - cursor.execute(""" - CREATE TABLE #pytest_money_test ( - id INT IDENTITY PRIMARY KEY, - m MONEY, - sm SMALLMONEY - ) - """) - db_connection.commit() - - rows = [ - (None, None), - (decimal.Decimal("123.4500"), None), - (None, decimal.Decimal("67.8900")), - ] - cursor.executemany("INSERT INTO #pytest_money_test (m, sm) VALUES (?, ?)", rows) - db_connection.commit() - - cursor.execute("SELECT m, sm FROM #pytest_money_test ORDER BY id ASC") - results = cursor.fetchall() - assert len(results) == len(rows) - - for row, expected in zip(results, rows): - for val, exp_val in zip(row, expected): - if exp_val is None: - assert val is None - else: - assert val == exp_val - assert isinstance(val, decimal.Decimal) - - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - -def test_money_smallmoney_out_of_range_low(cursor, db_connection): - """Test inserting values just below the minimum MONEY/SMALLMONEY range raises error""" - try: - drop_table_if_exists(cursor, "#pytest_money_test") - cursor.execute("CREATE TABLE #pytest_money_test (m MONEY, sm SMALLMONEY)") - db_connection.commit() - - # Just below minimum MONEY - with pytest.raises(Exception): - cursor.execute( - "INSERT INTO #pytest_money_test (m) VALUES (?)", - (decimal.Decimal("-922337203685477.5809"),), - ) - - # Just below minimum SMALLMONEY - with pytest.raises(Exception): - cursor.execute( - "INSERT INTO #pytest_money_test (sm) VALUES (?)", - (decimal.Decimal("-214748.3649"),), - ) - finally: - drop_table_if_exists(cursor, "#pytest_money_test") - db_connection.commit() - - def test_uuid_insert_and_select_none(cursor, db_connection): """Test inserting and retrieving None in a nullable UUID column.""" table_name = "#pytest_uuid_nullable" diff --git a/tests/test_020_money_smallmoney.py b/tests/test_020_money_smallmoney.py new file mode 100644 index 00000000..912944c4 --- /dev/null +++ b/tests/test_020_money_smallmoney.py @@ -0,0 +1,642 @@ +""" +Tests for MONEY and SMALLMONEY type handling. + +Validates that Python Decimal values are correctly bound and round-tripped +through MONEY, SMALLMONEY, and DECIMAL columns with proper precision handling. + +Key implementation detail: MONEY-range Decimals use string binding (SQL_VARCHAR) +because SQL_NUMERIC binding fails with ODBC "Numeric value out of range" error. +String binding preserves full precision and SQL Server handles conversion. +""" + +import pytest +from decimal import Decimal + +# MONEY/SMALLMONEY range constants +MONEY_MIN = Decimal("-922337203685477.5808") +MONEY_MAX = Decimal("922337203685477.5807") +SMALLMONEY_MIN = Decimal("-214748.3648") +SMALLMONEY_MAX = Decimal("214748.3647") + + +def drop_table_if_exists(cursor, table_name): + """Drop the table if it exists.""" + try: + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") + except Exception: + pass # Ignore errors during cleanup + + +# ============================================================================= +# MONEY Decimal Binding Tests +# ============================================================================= + + +def test_money_positive_value(cursor, db_connection): + """Test positive Decimal value inserted into MONEY column.""" + table_name = "#pytest_money_pos" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("12345.6789") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_money_negative_value(cursor, db_connection): + """Test negative Decimal value inserted into MONEY column.""" + table_name = "#pytest_money_neg" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("-9999.9999") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_money_zero(cursor, db_connection): + """Test zero value inserted into MONEY column.""" + table_name = "#pytest_money_zero" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("0.0000") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == Decimal("0.0000") + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_money_max_value(cursor, db_connection): + """Test maximum MONEY value.""" + table_name = "#pytest_money_max" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (MONEY_MAX,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == MONEY_MAX + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_money_min_value(cursor, db_connection): + """Test minimum MONEY value.""" + table_name = "#pytest_money_min" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (MONEY_MIN,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == MONEY_MIN + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_money_null(cursor, db_connection): + """Test NULL value inserted into MONEY column.""" + table_name = "#pytest_money_null" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (None,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result is None + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# SMALLMONEY Decimal Binding Tests +# ============================================================================= + + +def test_smallmoney_positive_value(cursor, db_connection): + """Test positive Decimal value inserted into SMALLMONEY column.""" + table_name = "#pytest_smallmoney_pos" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + val = Decimal("1234.5678") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_smallmoney_negative_value(cursor, db_connection): + """Test negative Decimal value inserted into SMALLMONEY column.""" + table_name = "#pytest_smallmoney_neg" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + val = Decimal("-999.1234") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_smallmoney_max_value(cursor, db_connection): + """Test maximum SMALLMONEY value.""" + table_name = "#pytest_smallmoney_max" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (SMALLMONEY_MAX,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == SMALLMONEY_MAX + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_smallmoney_min_value(cursor, db_connection): + """Test minimum SMALLMONEY value.""" + table_name = "#pytest_smallmoney_min" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (SMALLMONEY_MIN,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == SMALLMONEY_MIN + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_smallmoney_null(cursor, db_connection): + """Test NULL value inserted into SMALLMONEY column.""" + table_name = "#pytest_smallmoney_null" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (None,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result is None + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# DECIMAL Column with MONEY-range Values (String Binding Path) +# ============================================================================= + + +def test_decimal_column_money_range_value(cursor, db_connection): + """Test MONEY-range Decimal inserted into DECIMAL column preserves precision.""" + table_name = "#pytest_money_dec_range" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val DECIMAL(38,20))") + db_connection.commit() + + # Value in MONEY range but with more than 4 decimal places + val = Decimal("100.123456789012345678") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + # String binding via format(param, "f") preserves all decimals + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_decimal_column_rounding_by_sql_server(cursor, db_connection): + """Test that SQL Server rounds to column precision (not driver).""" + table_name = "#pytest_money_dec_round" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val DECIMAL(10,4))") # Only 4 decimal places + db_connection.commit() + + val = Decimal("100.123456789") # 9 decimal places + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + # SQL Server rounds to scale 4 + assert result == Decimal("100.1235") + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# DECIMAL Column Outside MONEY Range (SQL_NUMERIC Binding Path) +# ============================================================================= + + +def test_decimal_above_money_max(cursor, db_connection): + """Test Decimal larger than MONEY_MAX uses SQL_NUMERIC binding.""" + table_name = "#pytest_money_dec_above" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val DECIMAL(20,2))") + db_connection.commit() + + val = Decimal("999999999999999.99") # Above MONEY_MAX + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_decimal_below_money_min(cursor, db_connection): + """Test Decimal smaller than MONEY_MIN uses SQL_NUMERIC binding.""" + table_name = "#pytest_money_dec_below" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val DECIMAL(20,2))") + db_connection.commit() + + val = Decimal("-999999999999999.99") # Below MONEY_MIN + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# Boundary Edge Cases +# ============================================================================= + + +def test_just_outside_smallmoney_to_money(cursor, db_connection): + """Test value just outside SMALLMONEY range works in MONEY column.""" + table_name = "#pytest_money_bound_out" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("214748.3648") # Just above SMALLMONEY_MAX + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_outside_smallmoney_fails_for_smallmoney_column(cursor, db_connection): + """Test value outside SMALLMONEY range fails for SMALLMONEY column.""" + table_name = "#pytest_money_bound_fail" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + val = Decimal("214748.3648") # Just above SMALLMONEY_MAX + with pytest.raises(Exception): + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_outside_money_fails_for_money_column(cursor, db_connection): + """Test value outside MONEY range fails for MONEY column.""" + table_name = "#pytest_money_bound_mfail" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("922337203685477.5808") # Just above MONEY_MAX + with pytest.raises(Exception): + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_large_money_value(cursor, db_connection): + """Test large value well within MONEY range but outside SMALLMONEY.""" + table_name = "#pytest_money_large" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("500000000000.1234") + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == val + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# Python Numeric Types to MONEY +# ============================================================================= + + +def test_python_float_to_money(cursor, db_connection): + """Test Python float inserted into MONEY column.""" + table_name = "#pytest_money_float" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = 123.4567 + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + # Float may have precision issues, compare as float + assert float(result) == pytest.approx(val, rel=1e-4) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_python_int_to_money(cursor, db_connection): + """Test Python int inserted into MONEY column.""" + table_name = "#pytest_money_int" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = 12345 + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + db_connection.commit() + + cursor.execute(f"SELECT val FROM {table_name}") + result = cursor.fetchone()[0] + assert result == Decimal("12345.0000") + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# String Binding Precision Verification +# ============================================================================= + + +def test_format_preserves_20_decimals(): + """Test that format(Decimal, 'f') preserves high precision.""" + # This tests the implementation detail that string binding preserves precision + val = Decimal("100.12345678901234567890") + formatted = format(val, "f") + assert formatted == "100.12345678901234567890" + + +# ============================================================================= +# Executemany Tests +# ============================================================================= + + +def test_executemany_money_smallmoney(cursor, db_connection): + """Test inserting multiple rows with executemany.""" + table_name = "#pytest_money_execmany" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f""" + CREATE TABLE {table_name} ( + id INT IDENTITY PRIMARY KEY, + m MONEY, + sm SMALLMONEY + ) + """) + db_connection.commit() + + test_data = [ + (Decimal("12345.6789"), Decimal("987.6543")), + (Decimal("0.0001"), Decimal("0.0100")), + (None, Decimal("42.4200")), + (Decimal("-1000.9900"), None), + ] + + cursor.executemany(f"INSERT INTO {table_name} (m, sm) VALUES (?, ?)", test_data) + db_connection.commit() + + cursor.execute(f"SELECT m, sm FROM {table_name} ORDER BY id") + results = cursor.fetchall() + assert len(results) == len(test_data) + + for row, expected in zip(results, test_data): + for val, exp_val in zip(row, expected): + if exp_val is None: + assert val is None + else: + assert val == exp_val + assert isinstance(val, Decimal) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# Invalid Input Handling +# ============================================================================= + + +def test_invalid_string_to_money(cursor, db_connection): + """Test that invalid string input raises error.""" + table_name = "#pytest_money_invalid" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + with pytest.raises(Exception): + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", ("invalid_string",)) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_below_money_min_fails(cursor, db_connection): + """Test value below MONEY_MIN raises error.""" + table_name = "#pytest_money_below_min" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val MONEY)") + db_connection.commit() + + val = Decimal("-922337203685477.5809") # Just below MONEY_MIN + with pytest.raises(Exception): + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_below_smallmoney_min_fails(cursor, db_connection): + """Test value below SMALLMONEY_MIN raises error.""" + table_name = "#pytest_money_below_sm_min" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (val SMALLMONEY)") + db_connection.commit() + + val = Decimal("-214748.3649") # Just below SMALLMONEY_MIN + with pytest.raises(Exception): + cursor.execute(f"INSERT INTO {table_name} VALUES (?)", (val,)) + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +# ============================================================================= +# Mixed NULL Scenarios +# ============================================================================= + + +def test_money_value_smallmoney_null(cursor, db_connection): + """Test MONEY has value while SMALLMONEY is NULL.""" + table_name = "#pytest_money_mixed1" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (m MONEY, sm SMALLMONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?, ?)", (Decimal("123.4500"), None)) + db_connection.commit() + + cursor.execute(f"SELECT m, sm FROM {table_name}") + row = cursor.fetchone() + assert row[0] == Decimal("123.4500") + assert row[1] is None + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_money_null_smallmoney_value(cursor, db_connection): + """Test MONEY is NULL while SMALLMONEY has value.""" + table_name = "#pytest_money_mixed2" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (m MONEY, sm SMALLMONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?, ?)", (None, Decimal("67.8900"))) + db_connection.commit() + + cursor.execute(f"SELECT m, sm FROM {table_name}") + row = cursor.fetchone() + assert row[0] is None + assert row[1] == Decimal("67.8900") + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit() + + +def test_both_null(cursor, db_connection): + """Test both MONEY and SMALLMONEY are NULL.""" + table_name = "#pytest_money_mixed3" + try: + drop_table_if_exists(cursor, table_name) + cursor.execute(f"CREATE TABLE {table_name} (m MONEY, sm SMALLMONEY)") + db_connection.commit() + + cursor.execute(f"INSERT INTO {table_name} VALUES (?, ?)", (None, None)) + db_connection.commit() + + cursor.execute(f"SELECT m, sm FROM {table_name}") + row = cursor.fetchone() + assert row[0] is None + assert row[1] is None + finally: + drop_table_if_exists(cursor, table_name) + db_connection.commit()