93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
与 e2e_platform_capability_smoke.py 相同业务验证，但走 FastAPI TestClient（不经过 TCP 8037）。
|
|||
|
|
|
|||
|
|
适用：本机存在多个进程占用 8037、HTTP 命中旧实例导致 /platform 路由 404 时。
|
|||
|
|
|
|||
|
|
用法:
|
|||
|
|
cd backend && .\\venv\\Scripts\\python.exe scripts/e2e_platform_capability_smoke_testclient.py
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
|
|||
|
|
from fastapi.testclient import TestClient
|
|||
|
|
|
|||
|
|
# 确保加载 backend 目录
|
|||
|
|
def main() -> int:
|
|||
|
|
os.environ.setdefault("PLATFORM_BASE_URL", "http://testserver")
|
|||
|
|
from app.main import app
|
|||
|
|
|
|||
|
|
c = TestClient(app)
|
|||
|
|
r = c.post(
|
|||
|
|
"/api/v1/auth/login",
|
|||
|
|
data={"username": os.getenv("PLATFORM_USERNAME", "admin"), "password": os.getenv("PLATFORM_PASSWORD", "123456")},
|
|||
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|||
|
|
)
|
|||
|
|
if r.status_code != 200:
|
|||
|
|
print("login:", r.status_code, r.text[:500], file=sys.stderr)
|
|||
|
|
return 1
|
|||
|
|
token = r.json().get("access_token")
|
|||
|
|
h = {"Authorization": f"Bearer {token}"}
|
|||
|
|
|
|||
|
|
g = c.get("/api/v1/platform/scene-templates", headers=h)
|
|||
|
|
if g.status_code != 200:
|
|||
|
|
print("scene-templates:", g.status_code, g.text[:500], file=sys.stderr)
|
|||
|
|
return 2
|
|||
|
|
templates = g.json() or []
|
|||
|
|
tid = "template_customer_service"
|
|||
|
|
ids = [x.get("id") for x in templates if isinstance(x, dict)]
|
|||
|
|
if tid not in ids:
|
|||
|
|
tid = ids[0] if ids else None
|
|||
|
|
if not tid:
|
|||
|
|
print("无模板", file=sys.stderr)
|
|||
|
|
return 3
|
|||
|
|
|
|||
|
|
name = os.getenv("SMOKE_AGENT_NAME") or f"冒烟_客服DSL_TC_{int(time.time())}"
|
|||
|
|
cr = c.post(
|
|||
|
|
"/api/v1/platform/agents/from-template",
|
|||
|
|
headers=h,
|
|||
|
|
json={
|
|||
|
|
"template_id": tid,
|
|||
|
|
"name": name,
|
|||
|
|
"description": "e2e testclient",
|
|||
|
|
"parameters": {"temperature": 0.35},
|
|||
|
|
"budget_config": {"max_steps": 50, "max_llm_invocations": 5, "max_tool_calls": 20},
|
|||
|
|
},
|
|||
|
|
)
|
|||
|
|
if cr.status_code not in (200, 201):
|
|||
|
|
print("from-template:", cr.status_code, cr.text[:800], file=sys.stderr)
|
|||
|
|
return 4
|
|||
|
|
aid = cr.json().get("id")
|
|||
|
|
print("agent:", aid, cr.json().get("name"))
|
|||
|
|
|
|||
|
|
body = {
|
|||
|
|
"agent_id": aid,
|
|||
|
|
"input_data": {
|
|||
|
|
"query": os.getenv("SMOKE_USER_QUERY") or "你好，请用一句话说明你能帮我做什么。",
|
|||
|
|
"scenario_dsl": {
|
|||
|
|
"version": "1",
|
|||
|
|
"scene": "customer_service_smoke_tc",
|
|||
|
|
"goal": "验证 TestClient 执行",
|
|||
|
|
"constraints": ["简短"],
|
|||
|
|
"deliverables": ["一句回复"],
|
|||
|
|
"acceptance": ["有内容"],
|
|||
|
|
"payload": {"channel": "smoke_tc"},
|
|||
|
|
},
|
|||
|
|
},
|
|||
|
|
}
|
|||
|
|
er = c.post("/api/v1/executions", headers=h, json=body)
|
|||
|
|
if er.status_code not in (200, 201):
|
|||
|
|
print("executions:", er.status_code, er.text[:800], file=sys.stderr)
|
|||
|
|
return 5
|
|||
|
|
eid = er.json()["id"]
|
|||
|
|
print("execution:", eid, "(异步任务需 Celery；此处仅校验创建成功)")
|
|||
|
|
print("OK — TestClient 路径：模板 + DSL 创建执行记录成功。请用 Celery 跑完后查 GET /api/v1/executions/{id}")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main())
|