lzf_0626
f09ea9e988
NL 线（含其并入的袁聪场外/推广域）。唯一冲突是 .gitignore —— 双方都往同一区域加了 .workdir/，取对方版本（他的更完整，含 .tmp/ 与说明），顺带修掉我之前用 Add-Content -Encoding utf8 造成的编码混合（read 工具当时报 invalid UTF-8）。 合并后修的问题 —— 都不是"改别人业务逻辑"，是让门禁能绿： 1. 缺运行依赖 python-docx。document_parser.py 解析 .docx 用它，但 requirements.txt 与 pyproject.toml 都没声明 —— 别人环境跑知识入库会直接 ModuleNotFoundError: No module named 'docx'。已补声明。 2. ruff 7 项：其中 tests/conftest.py 的 F821 Undefined name 'Path'（他的 tmp_path 修复 写了字符串注解 "Path" 却漏 import，运行时不求值所以没炸，但 mypy/ruff 会抓）、 tools/publish_customer_service_config.py 的 F841 inherited_keys 死变量（他改同 key 覆盖、换成 inherited_only 后忘删旧的）、3 处 E501，另 2 项 ruff --fix 自动修复。 3. 合规基线种子未跑：integration 的 test_compliance_seed_mysql 4 个用例要求 agent_negative_word 有 7 条 active 且已复核、agent_reply_template 覆盖 6 场景。 跑 tools/seed_compliance_baseline.py（11 条 active 规则 / 6 个场景模板）后 80 passed。 验证：ruff 干净 / mypy 180 文件 0 错 / unit+contract 1140 passed / integration 80 passed / 表数 68（alembic 已在 20260911_merge_risk_heads）。 唯一失败 tests/unit/repository/test_fund_readonly_contract.py 是双方一致的既有缺陷： 它断言 Base.metadata 里的 fin_* 表集合，而实测为空集 —— 即该测试依赖别的测试先导入模型的 副作用，单独跑必失败。NL 方也明确"不修不报"，此处照办，仅记录。
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
|