diff --git a/packages/sqlalchemy-spanner/tests/conftest.py b/packages/sqlalchemy-spanner/tests/conftest.py index ec16ad532dde..0aebf344b275 100644 --- a/packages/sqlalchemy-spanner/tests/conftest.py +++ b/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