fix(installer): enlist table retirement transaction

tests/test_module_system.py

@@ -20,7 +20,8 @@ from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import patch
 
-from sqlalchemy import Column, Integer, MetaData, Table, inspect
+from sqlalchemy import Column, Integer, MetaData, Table, create_engine, insert, inspect
+from sqlalchemy.orm import Session
 
 # Keep the default app import side effect from bootstrapping a development DB.
 _TEST_ROOT = Path(tempfile.mkdtemp(prefix="govoplan-module-tests-"))
@@ -2370,6 +2371,31 @@ finally:
         self.assertEqual("example", record["retirements"][0]["module_id"])
         self.assertIn("database_backup", record["snapshot"])
 
+    def test_drop_table_retirement_is_atomic_with_session_writes(self) -> None:
+        root = Path(tempfile.mkdtemp(prefix="govoplan-retirement-transaction-", dir=_TEST_ROOT))
+        engine = create_engine(
+            f"sqlite:///{root / 'retirement.db'}",
+            connect_args={"timeout": 0.1},
+        )
+        metadata = MetaData()
+        audit_table = Table("retirement_audit", metadata, Column("id", Integer, primary_key=True))
+        owned_table = Table("retirement_owned", metadata, Column("id", Integer, primary_key=True))
+        metadata.create_all(bind=engine)
+        owned_model = type("OwnedModel", (), {"__table__": owned_table})
+
+        with Session(engine) as session:
+            session.execute(insert(audit_table).values(id=1))
+            plan = drop_table_retirement_provider(owned_model, label="Owned")(session, "owned")
+            self.assertIsNotNone(plan.destroy_data_executor)
+            plan.destroy_data_executor(session, "owned")  # type: ignore[misc]
+            self.assertFalse(inspect(session.connection()).has_table("retirement_owned"))
+            session.rollback()
+
+        self.assertTrue(inspect(engine).has_table("retirement_owned"))
+        with engine.connect() as connection:
+            self.assertEqual([], connection.execute(audit_table.select()).all())
+        engine.dispose()
+
     def test_module_installer_dry_run_writes_run_record_without_applying(self) -> None:
         root = Path(tempfile.mkdtemp(prefix="govoplan-installer-dry-run-", dir=_TEST_ROOT))
         settings = _settings(root)