improve test isolation

Author: ptiurin

tests/integration/dbapi/async/V2/test_queries_async.py

@@ -773,18 +773,19 @@ async def test_transaction_commit(
     connection: Connection, create_drop_test_table_setup_teardown_async: Callable
 ) -> None:
     """Test transaction SQL statements with COMMIT."""
+    table_name = create_drop_test_table_setup_teardown_async
     async with connection.cursor() as c:
         # Test successful transaction with COMMIT
         result = await c.execute("BEGIN TRANSACTION")
         assert result == 0, "BEGIN TRANSACTION should return 0 rows"
 
-        await c.execute("INSERT INTO \"test_tbl\" VALUES (1, 'committed')")
+        await c.execute(f"INSERT INTO \"{table_name}\" VALUES (1, 'committed')")
 
         result = await c.execute("COMMIT TRANSACTION")
         assert result == 0, "COMMIT TRANSACTION should return 0 rows"
 
         # Verify the data was committed
-        await c.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+        await c.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
         data = await c.fetchall()
         assert len(data) == 1, "Committed data should be present"
         assert data[0] == [
@@ -797,23 +798,24 @@ async def test_transaction_rollback(
     connection: Connection, create_drop_test_table_setup_teardown_async: Callable
 ) -> None:
     """Test transaction SQL statements with ROLLBACK."""
+    table_name = create_drop_test_table_setup_teardown_async
     async with connection.cursor() as c:
         # Test transaction with ROLLBACK
         result = await c.execute("BEGIN")  # Test short form
         assert result == 0, "BEGIN should return 0 rows"
 
-        await c.execute("INSERT INTO \"test_tbl\" VALUES (1, 'rolled_back')")
+        await c.execute(f"INSERT INTO \"{table_name}\" VALUES (1, 'rolled_back')")
 
         # Verify data is visible within transaction
-        await c.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+        await c.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
         data = await c.fetchall()
         assert len(data) == 1, "Data should be visible within transaction"
 
         result = await c.execute("ROLLBACK")  # Test short form
         assert result == 0, "ROLLBACK should return 0 rows"
 
         # Verify the data was rolled back
-        await c.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+        await c.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
         data = await c.fetchall()
         assert len(data) == 0, "Rolled back data should not be present"
 
@@ -822,23 +824,24 @@ async def test_transaction_cursor_isolation(
     connection: Connection, create_drop_test_table_setup_teardown_async: Callable
 ) -> None:
     """Test that one cursor can't see another's data until it commits."""
+    table_name = create_drop_test_table_setup_teardown_async
     cursor1 = connection.cursor()
     cursor2 = connection.cursor()
 
     # Start transaction in cursor1 and insert data
     result = await cursor1.execute("BEGIN TRANSACTION")
     assert result == 0, "BEGIN TRANSACTION should return 0 rows"
 
-    await cursor1.execute("INSERT INTO \"test_tbl\" VALUES (1, 'isolated_data')")
+    await cursor1.execute(f"INSERT INTO \"{table_name}\" VALUES (1, 'isolated_data')")
 
     # Verify cursor1 can see its own uncommitted data
-    await cursor1.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+    await cursor1.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
     data1 = await cursor1.fetchall()
     assert len(data1) == 1, "Cursor1 should see its own uncommitted data"
     assert data1[0] == [1, "isolated_data"], "Cursor1 data should match inserted values"
 
     # Verify cursor2 cannot see cursor1's uncommitted data
-    await cursor2.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+    await cursor2.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
     data2 = await cursor2.fetchall()
     assert len(data2) == 0, "Cursor2 should not see cursor1's uncommitted data"
 
@@ -847,7 +850,7 @@ async def test_transaction_cursor_isolation(
     assert result == 0, "COMMIT TRANSACTION should return 0 rows"
 
     # Now cursor2 should be able to see the committed data
-    await cursor2.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+    await cursor2.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
     data2 = await cursor2.fetchall()
     assert len(data2) == 1, "Cursor2 should see committed data after commit"
     assert data2[0] == [1, "isolated_data"], "Cursor2 should see the committed data"

tests/integration/dbapi/conftest.py

@@ -1,4 +1,5 @@
 import os
+import uuid
 from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from logging import getLogger
@@ -12,8 +13,8 @@
 
 LOGGER = getLogger(__name__)
 
-CREATE_TEST_TABLE = 'CREATE TABLE IF NOT EXISTS "test_tbl" (id int, name string)'
-DROP_TEST_TABLE = 'DROP TABLE IF EXISTS "test_tbl" CASCADE'
+CREATE_TEST_TABLE = 'CREATE TABLE IF NOT EXISTS "{table}" (id int, name string)'
+DROP_TEST_TABLE = 'DROP TABLE IF EXISTS "{table}" CASCADE'
 
 LONG_SELECT_DEFAULT_V1 = 250000000000
 LONG_SELECT_DEFAULT_V2 = 350000000000
@@ -29,38 +30,27 @@ def long_test_value_with_default(default: int = 0) -> int:
     return long_test_value_with_default
 
 
-@fixture
-def create_drop_test_table_setup_teardown(connection: Connection) -> None:
-    with connection.cursor() as c:
-        c.execute(CREATE_TEST_TABLE)
-        yield c
-        c.execute(DROP_TEST_TABLE)
-
-
-@fixture
-async def create_server_side_test_table_setup_teardown_async(
-    connection: Connection,
-) -> None:
-    with connection.cursor() as c:
-        await c.execute(CREATE_TEST_TABLE)
-        yield c
-        await c.execute(DROP_TEST_TABLE)
+def generate_unique_table_name() -> str:
+    """Generate a unique table name for testing purposes."""
+    return f"test_table_{uuid.uuid4().hex}"
 
 
 @fixture
 def create_drop_test_table_setup_teardown(connection: Connection) -> None:
+    table = generate_unique_table_name()
     with connection.cursor() as c:
-        c.execute(CREATE_TEST_TABLE)
-        yield c
-        c.execute(DROP_TEST_TABLE)
+        c.execute(CREATE_TEST_TABLE.format(table=table))
+        yield table
+        c.execute(DROP_TEST_TABLE.format(table=table))
 
 
 @fixture
 async def create_drop_test_table_setup_teardown_async(connection: Connection) -> None:
+    table = generate_unique_table_name()
     async with connection.cursor() as c:
-        await c.execute(CREATE_TEST_TABLE)
-        yield c
-        await c.execute(DROP_TEST_TABLE)
+        await c.execute(CREATE_TEST_TABLE.format(table=table))
+        yield table
+        await c.execute(DROP_TEST_TABLE.format(table=table))
 
 
 @fixture

tests/integration/dbapi/sync/V2/test_queries.py

@@ -774,18 +774,19 @@ def test_transaction_commit(
     connection: Connection, create_drop_test_table_setup_teardown: Callable
 ) -> None:
     """Test transaction SQL statements with COMMIT."""
+    table_name = create_drop_test_table_setup_teardown
     with connection.cursor() as c:
         # Test successful transaction with COMMIT
         result = c.execute("BEGIN TRANSACTION")
         assert result == 0, "BEGIN TRANSACTION should return 0 rows"
 
-        c.execute("INSERT INTO \"test_tbl\" VALUES (1, 'committed')")
+        c.execute(f"INSERT INTO \"{table_name}\" VALUES (1, 'committed')")
 
         result = c.execute("COMMIT TRANSACTION")
         assert result == 0, "COMMIT TRANSACTION should return 0 rows"
 
         # Verify the data was committed
-        c.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+        c.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
         data = c.fetchall()
         assert len(data) == 1, "Committed data should be present"
         assert data[0] == [
@@ -798,23 +799,24 @@ def test_transaction_rollback(
     connection: Connection, create_drop_test_table_setup_teardown: Callable
 ) -> None:
     """Test transaction SQL statements with ROLLBACK."""
+    table_name = create_drop_test_table_setup_teardown
     with connection.cursor() as c:
         # Test transaction with ROLLBACK
         result = c.execute("BEGIN")  # Test short form
         assert result == 0, "BEGIN should return 0 rows"
 
-        c.execute("INSERT INTO \"test_tbl\" VALUES (1, 'rolled_back')")
+        c.execute(f"INSERT INTO \"{table_name}\" VALUES (1, 'rolled_back')")
 
         # Verify data is visible within transaction
-        c.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+        c.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
         data = c.fetchall()
         assert len(data) == 1, "Data should be visible within transaction"
 
         result = c.execute("ROLLBACK")  # Test short form
         assert result == 0, "ROLLBACK should return 0 rows"
 
         # Verify the data was rolled back
-        c.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+        c.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
         data = c.fetchall()
         assert len(data) == 0, "Rolled back data should not be present"
 
@@ -823,23 +825,24 @@ def test_transaction_cursor_isolation(
     connection: Connection, create_drop_test_table_setup_teardown: Callable
 ) -> None:
     """Test that one cursor can't see another's data until it commits."""
+    table_name = create_drop_test_table_setup_teardown
     cursor1 = connection.cursor()
     cursor2 = connection.cursor()
 
     # Start transaction in cursor1 and insert data
     result = cursor1.execute("BEGIN TRANSACTION")
     assert result == 0, "BEGIN TRANSACTION should return 0 rows"
 
-    cursor1.execute("INSERT INTO \"test_tbl\" VALUES (1, 'isolated_data')")
+    cursor1.execute(f"INSERT INTO \"{table_name}\" VALUES (1, 'isolated_data')")
 
     # Verify cursor1 can see its own uncommitted data
-    cursor1.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+    cursor1.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
     data1 = cursor1.fetchall()
     assert len(data1) == 1, "Cursor1 should see its own uncommitted data"
     assert data1[0] == [1, "isolated_data"], "Cursor1 data should match inserted values"
 
     # Verify cursor2 cannot see cursor1's uncommitted data
-    cursor2.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+    cursor2.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
     data2 = cursor2.fetchall()
     assert len(data2) == 0, "Cursor2 should not see cursor1's uncommitted data"
 
@@ -848,7 +851,7 @@ def test_transaction_cursor_isolation(
     assert result == 0, "COMMIT TRANSACTION should return 0 rows"
 
     # Now cursor2 should be able to see the committed data
-    cursor2.execute('SELECT * FROM "test_tbl" WHERE id = 1')
+    cursor2.execute(f'SELECT * FROM "{table_name}" WHERE id = 1')
     data2 = cursor2.fetchall()
     assert len(data2) == 1, "Cursor2 should see committed data after commit"
     assert data2[0] == [1, "isolated_data"], "Cursor2 should see the committed data"