张胜宇
e239eb778b
1) 客服 Agent 四份交付文档 + 构建脚手架：品牌由包装占位 XX科技 / 旧名 南方财富 统一为南方基金（热线 400-889-8899 / 官网 nffund.com），系统名改为「智能服务系统」； 同步追加 §0.4 修订记录行，工程记录行保留原占位字面以支撑硬编码扫描验收。 2) 开发文档：清理 28 份已作废/残留文档（14 份移出归档 + 14 份仓库副本）， 新增《文档规整方案与开发前待决事项-2026-09-17》。 3) 客服agent 四份交付文档首次纳入本分支。
105 lines
4.8 KiB
Python
105 lines
4.8 KiB
Python
"""测试共享夹具，以及测试期的数据库连接池隔离。
|
||
|
||
为什么在这里替换引擎（必须在导入任何 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
|