|
14 | 14 | import decimal |
15 | 15 | from contextlib import closing |
16 | 16 | from mssql_python import Connection, row |
| 17 | +import uuid |
17 | 18 |
|
18 | 19 | # Setup test table |
19 | 20 | TEST_TABLE = """ |
@@ -6823,6 +6824,159 @@ def test_money_smallmoney_invalid_values(cursor, db_connection): |
6823 | 6824 | drop_table_if_exists(cursor, "dbo.money_test") |
6824 | 6825 | db_connection.commit() |
6825 | 6826 |
|
| 6827 | +def test_uuid_insert_and_select_none(cursor, db_connection): |
| 6828 | + """Test inserting and retrieving None in a nullable UUID column.""" |
| 6829 | + table_name = "#pytest_uuid_nullable" |
| 6830 | + try: |
| 6831 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6832 | + cursor.execute(f""" |
| 6833 | + CREATE TABLE {table_name} ( |
| 6834 | + id UNIQUEIDENTIFIER, |
| 6835 | + name NVARCHAR(50) |
| 6836 | + ) |
| 6837 | + """) |
| 6838 | + db_connection.commit() |
| 6839 | + |
| 6840 | + # Insert a row with None for the UUID |
| 6841 | + cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Bob"]) |
| 6842 | + db_connection.commit() |
| 6843 | + |
| 6844 | + # Fetch the row |
| 6845 | + cursor.execute(f"SELECT id, name FROM {table_name}") |
| 6846 | + row = cursor.fetchone() |
| 6847 | + retrieved_uuid, retrieved_name = row |
| 6848 | + |
| 6849 | + # Assert that the retrieved UUID is None |
| 6850 | + assert retrieved_uuid is None, f"Expected None, got {type(retrieved_uuid)}" |
| 6851 | + |
| 6852 | + finally: |
| 6853 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6854 | + db_connection.commit() |
| 6855 | + |
| 6856 | +def test_insert_multiple_uuids(cursor, db_connection): |
| 6857 | + """Test inserting multiple UUIDs and verifying retrieval.""" |
| 6858 | + table_name = "#pytest_uuid_multiple" |
| 6859 | + try: |
| 6860 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6861 | + cursor.execute(f""" |
| 6862 | + CREATE TABLE {table_name} ( |
| 6863 | + id UNIQUEIDENTIFIER PRIMARY KEY, |
| 6864 | + description NVARCHAR(50) |
| 6865 | + ) |
| 6866 | + """) |
| 6867 | + db_connection.commit() |
| 6868 | + |
| 6869 | + uuids_to_insert = {f"Item {i}": uuid.uuid4() for i in range(5)} |
| 6870 | + |
| 6871 | + # Insert UUIDs and descriptions |
| 6872 | + for desc, uid in uuids_to_insert.items(): |
| 6873 | + cursor.execute(f"INSERT INTO {table_name} (id, description) VALUES (?, ?)", [uid, desc]) |
| 6874 | + db_connection.commit() |
| 6875 | + |
| 6876 | + # Fetch all data |
| 6877 | + cursor.execute(f"SELECT id, description FROM {table_name}") |
| 6878 | + rows = cursor.fetchall() |
| 6879 | + |
| 6880 | + # Verify each fetched row against the original data |
| 6881 | + assert len(rows) == len(uuids_to_insert), "Number of fetched rows does not match." |
| 6882 | + |
| 6883 | + for row in rows: |
| 6884 | + retrieved_uuid, retrieved_desc = row |
| 6885 | + |
| 6886 | + # Assert type is correct |
| 6887 | + assert isinstance(retrieved_uuid, uuid.UUID), f"Expected uuid.UUID, got {type(retrieved_uuid)}" |
| 6888 | + |
| 6889 | + # Use the description to look up the original UUID |
| 6890 | + expected_uuid = uuids_to_insert.get(retrieved_desc) |
| 6891 | + |
| 6892 | + assert expected_uuid is not None, f"Retrieved description '{retrieved_desc}' was not in the original data." |
| 6893 | + assert retrieved_uuid == expected_uuid, f"UUID mismatch for '{retrieved_desc}': expected {expected_uuid}, got {retrieved_uuid}" |
| 6894 | + finally: |
| 6895 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6896 | + db_connection.commit() |
| 6897 | + |
| 6898 | +def test_uuid_insert_with_none(cursor, db_connection): |
| 6899 | + """Test that inserting None into a UUID column raises an error (or is handled).""" |
| 6900 | + table_name = "#pytest_uuid_none" |
| 6901 | + try: |
| 6902 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6903 | + cursor.execute(f""" |
| 6904 | + CREATE TABLE {table_name} ( |
| 6905 | + id UNIQUEIDENTIFIER, |
| 6906 | + name NVARCHAR(50) |
| 6907 | + ) |
| 6908 | + """) |
| 6909 | + db_connection.commit() |
| 6910 | + |
| 6911 | + cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Bob"]) |
| 6912 | + db_connection.commit() |
| 6913 | + |
| 6914 | + cursor.execute(f"SELECT id, name FROM {table_name}") |
| 6915 | + row = cursor.fetchone() |
| 6916 | + assert row[0] is None, f"Expected NULL UUID, got {row[0]}" |
| 6917 | + assert row[1] == "Bob" |
| 6918 | + |
| 6919 | + |
| 6920 | + finally: |
| 6921 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6922 | + db_connection.commit() |
| 6923 | + |
| 6924 | +def test_executemany_uuid_insert_and_select(cursor, db_connection): |
| 6925 | + """Test inserting multiple UUIDs using executemany and verifying retrieval.""" |
| 6926 | + table_name = "#pytest_uuid_executemany" |
| 6927 | + |
| 6928 | + try: |
| 6929 | + # Drop and create a temporary table for the test |
| 6930 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6931 | + cursor.execute(f""" |
| 6932 | + CREATE TABLE {table_name} ( |
| 6933 | + id UNIQUEIDENTIFIER PRIMARY KEY, |
| 6934 | + description NVARCHAR(50) |
| 6935 | + ) |
| 6936 | + """) |
| 6937 | + db_connection.commit() |
| 6938 | + |
| 6939 | + # Generate data for insertion |
| 6940 | + data_to_insert = [] |
| 6941 | + uuids_to_check = {} |
| 6942 | + for i in range(5): |
| 6943 | + new_uuid = uuid.uuid4() |
| 6944 | + description = f"Item {i}" |
| 6945 | + data_to_insert.append((new_uuid, description)) |
| 6946 | + uuids_to_check[description] = new_uuid |
| 6947 | + |
| 6948 | + # Insert all data with a single call to executemany |
| 6949 | + sql = f"INSERT INTO {table_name} (id, description) VALUES (?, ?)" |
| 6950 | + cursor.executemany(sql, data_to_insert) |
| 6951 | + db_connection.commit() |
| 6952 | + |
| 6953 | + # Verify the number of rows inserted |
| 6954 | + assert cursor.rowcount == 5, f"Expected 5 rows inserted, but got {cursor.rowcount}" |
| 6955 | + |
| 6956 | + # Fetch all data from the table |
| 6957 | + cursor.execute(f"SELECT id, description FROM {table_name}") |
| 6958 | + rows = cursor.fetchall() |
| 6959 | + |
| 6960 | + # Verify the number of fetched rows |
| 6961 | + assert len(rows) == len(data_to_insert), "Number of fetched rows does not match." |
| 6962 | + |
| 6963 | + # Verify each fetched row's data and type |
| 6964 | + for row in rows: |
| 6965 | + retrieved_uuid, retrieved_desc = row |
| 6966 | + |
| 6967 | + # Assert the type is correct |
| 6968 | + assert isinstance(retrieved_uuid, uuid.UUID), f"Expected uuid.UUID, got {type(retrieved_uuid)}" |
| 6969 | + |
| 6970 | + # Assert the value matches the original data |
| 6971 | + expected_uuid = uuids_to_check.get(retrieved_desc) |
| 6972 | + assert expected_uuid is not None, f"Retrieved description '{retrieved_desc}' was not in the original data." |
| 6973 | + assert retrieved_uuid == expected_uuid, f"UUID mismatch for '{retrieved_desc}': expected {expected_uuid}, got {retrieved_uuid}" |
| 6974 | + |
| 6975 | + finally: |
| 6976 | + # Clean up the temporary table |
| 6977 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6978 | + db_connection.commit() |
| 6979 | + |
6826 | 6980 | def test_close(db_connection): |
6827 | 6981 | """Test closing the cursor""" |
6828 | 6982 | try: |
|
0 commit comments