Use pytest fixtures for integration test teardown

Replace inline create_all/drop_all with a yielding engine fixture to
ensure tables are always cleaned up, even when tests fail mid-execution.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: antoineleclair

tests/integration/test_orm_operations.py

@@ -1,6 +1,7 @@
 """Integration tests for ORM operations."""
 
 import datetime
+from collections.abc import Generator
 
 import pytest
 from sqlalchemy import (
@@ -16,6 +17,7 @@
     create_engine,
     text,
 )
+from sqlalchemy.engine import Engine
 from sqlalchemy.orm import Session, declarative_base
 
 Base = declarative_base()
@@ -63,6 +65,14 @@ class DateTimeTest(Base):  # type: ignore[valid-type,misc]
 
 @pytest.mark.integration
 class TestORMOperations:
+    @pytest.fixture
+    def engine(self, engine_url: str) -> Generator[Engine]:
+        engine = create_engine(engine_url)
+        Base.metadata.create_all(engine)
+        yield engine
+        Base.metadata.drop_all(engine)
+        engine.dispose()
+
     def test_create_engine(self, engine_url: str) -> None:
         engine = create_engine(engine_url)
         assert engine is not None
@@ -79,108 +89,65 @@ def test_raw_sql(self, engine_url: str) -> None:
 
         engine.dispose()
 
-    def test_create_table_and_insert(self, engine_url: str) -> None:
-        engine = create_engine(engine_url)
-
-        # Create tables
-        Base.metadata.create_all(engine)
-
-        # Insert data
+    def test_create_table_and_insert(self, engine: Engine) -> None:
         with Session(engine) as session:
             user = User(name="Alice", email="alice@example.com")
             session.add(user)
             session.commit()
 
-            # Query data
             users = session.query(User).filter_by(name="Alice").all()
             assert len(users) == 1
             assert users[0].email == "alice@example.com"
 
-        # Cleanup
-        Base.metadata.drop_all(engine)
-        engine.dispose()
-
-    def test_unicode_text(self, engine_url: str) -> None:
+    def test_unicode_text(self, engine: Engine) -> None:
         """Test Unicode text handling including emojis, CJK, RTL."""
-        engine = create_engine(engine_url)
-        Base.metadata.create_all(engine)
-
         unicode_values = [
-            # Emojis (4-byte UTF-8)
-            "Hello 🎉 World",
-            "🎉🎊🎁🎂",
-            # CJK characters
-            "中文测试",
-            "日本語テスト",
-            "한국어 테스트",
-            # RTL languages
-            "العربية",
-            "עברית",
-            # Mixed scripts
-            "Hello 世界 🌍",
-            # Combining characters
-            "café résumé naïve",
+            "Hello \U0001f389 World",
+            "\U0001f389\U0001f38a\U0001f381\U0001f382",
+            "\u4e2d\u6587\u6d4b\u8bd5",
+            "\u65e5\u672c\u8a9e\u30c6\u30b9\u30c8",
+            "\ud55c\uad6d\uc5b4 \ud14c\uc2a4\ud2b8",
+            "\u0627\u0644\u0639\u0631\u0628\u064a\u0629",
+            "\u05e2\u05d1\u05e8\u05d9\u05ea",
+            "Hello \u4e16\u754c \U0001f30d",
+            "caf\u00e9 r\u00e9sum\u00e9 na\u00efve",
         ]
 
         with Session(engine) as session:
             for val in unicode_values:
-                # Insert
                 record = UnicodeTest(content=val)
                 session.add(record)
                 session.commit()
 
-                # Query and verify
                 result = session.query(UnicodeTest).filter_by(content=val).first()
                 assert result is not None, f"Failed to find: {repr(val)}"
                 assert result.content == val, f"Mismatch for: {repr(val)}"
 
-                # Cleanup
                 session.delete(result)
                 session.commit()
 
-        Base.metadata.drop_all(engine)
-        engine.dispose()
-
-    def test_binary_blob(self, engine_url: str) -> None:
+    def test_binary_blob(self, engine: Engine) -> None:
         """Test binary blob handling including null bytes."""
-        engine = create_engine(engine_url)
-        Base.metadata.create_all(engine)
-
         blob_values = [
             b"simple",
-            b"\x00\x01\x02\x03",  # Null bytes
-            b"\xff\xfe\xfd",  # High bytes
-            bytes(range(256)),  # All byte values
+            b"\x00\x01\x02\x03",
+            b"\xff\xfe\xfd",
+            bytes(range(256)),
         ]
 
         with Session(engine) as session:
             for val in blob_values:
