72 lines
2.9 KiB
Python
72 lines
2.9 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 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
|
|
)
|
|
|
|
import pytest # noqa: E402
|
|
|
|
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):
|
|
return review_output(result, context, config, memories)
|
|
|
|
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
|