|
14 | 14 | import decimal |
15 | 15 | from contextlib import closing |
16 | 16 | import mssql_python |
| 17 | +import uuid |
17 | 18 |
|
18 | 19 | # Setup test table |
19 | 20 | TEST_TABLE = """ |
@@ -10193,6 +10194,158 @@ def test_decimal_separator_calculations(cursor, db_connection): |
10193 | 10194 |
|
10194 | 10195 | # Cleanup |
10195 | 10196 | cursor.execute("DROP TABLE IF EXISTS #pytest_decimal_calc_test") |
| 10197 | + |
| 10198 | +def test_uuid_insert_and_select_none(cursor, db_connection): |
| 10199 | + """Test inserting and retrieving None in a nullable UUID column.""" |
| 10200 | + table_name = "#pytest_uuid_nullable" |
| 10201 | + try: |
| 10202 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10203 | + cursor.execute(f""" |
| 10204 | + CREATE TABLE {table_name} ( |
| 10205 | + id UNIQUEIDENTIFIER, |
| 10206 | + name NVARCHAR(50) |
| 10207 | + ) |
| 10208 | + """) |
| 10209 | + db_connection.commit() |
| 10210 | + |
| 10211 | + # Insert a row with None for the UUID |
| 10212 | + cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Bob"]) |
| 10213 | + db_connection.commit() |
| 10214 | + |
| 10215 | + # Fetch the row |
| 10216 | + cursor.execute(f"SELECT id, name FROM {table_name}") |
| 10217 | + row = cursor.fetchone() |
| 10218 | + retrieved_uuid, retrieved_name = row |
| 10219 | + |
| 10220 | + # Assert that the retrieved UUID is None |
| 10221 | + assert retrieved_uuid is None, f"Expected None, got {type(retrieved_uuid)}" |
| 10222 | + |
| 10223 | + finally: |
| 10224 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10225 | + db_connection.commit() |
| 10226 | + |
| 10227 | +def test_insert_multiple_uuids(cursor, db_connection): |
| 10228 | + """Test inserting multiple UUIDs and verifying retrieval.""" |
| 10229 | + table_name = "#pytest_uuid_multiple" |
| 10230 | + try: |
| 10231 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10232 | + cursor.execute(f""" |
| 10233 | + CREATE TABLE {table_name} ( |
| 10234 | + id UNIQUEIDENTIFIER PRIMARY KEY, |
| 10235 | + description NVARCHAR(50) |
| 10236 | + ) |
| 10237 | + """) |
| 10238 | + db_connection.commit() |
| 10239 | + |
| 10240 | + uuids_to_insert = {f"Item {i}": uuid.uuid4() for i in range(5)} |
| 10241 | + |
| 10242 | + # Insert UUIDs and descriptions |
| 10243 | + for desc, uid in uuids_to_insert.items(): |
| 10244 | + cursor.execute(f"INSERT INTO {table_name} (id, description) VALUES (?, ?)", [uid, desc]) |
| 10245 | + db_connection.commit() |
| 10246 | + |
| 10247 | + # Fetch all data |
| 10248 | + cursor.execute(f"SELECT id, description FROM {table_name}") |
| 10249 | + rows = cursor.fetchall() |
| 10250 | + |
| 10251 | + # Verify each fetched row against the original data |
| 10252 | + assert len(rows) == len(uuids_to_insert), "Number of fetched rows does not match." |
| 10253 | + |
| 10254 | + for row in rows: |
| 10255 | + retrieved_uuid, retrieved_desc = row |
| 10256 | + |
| 10257 | + # Assert type is correct |
| 10258 | + assert isinstance(retrieved_uuid, uuid.UUID), f"Expected uuid.UUID, got {type(retrieved_uuid)}" |
| 10259 | + |
| 10260 | + # Use the description to look up the original UUID |
| 10261 | + expected_uuid = uuids_to_insert.get(retrieved_desc) |
| 10262 | + |
| 10263 | + assert expected_uuid is not None, f"Retrieved description '{retrieved_desc}' was not in the original data." |
| 10264 | + assert retrieved_uuid == expected_uuid, f"UUID mismatch for '{retrieved_desc}': expected {expected_uuid}, got {retrieved_uuid}" |
| 10265 | + finally: |
| 10266 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10267 | + db_connection.commit() |
| 10268 | + |
| 10269 | +def test_uuid_insert_with_none(cursor, db_connection): |
| 10270 | + """Test that inserting None into a UUID column raises an error (or is handled).""" |
| 10271 | + table_name = "#pytest_uuid_none" |
| 10272 | + try: |
| 10273 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10274 | + cursor.execute(f""" |
| 10275 | + CREATE TABLE {table_name} ( |
| 10276 | + id UNIQUEIDENTIFIER, |
| 10277 | + name NVARCHAR(50) |
| 10278 | + ) |
| 10279 | + """) |
| 10280 | + db_connection.commit() |
| 10281 | + |
| 10282 | + cursor.execute(f"INSERT INTO {table_name} (id, name) VALUES (?, ?)", [None, "Bob"]) |
| 10283 | + db_connection.commit() |
| 10284 | + |
| 10285 | + cursor.execute(f"SELECT id, name FROM {table_name}") |
| 10286 | + row = cursor.fetchone() |
| 10287 | + assert row[0] is None, f"Expected NULL UUID, got {row[0]}" |
| 10288 | + assert row[1] == "Bob" |
| 10289 | + |
| 10290 | + |
| 10291 | + finally: |
| 10292 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10293 | + db_connection.commit() |
| 10294 | + |
| 10295 | +def test_executemany_uuid_insert_and_select(cursor, db_connection): |
| 10296 | + """Test inserting multiple UUIDs using executemany and verifying retrieval.""" |
| 10297 | + table_name = "#pytest_uuid_executemany" |
| 10298 | + |
| 10299 | + try: |
| 10300 | + # Drop and create a temporary table for the test |
| 10301 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
| 10302 | + cursor.execute(f""" |
| 10303 | + CREATE TABLE {table_name} ( |
| 10304 | + id UNIQUEIDENTIFIER PRIMARY KEY, |
| 10305 | + description NVARCHAR(50) |
| 10306 | + ) |
| 10307 | + """) |
| 10308 | + db_connection.commit() |
| 10309 | + |
| 10310 | + # Generate data for insertion |
| 10311 | + data_to_insert = [] |
| 10312 | + uuids_to_check = {} |
| 10313 | + for i in range(5): |
| 10314 | + new_uuid = uuid.uuid4() |
| 10315 | + description = f"Item {i}" |
| 10316 | + data_to_insert.append((new_uuid, description)) |
| 10317 | + uuids_to_check[description] = new_uuid |
| 10318 | + |
| 10319 | + # Insert all data with a single call to executemany |
| 10320 | + sql = f"INSERT INTO {table_name} (id, description) VALUES (?, ?)" |
| 10321 | + cursor.executemany(sql, data_to_insert) |
| 10322 | + db_connection.commit() |
| 10323 | + |
| 10324 | + # Verify the number of rows inserted |
| 10325 | + assert cursor.rowcount == 5, f"Expected 5 rows inserted, but got {cursor.rowcount}" |
| 10326 | + |
| 10327 | + # Fetch all data from the table |
| 10328 | + cursor.execute(f"SELECT id, description FROM {table_name}") |
| 10329 | + rows = cursor.fetchall() |
| 10330 | + |
| 10331 | + # Verify the number of fetched rows |
| 10332 | + assert len(rows) == len(data_to_insert), "Number of fetched rows does not match." |
| 10333 | + |
| 10334 | + # Verify each fetched row's data and type |
| 10335 | + for row in rows: |
| 10336 | + retrieved_uuid, retrieved_desc = row |
| 10337 | + |
| 10338 | + # Assert the type is correct |
| 10339 | + assert isinstance(retrieved_uuid, uuid.UUID), f"Expected uuid.UUID, got {type(retrieved_uuid)}" |
| 10340 | + |
| 10341 | + # Assert the value matches the original data |
| 10342 | + expected_uuid = uuids_to_check.get(retrieved_desc) |
| 10343 | + assert expected_uuid is not None, f"Retrieved description '{retrieved_desc}' was not in the original data." |
| 10344 | + assert retrieved_uuid == expected_uuid, f"UUID mismatch for '{retrieved_desc}': expected {expected_uuid}, got {retrieved_uuid}" |
| 10345 | + |
| 10346 | + finally: |
| 10347 | + # Clean up the temporary table |
| 10348 | + cursor.execute(f"DROP TABLE IF EXISTS {table_name}") |
10196 | 10349 | db_connection.commit() |
10197 | 10350 |
|
10198 | 10351 | def test_close(db_connection): |
|
0 commit comments