320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""独立验收脚本：真实 HTTP + 真实 JWT + 真实 MySQL + 显式注册的测试 Agent。
|
||
|
||
覆盖 P0-1 越权拦截、P0-2 未注册拒绝、P0-3 端到端闭环、P1-3 SSE delta。
|
||
生产代码不注册业务 Agent，本脚本按 TODO 阶段 8 的约定显式注入测试注册表。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import datetime as dt
|
||
import sys
|
||
import uuid
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import jwt
|
||
from sqlalchemy import delete, select
|
||
|
||
import app.service.admin_service as admin_service
|
||
import app.service.agent_run_application_service as run_service
|
||
import app.service.public_platform_service as public_platform_service
|
||
from app.core.config import get_settings
|
||
from app.core.contracts import (
|
||
AgentDefinition,
|
||
AgentRequest,
|
||
CoreResult,
|
||
RequestContext,
|
||
ResolvedAgentConfig,
|
||
)
|
||
from app.infrastructure.db import SessionFactory
|
||
from app.main import create_app
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.conversation import ConversationMessage
|
||
from app.model.platform import AgentRun, DomainEventOutbox, OutboxDelivery, RequestIdempotency
|
||
from app.service.agent.base import BaseAgent
|
||
from app.service.agent.bootstrap import get_agent_factory
|
||
from app.service.agent.factory import AgentFactory
|
||
from app.worker.runtime import WorkerRuntime
|
||
|
||
# 密钥路径统一从配置读（.env 的 JWT_PRIVATE_KEY_PATH），换密钥只改配置，不用改脚本。
|
||
PRIVATE_KEY = Path(get_settings().jwt_private_key_path).read_text(encoding="utf-8")
|
||
CUSTOMER_ID = "9001" # tools/seed_test_rbac.py 造的数据
|
||
|
||
_results: list[tuple[str, str, str]] = []
|
||
|
||
|
||
def record(name: str, expected: str, actual: str) -> None:
|
||
_results.append(("PASS" if expected == actual else "FAIL", name, f"期望 {expected} / 实际 {actual}"))
|
||
|
||
|
||
def token(sub: str) -> str:
|
||
now = dt.datetime.now(dt.UTC)
|
||
return jwt.encode(
|
||
{
|
||
"sub": sub,
|
||
"iss": "jr-local",
|
||
"aud": "jr-agent-platform",
|
||
"exp": now + dt.timedelta(minutes=30),
|
||
"nbf": now - dt.timedelta(seconds=5),
|
||
"jti": str(uuid.uuid4()),
|
||
},
|
||
PRIVATE_KEY,
|
||
algorithm="RS256",
|
||
)
|
||
|
||
|
||
async def ensure_onboarding(client: httpx.AsyncClient, auth: dict[str, str]) -> None:
|
||
"""让独立验收客户满足业务接口的首次登录前置条件。"""
|
||
response = await client.get("/api/v1/onboarding/risk-questionnaire", headers=auth)
|
||
record("开户问卷状态可查询", "200", str(response.status_code))
|
||
if response.status_code != 200:
|
||
return
|
||
questionnaire = response.json().get("data") or {}
|
||
if not questionnaire.get("required"):
|
||
record("测试客户已完成开户问卷", "completed", "completed")
|
||
return
|
||
submission = {
|
||
"answers": {
|
||
"q1": 12, "q2": 4, "q3": 5, "q4": 4, "q5": 4, "q6": 5,
|
||
"q7": 5, "q8": 4, "q9": 4, "q10": 1, "q11": 5, "q12": 4,
|
||
"q13": 4,
|
||
},
|
||
"declaration_accepted": True,
|
||
}
|
||
response = await client.post(
|
||
"/api/v1/onboarding/risk-questionnaire/submissions",
|
||
json=submission,
|
||
headers={**auth, "Idempotency-Key": uuid.uuid4().hex},
|
||
)
|
||
record("首次登录问卷提交", "201", str(response.status_code))
|
||
|
||
|
||
class StubGovernance:
|
||
async def resolve(self, definition: AgentDefinition, context: RequestContext) -> ResolvedAgentConfig:
|
||
del definition, context
|
||
return ResolvedAgentConfig(
|
||
config_version="acceptance", prompt_version="acceptance", model_endpoint="stub"
|
||
)
|
||
|
||
async def recall(self, context: RequestContext) -> tuple[Any, ...]:
|
||
del context
|
||
return ()
|
||
|
||
async def review(
|
||
self, result: Any, context: RequestContext, config: Any, memories: Any,
|
||
*, agent_type: str = "",
|
||
) -> Any:
|
||
del context, config, memories, agent_type
|
||
return result
|
||
|
||
|
||
class AcceptanceAgent(BaseAgent):
|
||
async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
|
||
del request, context
|
||
return CoreResult(text="独立验收通过")
|
||
|
||
|
||
def register_probes(factory: AgentFactory, *, version: str = "acceptance") -> None:
|
||
"""向工厂注册两个探针 Agent（customer_service / risk）。
|
||
|
||
生产代码不注册业务 Agent（底座只提供框架），而验收需要走完整 HTTP 链路，
|
||
因此通过工厂的公开 `register` 接口注册——这与业务组员接入 Agent 的方式一致。
|
||
"""
|
||
for agent_type, roles in (
|
||
("customer_service", ("customer",)),
|
||
("risk", ("risk_operator",)),
|
||
):
|
||
definition = AgentDefinition(
|
||
agent_type=agent_type,
|
||
version=version,
|
||
allowed_roles=roles,
|
||
allowed_portals=("api",),
|
||
)
|
||
def builder(
|
||
_ctx: RequestContext, _def: AgentDefinition = definition
|
||
) -> BaseAgent:
|
||
return AcceptanceAgent(_def)
|
||
|
||
factory.register(definition, builder)
|
||
|
||
|
||
def build_registry() -> AgentFactory:
|
||
"""替身治理装配：供 HTTP 契约验收使用（**不验证生产装配**）。"""
|
||
factory = AgentFactory(StubGovernance())
|
||
register_probes(factory)
|
||
return factory
|
||
|
||
|
||
def production_factory() -> AgentFactory:
|
||
"""生产装配 + 探针 Agent，并**断言生产装配完整**。
|
||
|
||
与旧做法的关键差别：不再无条件把测试工厂塞进生产模块的全局名字——那样做会让
|
||
"生产装配到底可不可用"完全验收不到。这里直接用生产工厂，先断言模型服务、
|
||
工具执行器、意图分类器都已注入，再把探针 Agent 注册进去。
|
||
"""
|
||
factory = get_agent_factory()
|
||
missing = [
|
||
name
|
||
for name, value in (
|
||
("model_service", factory._model_service),
|
||
("tool_executor", factory._tool_executor),
|
||
("intent_classifier", factory._intent_classifier),
|
||
)
|
||
if value is None
|
||
]
|
||
if missing:
|
||
raise SystemExit(f"生产装配不完整，缺少：{missing}")
|
||
register_probes(factory, version="acceptance-production")
|
||
return factory
|
||
|
||
|
||
def inject_factory(factory: AgentFactory) -> None:
|
||
"""把替身工厂绑定到服务模块的全局名字——**仅限默认的 HTTP 契约验收模式**。
|
||
|
||
这是本脚本唯一的注入手段：Controller 直接构造 `AgentRunApplicationService(session)`
|
||
而不接收工厂，因此没有依赖注入点可用。生产装配的验证由 `--production` 模式与
|
||
`tools/memory_chain_probe.py` 承担，不会被这个注入掩盖。
|
||
"""
|
||
for module in (run_service, public_platform_service, admin_service):
|
||
if hasattr(module, "get_agent_factory"):
|
||
module.get_agent_factory = lambda: factory # type: ignore[attr-defined]
|
||
|
||
|
||
async def _collect_events(
|
||
client: httpx.AsyncClient, run_id: str, auth: dict[str, str]
|
||
) -> list[str]:
|
||
events: list[str] = []
|
||
async with client.stream(
|
||
"GET", f"/api/v1/agent-runs/{run_id}/events", headers=auth, timeout=30
|
||
) as stream:
|
||
async for line in stream.aiter_lines():
|
||
if line.startswith("event:"):
|
||
events.append(line.split(":", 1)[1].strip())
|
||
if events[-1] in {"done", "error"}:
|
||
break
|
||
return events
|
||
|
||
|
||
async def cleanup(session_id: str, run_id: str | None) -> None:
|
||
async with SessionFactory() as session, session.begin():
|
||
if run_id:
|
||
event_ids = select(DomainEventOutbox.event_id).where(
|
||
DomainEventOutbox.aggregate_id == run_id
|
||
)
|
||
await session.execute(
|
||
delete(OutboxDelivery).where(OutboxDelivery.event_id.in_(event_ids))
|
||
)
|
||
await session.execute(
|
||
delete(DomainEventOutbox).where(DomainEventOutbox.aggregate_id == run_id)
|
||
)
|
||
await session.execute(
|
||
delete(InteractionAudit).where(InteractionAudit.session_id == session_id)
|
||
)
|
||
await session.execute(delete(AgentRun).where(AgentRun.session_id == session_id))
|
||
await session.execute(
|
||
delete(RequestIdempotency).where(RequestIdempotency.session_id == session_id)
|
||
)
|
||
await session.execute(
|
||
delete(ConversationMessage).where(ConversationMessage.session_id == session_id)
|
||
)
|
||
|
||
|
||
async def main() -> None:
|
||
if "--production" in sys.argv:
|
||
factory = production_factory()
|
||
print("模式：生产装配验收（真实治理链 + 探针 Agent）")
|
||
else:
|
||
factory = build_registry()
|
||
inject_factory(factory)
|
||
print("模式：HTTP 契约验收（替身治理）——未验证生产装配；加 --production 验生产链路")
|
||
|
||
app = create_app()
|
||
session_id = f"acceptance-{uuid.uuid4()}"
|
||
run_id: str | None = None
|
||
try:
|
||
async with httpx.AsyncClient(
|
||
transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=30
|
||
) as client:
|
||
auth = {"Authorization": f"Bearer {token(CUSTOMER_ID)}"}
|
||
await ensure_onboarding(client, auth)
|
||
|
||
response = await client.post(
|
||
"/api/v1/agent-runs",
|
||
json={
|
||
"agent_type": "customer_service",
|
||
"message": "独立验收消息",
|
||
"session_id": session_id,
|
||
"idempotency_key": uuid.uuid4().hex,
|
||
},
|
||
headers=auth,
|
||
)
|
||
record("客户调用已授权 Agent", "202", str(response.status_code))
|
||
if response.status_code == 202:
|
||
run_id = response.json().get("data", {}).get("run_id")
|
||
|
||
response = await client.post(
|
||
"/api/v1/agent-runs",
|
||
json={
|
||
"agent_type": "risk",
|
||
"message": "越权尝试",
|
||
"session_id": session_id,
|
||
"idempotency_key": uuid.uuid4().hex,
|
||
},
|
||
headers=auth,
|
||
)
|
||
record("客户越权调用 risk Agent", "403", str(response.status_code))
|
||
|
||
response = await client.post(
|
||
"/api/v1/agent-runs",
|
||
json={
|
||
"agent_type": "no_such_agent",
|
||
"message": "x",
|
||
"session_id": session_id,
|
||
"idempotency_key": uuid.uuid4().hex,
|
||
},
|
||
headers=auth,
|
||
)
|
||
record("未注册 agent_type", "404", str(response.status_code))
|
||
|
||
if run_id:
|
||
runtime = WorkerRuntime(factory)
|
||
await runtime.dispatch_one(run_id=run_id)
|
||
|
||
# 先建立 SSE 订阅（此时 run 仍为 queued），再执行，才能观察到 delta 分块
|
||
subscription = asyncio.create_task(_collect_events(client, run_id, auth))
|
||
await asyncio.sleep(1.0)
|
||
await runtime.execute(run_id)
|
||
events = await subscription
|
||
print(f"SSE events: {events}")
|
||
|
||
record(
|
||
"SSE 含内容事件", "yes",
|
||
"yes" if {"delta", "replace"}.intersection(events) else "no",
|
||
)
|
||
record("SSE 终止事件", "done", events[-1] if events else "无事件")
|
||
|
||
response = await client.get(f"/api/v1/agent-runs/{run_id}", headers=auth)
|
||
# 文档 §3.3：单资源成功响应是 {data, meta} 信封。
|
||
body = response.json().get("data") or {}
|
||
record("运行终态", "succeeded", str(body.get("status")))
|
||
record(
|
||
"运行结果内容",
|
||
"独立验收通过",
|
||
str((body.get("result") or {}).get("content")),
|
||
)
|
||
finally:
|
||
await cleanup(session_id, run_id)
|
||
|
||
width = max(len(name) for _, name, _ in _results)
|
||
print()
|
||
for verdict, name, detail in _results:
|
||
print(f"[{verdict}] {name.ljust(width)} {detail}")
|
||
failed = sum(1 for verdict, _, _ in _results if verdict == "FAIL")
|
||
print(f"\n合计 {len(_results)} 项，失败 {failed} 项")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|