cfc/cfc-langgraph/conftest.py

"""Root-level conftest to patch ctranslate2 before tests import app.main."""
import sys


class _MockModels:
    class WhisperGenerationResult:
        pass


class _MockCTranslate2:
    StorageView = type("StorageView", (), {})
    models = _MockModels()


# Install mock before any real import of ctranslate2
if "ctranslate2" not in sys.modules:
    sys.modules["ctranslate2"] = _MockCTranslate2()


import pytest_asyncio  # noqa: E402


@pytest_asyncio.fixture
async def init_checkpointer():
    """为测试初始化 LangGraph checkpointer（等价于 app startup 中的 init_checkpointer()）。

    pytest（httpx ASGITransport）不触发 FastAPI startup，导致依赖
    get_checkpointer() 的图（chat/health_coach）抛 "checkpointer 未初始化"。

    pytest-asyncio strict 模式每个测试独立事件循环，而 chat.py 模块级缓存图在
    编译期绑定 saver 单例。因此 teardown 必须：① 关闭连接并重置 saver 单例；
    ② 清空 chat._graph 缓存，让下一个测试在自己的 loop 内重建图，
    否则跨测试复用旧 loop 的 saver 会报 "no active connection"。
    """
    import app.api.chat as chat_mod
    from app import checkpointer as cp

    await cp.init_checkpointer()
    yield
    if cp._saver is not None:
        closer = getattr(cp._saver, "aclose", None)
        if closer is not None:
            try:
                await closer()
            except Exception:
                pass
        conn = getattr(cp._saver, "conn", None)
        if conn is not None:
            try:
                await conn.close()
            except Exception:
                pass
    cp._saver = None
    chat_mod._graph = None