Testing: Streamline table reset procedure using SA's drop_all_tables

Author: amotl

tests/conftest.py

@@ -1,17 +1,13 @@
 import pytest
 import sqlalchemy as sa
+from sqlalchemy.testing.util import drop_all_tables
 
 
 @pytest.fixture
 def engine():
     engine = sa.create_engine("crate://")
-    with engine.connect() as conn:
-        # Clean up stale "read-only" mode states.
-        try:
-            conn.execute(sa.text('ALTER TABLE testdrive.foo RESET ("blocks.write");'))
-        except Exception:  # noqa: S110
-            pass
-        conn.execute(sa.text("DROP TABLE IF EXISTS testdrive.foo"))
-        conn.execute(sa.text("DROP TABLE IF EXISTS testdrive.foo_alter_tmp"))
+    with engine.connect() as connection:
+        inspector = sa.inspect(connection)
+        drop_all_tables(engine, inspector, schema="testdrive")
     yield engine
     engine.dispose()

tests/test_integration.py

@@ -7,6 +7,7 @@
 import pytest
 import sqlalchemy as sa
 from sqlalchemy.sql.type_api import UserDefinedType
+from sqlalchemy.testing.util import drop_all_tables
 from sqlalchemy_cratedb import ObjectType
 
 pytestmark = pytest.mark.sdktester
@@ -22,14 +23,8 @@ def run(command: str, background: bool = False):
 @pytest.fixture(autouse=True)
 def reset_tables(engine):
     with engine.connect() as connection:
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.all_types"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.all_types_alter_tmp"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.campaign"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.campaign_alter_tmp"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.composite_table"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.composite_table_alter_tmp"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.transaction"))
-        connection.execute(sa.text("DROP TABLE IF EXISTS tester.transaction_alter_tmp"))
+        inspector = sa.inspect(connection)
+        drop_all_tables(engine, inspector, schema="tester")
 
 
 @pytest.fixture()