-                # Insert
                 record = BlobTest(data=val)
                 session.add(record)
                 session.commit()
 
-                # Query and verify
                 result = session.query(BlobTest).order_by(BlobTest.id.desc()).first()
                 assert result is not None
                 assert result.data == val, f"Mismatch for blob: {repr(val)}"
 
-        Base.metadata.drop_all(engine)
-        engine.dispose()
-
-    def test_numeric_types(self, engine_url: str) -> None:
-        """Test integer, bigint, float, and boolean types.
-
-        Note: dqlite has a known limitation where BOOLEAN NULL values are
-        returned as False (type BOOLEAN with value 0) instead of NULL.
-        This is because dqlite returns the column's declared type even for
-        NULL values, and 0 is indistinguishable from NULL for BOOLEAN columns.
-        """
-        engine = create_engine(engine_url)
-        Base.metadata.create_all(engine)
-
+    def test_numeric_types(self, engine: Engine) -> None:
+        """Test integer, bigint, float, and boolean types."""
         test_cases = [
-            # (int, bigint, float, bool)
             (0, 0, 0.0, False),
             (1, 1, 1.0, True),
             (-1, -1, -1.0, False),
@@ -199,7 +166,6 @@ def test_numeric_types(self, engine_url: str) -> None:
                 session.add(record)
                 session.commit()
 
-                # Query and verify
                 result = session.query(NumericTest).order_by(NumericTest.id.desc()).first()
                 assert result is not None
                 assert result.int_val == int_val
@@ -208,56 +174,34 @@ def test_numeric_types(self, engine_url: str) -> None:
                     assert abs(result.float_val - float_val) < 1e-9
                 assert result.bool_val == bool_val
 
-        Base.metadata.drop_all(engine)
-        engine.dispose()
-
-    def test_datetime_types(self, engine_url: str) -> None:
-        """Test DateTime column type.
-
-        Note: dqlite has a known limitation where DATETIME NULL values are
-        returned as empty string instead of NULL. This causes SQLAlchemy's
-        DateTime processor to fail when parsing. Avoid using NULL datetime
-        values with dqlite - use a sentinel value if needed.
-        """
-        engine = create_engine(engine_url)
-        Base.metadata.create_all(engine)
-
+    def test_datetime_types(self, engine: Engine) -> None:
+        """Test DateTime column type."""
         test_dates = [
             datetime.datetime(2024, 1, 15, 10, 30, 45),
-            datetime.datetime(1970, 1, 1, 0, 0, 0),  # Unix epoch
-            datetime.datetime(2038, 1, 19, 3, 14, 7),  # Near Y2038
+            datetime.datetime(1970, 1, 1, 0, 0, 0),
+            datetime.datetime(2038, 1, 19, 3, 14, 7),
             datetime.datetime.now().replace(microsecond=0),
         ]
 
         with Session(engine) as session:
             for dt in test_dates:
-                # Note: We set updated_at to a value, not NULL, due to dqlite limitation
                 record = DateTimeTest(created_at=dt, updated_at=dt)
                 session.add(record)
                 session.commit()
 
-                # Query and verify
                 result = session.query(DateTimeTest).order_by(DateTimeTest.id.desc()).first()
                 assert result is not None
 
-                # SQLite stores datetime as text, compare with second precision
                 assert result.created_at.year == dt.year
                 assert result.created_at.month == dt.month
                 assert result.created_at.day == dt.day
                 assert result.created_at.hour == dt.hour
                 assert result.created_at.minute == dt.minute
                 assert result.created_at.second == dt.second
 
-        Base.metadata.drop_all(engine)
-        engine.dispose()
-
-    def test_null_handling(self, engine_url: str) -> None:
+    def test_null_handling(self, engine: Engine) -> None:
         """Test NULL values across different column types."""
-        engine = create_engine(engine_url)
-        Base.metadata.create_all(engine)
-
         with Session(engine) as session:
-            # Insert record with all nullable fields as NULL
             record = NumericTest(
                 int_val=None,
                 bigint_val=None,
@@ -273,6 +217,3 @@ def test_null_handling(self, engine_url: str) -> None:
             assert result.bigint_val is None
             assert result.float_val is None
             assert result.bool_val is None
-
-        Base.metadata.drop_all(engine)
-        engine.dispose()