group_fqcd_jr/tests/conftest.py

"""测试共享夹具，以及测试期的数据库连接池隔离。

为什么在这里替换引擎（必须在导入任何 app 模块之前执行）：

`app/infrastructure/db.py` 在导入时就建好了全局异步引擎，连接池里的 MySQL 连接绑定在
"建立它的那个事件循环"上。测试环境里同时存在多个循环——pytest-asyncio 每个用例一个、
每个 `TestClient` 一个 portal 线程循环、用例内 `asyncio.run(...)` 再建一个。连接被跨循环
复用时 SQLAlchemy/asyncmy 抛
`got Future <Future pending> attached to a different loop`，失败集合随执行顺序变化，
真实缺陷被噪声掩盖（同一用例全量跑失败、单条跑通过）。

NullPool 让每次取用都新建连接、归还即关闭，从根上消除跨循环复用。生产路径不受影响：
`app/infrastructure/db.py` 未改动，仍使用默认连接池。
"""

import asyncio
import sys
from pathlib import Path

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

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

from app.infrastructure import db as _db  # noqa: E402

_db.engine = create_async_engine(
    _db.engine.url, pool_pre_ping=True, poolclass=NullPool
)
_db.SessionFactory = async_sessionmaker(  # type: ignore[assignment]
    _db.engine, class_=AsyncSession, expire_on_commit=False
)

@pytest.fixture
def tmp_path() -> Path:
    """把 `tmp_path` 的根目录改到**仓库内**（`.workdir/pytest-tmp`）。

    为什么必须覆盖：pytest 默认在系统临时目录建 `%TEMP%\\pytest-of-<user>`。Windows 上
    那个目录一旦被**权限更高的会话**（例如以管理员身份跑过一次测试）建过，当前用户就
    无权再写入，于是**所有**用 `tmp_path` 的用例在 setup 阶段批量失败
    （实测 33 个用例 ERROR、报 `PermissionError: [WinError 5]`），而它们与真实缺陷无关——
    这种噪声会让"到底哪里坏了"完全看不出来。

    覆盖后临时目录落在仓库内：不依赖系统临时目录权限，随 `.workdir/` 一起被 .gitignore
    忽略，清理范围可见。语义不变——每个用例拿到一个**新建的空目录**（残留目录会让上一条
    用例的产物污染断言）。

    为什么只覆盖 `tmp_path` 而不动 `tmp_path_factory`：后者是 pytest 的私有构造
    （`TempPathFactory.__init__` 的参数随版本变化，实测直接实例化会 `TypeError`），
    而本仓库的用例只用 `tmp_path`。改动面越小，越不容易在下一次升级时炸。
    """
    import shutil
    from pathlib import Path

    root = Path(".workdir") / "pytest-tmp"
    root.mkdir(parents=True, exist_ok=True)
    # 用目录数量推序号：不依赖 pytest 内部状态，重跑时自动接着编号。
    index = len([item for item in root.iterdir() if item.is_dir()]) + 1
    path = root / f"test{index}"
    shutil.rmtree(path, ignore_errors=True)
    path.mkdir(parents=True)
    return path


from app.core.contracts import AgentDefinition, CoreResult, ResolvedAgentConfig  # noqa: E402
from app.service.agent.base import BaseAgent  # noqa: E402
from app.service.agent.factory import AgentFactory  # noqa: E402
from app.service.agent.governance import review_output  # noqa: E402


@pytest.fixture
def governance():
    class TestGovernance:
        async def resolve(self, definition, context):
            return ResolvedAgentConfig(config_version="test", prompt_version="test",
                                       model_endpoint="test")

        async def recall(self, context):
            return ()

        async def review(self, result, context, config, memories, *, agent_type: str = ""):
            # 透传 `agent_type`：门禁 F5（面向客户输出 100% 附固定话术）由它判定，
            # 替身若不转发，`test_worker_runtime_mysql` 那条端到端断言会拿不到免责声明。
            return review_output(result, context, config, memories, agent_type=agent_type)

    return TestGovernance()


@pytest.fixture
def acceptance_registry(monkeypatch, governance):
    """Explicit test registration; production does not silently create business Agents."""
    class TestAgent(BaseAgent):
        async def handle(self, request, context):
            return CoreResult(text="test result")

    definition = AgentDefinition(agent_type="customer_service", version="test",
                                 allowed_roles=("customer",), allowed_portals=("api",))
    registry = AgentFactory(governance)
    registry.register(definition, lambda _: TestAgent(definition))
    monkeypatch.setattr("app.service.agent_run_application_service.get_agent_factory",
                        lambda: registry)
    return registry