浅念
121 lines
4.0 KiB
Python
121 lines
4.0 KiB
Python
import os
|
|
import tempfile
|
|
import asyncio
|
|
|
|
import pytest
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from app.models.database import Base
|
|
|
|
# `chat_rest` is created in Task A4. Tolerate its absence so earlier tasks
|
|
# (e.g. A0's /health smoke) can use the `client` fixture without it.
|
|
# Import `app.main` FIRST so the natural main -> chat_rest load order resolves
|
|
# (chat_rest does `from app.main import async_session`; if conftest imports
|
|
# chat_rest before main is loaded, main's `from chat_rest import router` hits
|
|
# a partially-initialized module -> circular ImportError).
|
|
_HAS_CHAT_REST = False
|
|
chat_rest_mod = None
|
|
try:
|
|
import app.main # noqa: F401
|
|
import app.api.routes.chat_rest as chat_rest_mod
|
|
_HAS_CHAT_REST = True
|
|
except ModuleNotFoundError:
|
|
pass
|
|
|
|
|
|
async def _create_all(engine):
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def _dispose_main_engine():
|
|
"""Dispose `app.main.engine` at session end.
|
|
|
|
Best-effort cleanup: explicitly disposing the app-level engine before
|
|
interpreter shutdown reduces (but does NOT eliminate) the noisy
|
|
`Windows fatal exception: access violation` traces that aiosqlite worker
|
|
threads emit when interrupted mid-cleanup at interpreter exit.
|
|
|
|
Reality check on Windows:
|
|
- All tests pass and pytest exits 0; the traces are written to stderr
|
|
AFTER the run completes and do not affect pass/fail or the exit code.
|
|
- They originate from aiosqlite background threads being torn down by the
|
|
interpreter and are harmless.
|
|
- They do not occur on Linux/CI.
|
|
"""
|
|
yield
|
|
from app.main import engine as main_engine
|
|
try:
|
|
asyncio.run(main_engine.dispose())
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def async_db():
|
|
"""Fresh temp-file SQLite sessionmaker for direct async-DB tests
|
|
(e.g. testing ConversationManager without HTTP). Isolated per test.
|
|
"""
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
engine = create_async_engine(
|
|
f"sqlite+aiosqlite:///{path}", poolclass=NullPool
|
|
)
|
|
asyncio.run(_create_all(engine))
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
yield sm
|
|
asyncio.run(engine.dispose())
|
|
try:
|
|
os.remove(path)
|
|
except PermissionError:
|
|
pass
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _reset_rate_limit():
|
|
"""Reset the in-memory rate limiter before and after every test so that
|
|
module-global bucket state can never leak across tests."""
|
|
from app.services.rate_limit import _reset_for_test
|
|
_reset_for_test()
|
|
yield
|
|
_reset_for_test()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch):
|
|
"""Temporary file-SQLite TestClient with `chat_rest.async_session` overridden.
|
|
|
|
Design notes:
|
|
- File DB (not :memory:) avoids the StaticPool cross-loop reuse pitfall.
|
|
- `NullPool` closes each connection when the session returns it, so no
|
|
aiosqlite worker thread outlives the event loop that created it.
|
|
Together with `_dispose_main_engine`, this REDUCES but does NOT
|
|
eliminate the harmless Windows access-violation stderr traces described
|
|
above; it has no effect on test correctness or on Linux/CI.
|
|
- Monkeypatching the module-level `async_session` binding in `chat_rest`
|
|
works because the route handlers resolve it as a module global at call.
|
|
"""
|
|
from fastapi.testclient import TestClient
|
|
from app.main import app
|
|
|
|
fd, path = tempfile.mkstemp(suffix=".db")
|
|
os.close(fd)
|
|
engine = create_async_engine(
|
|
f"sqlite+aiosqlite:///{path}", poolclass=NullPool
|
|
)
|
|
asyncio.run(_create_all(engine))
|
|
sm = async_sessionmaker(engine, expire_on_commit=False)
|
|
if _HAS_CHAT_REST:
|
|
monkeypatch.setattr(chat_rest_mod, "async_session", sm)
|
|
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
asyncio.run(engine.dispose())
|
|
try:
|
|
os.remove(path)
|
|
except PermissionError:
|
|
pass
|