fix(tests): resolve loop issues with per-test engine and clean up typing (formatted)

Author: hyhmrright

backend/app/core/limiter.py

@@ -1,6 +1,5 @@
 import functools
 import ipaddress
-import os
 
 import structlog
 from fastapi import Request
@@ -53,6 +52,4 @@ def _get_user_or_ip(request: Request) -> str:
     return get_trusted_client_ip(request) or "unknown"
 
 
-# In CI/Test environments, we disable the limiter entirely to prevent 429 errors.
-_enabled = not (os.getenv("CI") or os.getenv("PYTEST_CURRENT_TEST"))
-limiter = Limiter(key_func=_get_user_or_ip, enabled=_enabled)
+limiter = Limiter(key_func=_get_user_or_ip)

backend/app/db/session.py

@@ -1,57 +1,23 @@
 from collections.abc import AsyncGenerator, AsyncIterator
 from contextlib import asynccontextmanager
-from typing import Any
 
-from sqlalchemy.ext.asyncio import (
-    AsyncEngine,
-    AsyncSession,
-    async_sessionmaker,
-    create_async_engine,
-)
+from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
 from app.core.config import settings
 
-# Private globals for lazy initialization
-_engine: AsyncEngine | None = None
-_sessionmaker: async_sessionmaker[AsyncSession] | None = None
-
-
-def _get_engine() -> AsyncEngine:
-    """Internal helper to get or create the engine. Facilitates testing."""
-    global _engine
-    if _engine is None:
-        _engine = create_async_engine(settings.database_url, echo=False)
-    return _engine
-
-
-def _get_sessionmaker() -> async_sessionmaker[AsyncSession]:
-    """Internal helper to get or create the sessionmaker. Facilitates testing."""
-    global _sessionmaker
-    if _sessionmaker is None:
-        _sessionmaker = async_sessionmaker(_get_engine(), expire_on_commit=False)
-    return _sessionmaker
-
-
-def __getattr__(name: str) -> Any:
-    """Module-level getattr to support legacy engine and session access."""
-    if name == "engine":
-        return _get_engine()
-    if name == "AsyncSessionLocal":
-        return _get_sessionmaker()
-    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+engine = create_async_engine(settings.database_url, echo=False)
+AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
 
 
 async def get_db() -> AsyncGenerator[AsyncSession]:
-    session_factory = _get_sessionmaker()
-    async with session_factory() as session:
+    async with AsyncSessionLocal() as session:
         yield session
 
 
 @asynccontextmanager
 async def isolated_session() -> AsyncIterator[AsyncSession]:
-    """Isolated session manager for background tasks."""
-    session_factory = _get_sessionmaker()
-    async with session_factory() as session:
+    """Context manager for a one-off DB session."""
+    async with AsyncSessionLocal() as session:
         try:
             yield session
             await session.commit()

backend/app/worker.py

@@ -283,7 +283,7 @@ async def cleanup_old_executions(ctx: dict) -> None:
         await db.commit()
     logger.info(
         "job_executions_cleanup",
-        deleted=result.rowcount,
+        deleted=result.rowcount,  # type: ignore[attr-defined]
         retention_days=settings.cron_execution_retention_days,
     )
 
@@ -299,7 +299,7 @@ async def cleanup_old_deliveries(ctx: dict) -> None:
         await db.commit()
     logger.info(
         "webhook_deliveries_cleanup",
-        deleted=result.rowcount,
+        deleted=result.rowcount,  # type: ignore[attr-defined]
         retention_days=retention,
     )
 

backend/tests/conftest.py

@@ -6,25 +6,33 @@
 import pytest
 from sqlalchemy.pool import NullPool
 
-# --- GLOBAL MOCKS ---
+# --- GLOBAL INFRA MOCKS ---
 
-patch("slowapi.Limiter.limit", lambda *args, **kw: lambda f: f).start()
 patch("app.infra.qdrant.get_qdrant_client", AsyncMock()).start()
 patch("app.infra.minio.get_minio_client", MagicMock()).start()
 patch("redis.asyncio.Redis.from_url", return_value=AsyncMock()).start()
 patch("arq.create_pool", return_value=AsyncMock()).start()
 patch("app.scheduler.runner.start_scheduler", AsyncMock()).start()
 patch("app.scheduler.runner.stop_scheduler", AsyncMock()).start()
 
-# --- APP AND DB FIXTURES ---
+# --- APP FIXTURES ---
 
+from app.core.limiter import limiter
 from app.db.session import get_db
 from app.main import app
 
-# Disable lifespan
+# Disable app lifespan
 app.router.lifespan_context = MagicMock()
 
 
+@pytest.fixture(autouse=True)
+def disable_rate_limiting():
+    """测试期间禁用频率限制。"""
+    limiter.enabled = False
+    yield
+    limiter.enabled = True
+
+
 @pytest.fixture(scope="session")
 def anyio_backend():
     return "asyncio"
@@ -56,41 +64,41 @@ def setup_tables():
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
 
-@pytest.fixture(scope="session")
-async def engine(setup_tables):
+@pytest.fixture
+async def db_session(setup_tables):
+    """FRESH engine per test to eliminate loop issues."""
     try:
         _pw = os.environ["POSTGRES_PASSWORD"]
     except KeyError:
         raise RuntimeError("POSTGRES_PASSWORD is required") from None
 
     test_url = f"postgresql+asyncpg://jarvis:{_pw}@localhost:5432/jarvis_test"
     engine = create_async_engine(test_url, echo=False, poolclass=NullPool)
-    yield engine
-    await engine.dispose()
-
-
-@pytest.fixture
-async def db_session(engine):
-    async with engine.connect() as conn:
-        await conn.begin()
-        session = AsyncSession(
-            bind=conn, expire_on_commit=False, join_transaction_mode="create_savepoint"
-        )
-
-        m_iso = MagicMock()
-        m_iso.__aenter__ = AsyncMock(return_value=session)
-        m_iso.__aexit__ = AsyncMock(return_value=None)
-
-        # The key: Inject the CURRENT loop's session into the lazy getters
-        with (
-            patch("app.db.session._get_sessionmaker", return_value=lambda: session),
-            patch("app.db.session.isolated_session", return_value=m_iso),
-        ):
-            try:
-                yield session
-            finally:
-                await session.close()
-                await conn.rollback()
+    try:
+        async with engine.connect() as conn:
+            await conn.begin()
+            session = AsyncSession(
+                bind=conn,
+                expire_on_commit=False,
+                join_transaction_mode="create_savepoint",
+            )
+
+            m_iso = MagicMock()
+            m_iso.__aenter__ = AsyncMock(return_value=session)
+            m_iso.__aexit__ = AsyncMock(return_value=None)
+
+            # Inject this session into all possible targets
+            with (
+                patch("app.db.session.AsyncSessionLocal", return_value=session),
+                patch("app.db.session.isolated_session", return_value=m_iso),
+            ):
+                try:
+                    yield session
+                finally:
+                    await session.close()
+                    await conn.rollback()
+    finally:
+        await engine.dispose()
 
 
 @pytest.fixture