|
14 | 14 | import decimal |
15 | 15 | from contextlib import closing |
16 | 16 | import mssql_python |
| 17 | +import uuid |
| 18 | + |
17 | 19 |
|
18 | 20 | # Setup test table |
19 | 21 | TEST_TABLE = """ |
@@ -6942,6 +6944,208 @@ def test_money_smallmoney_invalid_values(cursor, db_connection): |
6942 | 6944 | drop_table_if_exists(cursor, "dbo.money_test") |
6943 | 6945 | db_connection.commit() |
6944 | 6946 |
|
| 6947 | +def test_uuid_insert_and_select_none(cursor, db_connection): |
| 6948 | + """Test inserting and retrieving None in a nullable UUID column.""" |
| 6949 | + table_name = "#pytest_uuid_nullable" |
| 6950 | + try: |
| 6951 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6952 | + cursor.execute(f""" |
| 6953 | + CREATE TABLE {table_name} ( |
| 6954 | + id UNIQUEIDENTIFIER, |
| 6955 | + name NVARCHAR(50) |
| 6956 | + ) |
| 6957 | + """) |
| 6958 | + db_connection.commit() |
| 6959 | + |
| 6960 | + # Insert a row with None for the UUID |
| 6961 | + cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Bob"]) |
| 6962 | + db_connection.commit() |
| 6963 | + |
| 6964 | + # Fetch the row |
| 6965 | + cursor.execute(f"SELECT id, name FROM {table_name}") |
| 6966 | + retrieved_uuid, retrieved_name = cursor.fetchone() |
| 6967 | + |
| 6968 | + # Assert correct results |
| 6969 | + assert retrieved_uuid is None, f"Expected None, got {retrieved_uuid}" |
| 6970 | + assert retrieved_name == "Bob" |
| 6971 | + finally: |
| 6972 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6973 | + db_connection.commit() |
| 6974 | + |
| 6975 | + |
| 6976 | +def test_insert_multiple_uuids(cursor, db_connection): |
| 6977 | + """Test inserting multiple UUIDs and verifying retrieval.""" |
| 6978 | + table_name = "#pytest_uuid_multiple" |
| 6979 | + try: |
| 6980 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 6981 | + cursor.execute(f""" |
| 6982 | + CREATE TABLE {table_name} ( |
| 6983 | + id UNIQUEIDENTIFIER PRIMARY KEY, |
| 6984 | + description NVARCHAR(50) |
| 6985 | + ) |
| 6986 | + """) |
| 6987 | + db_connection.commit() |
| 6988 | + |
| 6989 | + # Prepare test data |
| 6990 | + uuids_to_insert = {f"Item {i}": uuid.uuid4() for i in range(5)} |
| 6991 | + |
| 6992 | + # Insert UUIDs and descriptions |
| 6993 | + for desc, uid in uuids_to_insert.items(): |
| 6994 | + cursor.execute(f"INSERT INTO {table_name} (id, description) VALUES (?, ?)", [uid, desc]) |
| 6995 | + db_connection.commit() |
| 6996 | + |
| 6997 | + # Fetch all rows |
| 6998 | + cursor.execute(f"SELECT id, description FROM {table_name}") |
| 6999 | + rows = cursor.fetchall() |
| 7000 | + |
| 7001 | + # Verify each fetched row |
| 7002 | + assert len(rows) == len(uuids_to_insert), "Fetched row count mismatch" |
| 7003 | + |
| 7004 | + for retrieved_uuid, retrieved_desc in rows: |
| 7005 | + assert isinstance(retrieved_uuid, uuid.UUID), f"Expected uuid.UUID, got {type(retrieved_uuid)}" |
| 7006 | + expected_uuid = uuids_to_insert[retrieved_desc] |
| 7007 | + assert retrieved_uuid == expected_uuid, f"UUID mismatch for '{retrieved_desc}': expected {expected_uuid}, got {retrieved_uuid}" |
| 7008 | + finally: |
| 7009 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7010 | + db_connection.commit() |
| 7011 | + |
| 7012 | + |
| 7013 | +def test_fetchmany_uuids(cursor, db_connection): |
| 7014 | + """Test fetching multiple UUID rows with fetchmany().""" |
| 7015 | + table_name = "#pytest_uuid_fetchmany" |
| 7016 | + try: |
| 7017 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7018 | + cursor.execute(f""" |
| 7019 | + CREATE TABLE {table_name} ( |
| 7020 | + id UNIQUEIDENTIFIER PRIMARY KEY, |
| 7021 | + description NVARCHAR(50) |
| 7022 | + ) |
| 7023 | + """) |
| 7024 | + db_connection.commit() |
| 7025 | + |
| 7026 | + uuids_to_insert = {f"Item {i}": uuid.uuid4() for i in range(10)} |
| 7027 | + |
| 7028 | + for desc, uid in uuids_to_insert.items(): |
| 7029 | + cursor.execute(f"INSERT INTO {table_name} (id, description) VALUES (?, ?)", [uid, desc]) |
| 7030 | + db_connection.commit() |
| 7031 | + |
| 7032 | + cursor.execute(f"SELECT id, description FROM {table_name}") |
| 7033 | + |
| 7034 | + # Fetch in batches of 3 |
| 7035 | + batch_size = 3 |
| 7036 | + fetched_rows = [] |
| 7037 | + while True: |
| 7038 | + batch = cursor.fetchmany(batch_size) |
| 7039 | + if not batch: |
| 7040 | + break |
| 7041 | + fetched_rows.extend(batch) |
| 7042 | + |
| 7043 | + # Verify all rows |
| 7044 | + assert len(fetched_rows) == len(uuids_to_insert), "Fetched row count mismatch" |
| 7045 | + for retrieved_uuid, retrieved_desc in fetched_rows: |
| 7046 | + assert isinstance(retrieved_uuid, uuid.UUID) |
| 7047 | + expected_uuid = uuids_to_insert[retrieved_desc] |
| 7048 | + assert retrieved_uuid == expected_uuid |
| 7049 | + finally: |
| 7050 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7051 | + db_connection.commit() |
| 7052 | + |
| 7053 | + |
| 7054 | +def test_uuid_insert_with_none(cursor, db_connection): |
| 7055 | + """Test inserting None into a UUID column results in a NULL value.""" |
| 7056 | + table_name = "#pytest_uuid_none" |
| 7057 | + try: |
| 7058 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7059 | + cursor.execute(f""" |
| 7060 | + CREATE TABLE {table_name} ( |
| 7061 | + id UNIQUEIDENTIFIER, |
| 7062 | + name NVARCHAR(50) |
| 7063 | + ) |
| 7064 | + """) |
| 7065 | + db_connection.commit() |
| 7066 | + |
| 7067 | + cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Alice"]) |
| 7068 | + db_connection.commit() |
| 7069 | + |
| 7070 | + cursor.execute(f"SELECT id, name FROM {table_name}") |
| 7071 | + retrieved_uuid, retrieved_name = cursor.fetchone() |
| 7072 | + |
| 7073 | + assert retrieved_uuid is None, f"Expected NULL UUID, got {retrieved_uuid}" |
| 7074 | + assert retrieved_name == "Alice" |
| 7075 | + finally: |
| 7076 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7077 | + db_connection.commit() |
| 7078 | + |
| 7079 | +def test_invalid_uuid_inserts(cursor, db_connection): |
| 7080 | + """Test inserting invalid UUID values raises appropriate errors.""" |
| 7081 | + table_name = "#pytest_uuid_invalid" |
| 7082 | + try: |
| 7083 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7084 | + cursor.execute(f"CREATE TABLE {table_name} (id UNIQUEIDENTIFIER)") |
| 7085 | + db_connection.commit() |
| 7086 | + |
| 7087 | + invalid_values = [ |
| 7088 | + "12345", # Too short |
| 7089 | + "not-a-uuid", # Not a UUID string |
| 7090 | + 123456789, # Integer |
| 7091 | + 12.34, # Float |
| 7092 | + object() # Arbitrary object |
| 7093 | + ] |
| 7094 | + |
| 7095 | + for val in invalid_values: |
| 7096 | + with pytest.raises(Exception): |
| 7097 | + cursor.execute(f"INSERT INTO {table_name} (id) VALUES (?)", [val]) |
| 7098 | + db_connection.commit() |
| 7099 | + finally: |
| 7100 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7101 | + db_connection.commit() |
| 7102 | + |
| 7103 | +def test_duplicate_uuid_inserts(cursor, db_connection): |
| 7104 | + """Test that inserting duplicate UUIDs into a PK column raises an error.""" |
| 7105 | + table_name = "#pytest_uuid_duplicate" |
| 7106 | + try: |
| 7107 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7108 | + cursor.execute(f"CREATE TABLE {table_name} (id UNIQUEIDENTIFIER PRIMARY KEY)") |
| 7109 | + db_connection.commit() |
| 7110 | + |
| 7111 | + uid = uuid.uuid4() |
| 7112 | + cursor.execute(f"INSERT INTO {table_name} (id) VALUES (?)", [uid]) |
| 7113 | + db_connection.commit() |
| 7114 | + |
| 7115 | + with pytest.raises(Exception): |
| 7116 | + cursor.execute(f"INSERT INTO {table_name} (id) VALUES (?)", [uid]) |
| 7117 | + db_connection.commit() |
| 7118 | + finally: |
| 7119 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7120 | + db_connection.commit() |
| 7121 | + |
| 7122 | +def test_extreme_uuids(cursor, db_connection): |
| 7123 | + """Test inserting extreme but valid UUIDs.""" |
| 7124 | + table_name = "#pytest_uuid_extreme" |
| 7125 | + try: |
| 7126 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7127 | + cursor.execute(f"CREATE TABLE {table_name} (id UNIQUEIDENTIFIER)") |
| 7128 | + db_connection.commit() |
| 7129 | + |
| 7130 | + extreme_uuids = [ |
| 7131 | + uuid.UUID(int=0), # All zeros |
| 7132 | + uuid.UUID(int=(1 << 128) - 1), # All ones |
| 7133 | + ] |
| 7134 | + |
| 7135 | + for uid in extreme_uuids: |
| 7136 | + cursor.execute(f"INSERT INTO {table_name} (id) VALUES (?)", [uid]) |
| 7137 | + db_connection.commit() |
| 7138 | + |
| 7139 | + cursor.execute(f"SELECT id FROM {table_name}") |
| 7140 | + rows = cursor.fetchall() |
| 7141 | + fetched_uuids = [row[0] for row in rows] |
| 7142 | + |
| 7143 | + for uid in extreme_uuids: |
| 7144 | + assert uid in fetched_uuids, f"Extreme UUID {uid} not retrieved correctly" |
| 7145 | + finally: |
| 7146 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 7147 | + db_connection.commit() |
| 7148 | + |
6945 | 7149 | def test_decimal_separator_with_multiple_values(cursor, db_connection): |
6946 | 7150 | """Test decimal separator with multiple different decimal values""" |
6947 | 7151 | original_separator = mssql_python.getDecimalSeparator() |
@@ -10193,7 +10397,6 @@ def test_decimal_separator_calculations(cursor, db_connection): |
10193 | 10397 |
|
10194 | 10398 | # Cleanup |
10195 | 10399 | cursor.execute("DROP TABLE IF EXISTS #pytest_decimal_calc_test") |
10196 | | - db_connection.commit() |
10197 | 10400 |
|
10198 | 10401 | def test_close(db_connection): |
10199 | 10402 | """Test closing the cursor""" |
|
0 commit comments