qyqy
8cff6b35f9
问题：本机 `%TEMP%\pytest-of-Windows` 被权限更高的会话建过，当前用户无权写入， 于是所有用 `tmp_path` 的用例在 setup 阶段批量 ERROR（WinError 5）——实测 33 个， 散布在 offsite 附件预览、通知发送、promotion、worker 等处。这类噪声会让 "到底哪里坏了"完全看不出来（同事这批新用例首次把它暴露出来）。 修法：在 tests/conftest.py 覆盖 `tmp_path`，把根目录改到仓库内 `.workdir/pytest-tmp` （已在 .gitignore）。语义不变——每个用例仍拿到一个**新建的空目录**（残留目录会污染断言）。 只覆盖 `tmp_path` 而不动 `tmp_path_factory`：后者是 pytest 私有构造，参数随版本变化 （实测直接实例化 `TempPathFactory(...)` 会 TypeError），而本仓库用例只用 `tmp_path`。 同时把 pyproject.toml 里那条 `basetemp = ...` 删掉：pytest **只在命令行认 basetemp**， 写在 ini 里会被静默忽略（实测无效），留着会让人误以为已配好。改为注释指向 conftest。 效果：不带任何参数 `pytest -q` 从「3 failed + 33 errors」变为「3 failed，0 error」。
106 lines
4.8 KiB
Python
106 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 typing import TYPE_CHECKING
|
||
|
||
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
|