fix(sqlalchemy-spanner): isolate literal round trip test tables with unique naming and teardown

packages/sqlalchemy-spanner/tests/conftest.py

@@ -15,6 +15,7 @@
 # limitations under the License.
 
 import importlib
+import uuid
 from contextlib import contextmanager
 from unittest.mock import MagicMock
 
@@ -57,11 +58,87 @@ def no_op_trace_call(*args, **kwargs):
 from sqlalchemy.testing.plugin.pytestplugin import *  # noqa: E402, F401, F403
 
 
+def _commit_ddl(connection):
+    """Flush and commit buffered DDL in Spanner DBAPI or underlying connections."""
+    # TODO: modern sqlalchemy uses dbapi_connection, remove the getattr check
+    # for "connection" once we drop support for sqlalchemy < 2.0
+    dbapi_conn = getattr(
+        connection, "dbapi_connection", getattr(connection, "connection", connection)
+    )
+    commit_fn = getattr(dbapi_conn, "commit", None)
+    if callable(commit_fn):
+        commit_fn()
+
+
+def _rollback_txn(connection):
+    """Roll back any active Spanner transaction to release schema and row locks."""
+    # TODO: modern sqlalchemy uses dbapi_connection, remove the getattr check
+    # for "connection" once we drop support for sqlalchemy < 2.0
+    dbapi_conn = getattr(
+        connection, "dbapi_connection", getattr(connection, "connection", connection)
+    )
+    rollback_fn = getattr(dbapi_conn, "rollback", None)
+    if callable(rollback_fn):
+        rollback_fn()
+
+
+@contextmanager
+def _temporary_spanner_table(metadata, connection, table_name, *columns):
+    """Context manager managing an isolated Spanner table lifecycle.
+
+    Guarantees that:
+    1. Table creation DDL is committed immediately.
+    2. Any active read-write transactions are rolled back prior to DDL drop.
+    3. The table is dropped and its DDL committed.
+    4. The table reference is evicted from SQLAlchemy metadata even if drop fails.
+    """
+    t = Table(table_name, metadata, *columns)
+    try:
+        t.create(connection)
+        _commit_ddl(connection)
+        yield t
+    finally:
+        _rollback_txn(connection)
+        try:
+            t.drop(connection, checkfirst=True)
+            _commit_ddl(connection)
+        except Exception:
+            pass
+        finally:
+            metadata.remove(t)
+
+
+def _build_literal_select(table, type_, input_val, compare, support_whereclause):
+    """Build a SELECT statement for literal testing with optional WHERE clause."""
+    if not support_whereclause:
+        return table.select()
+
+    if compare:
+        return table.select().where(
+            table.c.x == literal(compare, type_),
+            table.c.x == literal(input_val, type_),
+        )
+
+    target_val = compare if compare is not None else input_val
+    return table.select().where(table.c.x == literal(target_val, type_))
+
+
 @pytest.fixture
 def literal_round_trip_spanner(metadata, connection):
-    # for literal, we test the literal render in an INSERT
-    # into a typed column.  we can then SELECT it back as its
-    # official type;
+    """Test literal SQL rendering in INSERT and SELECT with full table isolation.
+
+    In SQLAlchemy, literal rendering tests verify that Python data types (such as
+    Strings with special escape characters, Decimals, or Booleans) are correctly
+    rendered directly into raw SQL text (e.g., 'SELECT ... WHERE x = 15.7563')
+    without syntax corruption or parameter escaping failures.
+
+    To ensure test isolation in Cloud Spanner, this fixture:
+    1. Generates a unique table name per invocation using the column type and
+       UUID entropy.
+    2. Uses _temporary_spanner_table to manage table creation, transaction
+       rollback, drop, and metadata eviction cleanly.
+    3. Executes the insert and select assertions with _build_literal_select.
+    """
 
     def run(
         type_,
@@ -71,44 +148,24 @@ def run(
         compare=None,
         support_whereclause=True,
     ):
-        t = Table("t_literal_round_trip_spanner", metadata, Column("x", type_))
-        t.create(connection)
-
-        for value in input_:
-            ins = t.insert().values(x=literal(value, type_))
-            connection.execute(ins)
-
-        if support_whereclause:
-            if compare:
-                stmt = t.select().where(
-                    t.c.x
-                    == literal(
-                        compare,
-                        type_,
-                    ),
-                    t.c.x
-                    == literal(
-                        input_[0],
-                        type_,
-                    ),
-                )
-            else:
-                stmt = t.select().where(
-                    t.c.x
-                    == literal(
-                        compare if compare is not None else input_[0],
-                        type_,
-                    )
-                )
-        else:
-            stmt = t.select()
-
-        rows = connection.execute(stmt).all()
-        assert rows, "No rows returned"
-        for row in rows:
-            value = row[0]
-            if filter_ is not None:
-                value = filter_(value)
-            assert value in output
+        table_name = f"t_lit_{type_.__class__.__name__.lower()}_{uuid.uuid4().hex[:6]}"
+
+        with _temporary_spanner_table(
+            metadata, connection, table_name, Column("x", type_)
+        ) as t:
+            for value in input_:
+                ins = t.insert().values(x=literal(value, type_))
+                connection.execute(ins)
+
+            stmt = _build_literal_select(
+                t, type_, input_[0], compare, support_whereclause
+            )
+            rows = connection.execute(stmt).all()
+            assert rows, "No rows returned"
+            for row in rows:
+                value = row[0]
+                if filter_ is not None:
+                    value = filter_(value)
+                assert value in output
 
     return run