ai-kefu-admin/tests/conftest.py

import asyncio
import os
import shutil
import tempfile
import warnings
from pathlib import Path

import pytest
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.engine.models import Base

# Task 5b 装配切换: app 不再有全局 engine/async_session——旧 conftest 里
# "先 import app.main 再 import chat_rest"的循环导入次序坑随全局单例一并
# 消失(routes 不再反向 import app.main),直接导入即可。
from app.main import app
from app.platform.registry import TenantRegistry

REPO_ROOT = Path(__file__).resolve().parents[1]
SEED_DIR = REPO_ROOT / "seed_profiles"


async def _create_all(engine):
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@pytest.fixture
def async_db():
    """Fresh temp-file SQLite sessionmaker for direct async-DB tests
    (e.g. testing ConversationManager without HTTP). Isolated per test.

    刻意不走 Registry 引擎: 本夹具被各测试反复 asyncio.run(每次全新事件
    循环),NullPool 连接即借即还才能跨循环安全;Registry 的 pool=2 常驻
    连接与创建它的循环绑定,只能活在单一事件循环里(client 夹具的
    TestClient 循环,见下)。
    """
    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.engine.services.rate_limit import _reset_for_test
    _reset_for_test()
    yield
    _reset_for_test()


@pytest.fixture
def client(tmp_path):
    """TestClient + 每测试独立 Registry(tmp_path 作 data_dir)。

    - 预置 app.state.registry(测试注入缝): lifespan 见 registry 已挂则
      沿用,不再加载仓库 seed_profiles/tenants——每测试拿到全新租户库,
      与生产数据互不污染。
    - 默认租户(jiancai)的 engine 由 lifespan 在 TestClient 的 portal 事件
      循环内预热,后续请求也在同一循环处理(连接/循环绑定安全)。测试
      线程需要直接造数时不得用该 engine——asyncio.run 另起循环,跨循环
      复用常驻连接会死锁;直连断言走 async_db,或经真实路由(WS/REST)。
    - 退出: lifespan finally 里 registry.dispose_all()(Windows 文件锁,
      不关则 tmp 目录删不掉);随后置空 app.state.registry(关残留窗——
      不让已 dispose 的测试 Registry 被后续无夹具的 lifespan 误用)并
      清理 tmp 目录(删不掉告警留痕,不静默吞)。
    """
    from fastapi.testclient import TestClient

    registry = TenantRegistry.load_from_dir(SEED_DIR, data_dir=tmp_path)
    app.state.registry = registry
    try:
        with TestClient(app) as c:
            yield c
    finally:
        app.state.registry = None
        try:
            shutil.rmtree(tmp_path)
        except OSError as exc:
            warnings.warn(f"client 夹具清理 tmp 目录失败(文件锁?): {exc}")
            shutil.rmtree(tmp_path, ignore_errors=True)