fix: drop worker database with force to handle active connections (#304)

src/fastapi_toolsets/pytest/utils.py

@@ -125,13 +125,17 @@ async def create_worker_database(
     engine = create_async_engine(_server_url, isolation_level="AUTOCOMMIT")
     try:
         async with engine.connect() as conn:
-            await conn.execute(text(f"DROP DATABASE IF EXISTS {worker_db_name}"))
+            await conn.execute(
+                text(f"DROP DATABASE IF EXISTS {worker_db_name} WITH (FORCE)")
+            )
         await create_database(db_name=worker_db_name, server_url=_server_url)
 
         yield worker_url
 
         async with engine.connect() as conn:
-            await conn.execute(text(f"DROP DATABASE IF EXISTS {worker_db_name}"))
+            await conn.execute(
+                text(f"DROP DATABASE IF EXISTS {worker_db_name} WITH (FORCE)")
+            )
     finally:
         await engine.dispose()
 

tests/test_pytest.py

@@ -626,6 +626,35 @@ class TestCreateWorkerDatabase:
                 assert result.scalar() == 1
             await engine.dispose()
 
+    @pytest.mark.anyio
+    async def test_drops_database_with_active_connections(
+        self, monkeypatch: pytest.MonkeyPatch
+    ):
+        """DROP DATABASE succeeds even when a connection is still open to it."""
+        monkeypatch.setenv("PYTEST_XDIST_WORKER", "gw_active_conn")
+        expected_db = make_url(
+            worker_database_url(DATABASE_URL, default_test_db="unused")
+        ).database
+
+        lingering_engine = None
+        async with create_worker_database(DATABASE_URL) as url:
+            # Open a connection to the worker DB and intentionally leave it open.
+            lingering_engine = create_async_engine(url)
+            async with lingering_engine.connect():
+                pass  # connection returned to pool but engine not disposed
+
+        # If WITH (FORCE) is absent the DROP above would raise; reaching here means it worked.
+        engine = create_async_engine(DATABASE_URL, isolation_level="AUTOCOMMIT")
+        async with engine.connect() as conn:
+            result = await conn.execute(
+                text("SELECT 1 FROM pg_database WHERE datname = :name"),
+                {"name": expected_db},
+            )
+            assert result.scalar() is None
+        await engine.dispose()
+        if lingering_engine:
+            await lingering_engine.dispose()
+
 
 class _LocalBase(DeclarativeBase):
     pass