"""HTTP 层端到端验证脚本（临时工具，不属于 pytest 套件）。 覆盖：SSE 流式契约、响应头、落库、会话路由、中断/恢复闭环、调度手动触发。 运行：python scripts/http_e2e.py """ from __future__ import annotations import json import sys import time import httpx BASE = "http://127.0.0.1:8080" USER = "http_e2e" PASSED: list[str] = [] FAILED: list[str] = [] def check(name: str, ok: bool, detail: str = "") -> None: (PASSED if ok else FAILED).append(name) print(f" {'PASS' if ok else 'FAIL'} {name}" + (f" — {detail}" if detail else "")) def sse_events(response) -> list[dict]: """从 httpx 流式响应里解析出全部事件。""" events: list[dict] = [] for line in response.iter_lines(): if line.startswith("data: "): try: events.append(json.loads(line[6:])) except json.JSONDecodeError: pass return events def test_stream_basic() -> str | None: print("\n[1] POST /api/chat 流式契约") with httpx.Client(timeout=180) as client: with client.stream( "POST", f"{BASE}/api/chat", data={"message": "用一句话说明矿井通风的目的", "user_name": USER, "mode": "full"}, ) as res: headers = res.headers events = sse_events(res) check("HTTP 200", res.status_code == 200, str(res.status_code)) check("Content-Type 是 event-stream", "text/event-stream" in headers.get("content-type", "")) check("响应头 X-Session-Id", bool(headers.get("x-session-id"))) check("响应头 X-Accel-Buffering=no", headers.get("x-accel-buffering") == "no") check("响应头 Cache-Control=no-cache", "no-cache" in headers.get("cache-control", "")) check("响应头 X-Trace-Id", bool(headers.get("x-trace-id"))) types = [e["type"] for e in events] check("有 agent_start", "agent_start" in types) check("有 token", "token" in types) check("有 citations", "citations" in types) check("有 done", "done" in types) check("无 error", "error" not in types, str([e.get("message", "")[:60] for e in events if e["type"] == "error"])) check("done 在最后一个", types[-1] == "done", str(types[-3:])) # 每帧必须带 session_id check("每帧带 session_id", all(e.get("session_id") for e in events)) sid = events[0].get("session_id") if events else None return sid def test_persist_and_history(sid: str) -> None: print("\n[2] 落库与会话路由") with httpx.Client(timeout=60) as client: r = client.get(f"{BASE}/api/chat/history/{sid}", params={"user_name": USER}) check("历史 200", r.status_code == 200, str(r.status_code)) items = r.json().get("items", []) roles = [m["msg_role"] for m in items] check("历史有 user+assistant", roles[:2] == ["user", "assistant"], str(roles)) check("顺序正确（user 在前）", roles[0] == "user") assistant = next((m for m in items if m["msg_role"] == "assistant"), {}) check("assistant 内容非空", bool((assistant.get("content") or "").strip())) # 正文里不该混入工具结果 JSON body = assistant.get("content") or "" check("正文未混入工具结果 JSON", not body.strip().startswith('{"data"'), body[:80].replace("\n", " ")) r = client.get(f"{BASE}/api/sessions", params={"user_name": USER}) check("会话列表 200", r.status_code == 200) check("列表含该会话", any(s["id"] == sid for s in r.json().get("items", []))) r = client.get(f"{BASE}/api/sessions/{sid}/context", params={"user_name": USER}) check("上下文用量 200", r.status_code == 200) latest = (r.json() or {}).get("latest") or {} check("有用量记录", int(latest.get("total_tokens") or 0) > 0, str(latest.get("total_tokens"))) check("口径为 provider", latest.get("usage_source") == "provider", str(latest.get("usage_source"))) r = client.get(f"{BASE}/api/usage/today", params={"user_name": USER}) check("用量 200", r.status_code == 200) check("当日有消耗", int(r.json().get("total_tokens") or 0) > 0) r = client.get(f"{BASE}/api/model") check("模型信息 200", r.status_code == 200) d = r.json() check("工具数 = 4", d["agent"]["tools"]["count"] == 4, str(d["agent"]["tools"]["count"])) check("interrupt_on 含计划审批", "request_plan_approval" in d["agent"]["interrupt_on"], str(d["agent"]["interrupt_on"])) def test_404_and_errors() -> None: print("\n[3] 错误语义") with httpx.Client(timeout=60) as client: r = client.post(f"{BASE}/api/chat", data={ "message": "hi", "user_name": USER, "session_id": "not-exist-id", }) check("不存在的会话 → 404", r.status_code == 404, str(r.status_code)) r = client.post(f"{BASE}/api/chat", data={ "message": "hi", "user_name": "other_user", "session_id": "not-exist-id", }) check("他人会话 → 404（不泄露存在性）", r.status_code == 404, str(r.status_code)) r = client.get(f"{BASE}/api/chat/interrupt", params={"session_id": "not-exist-id", "user_name": USER}) check("中断查询 不存在的会话 → 404", r.status_code == 404, str(r.status_code)) def test_interrupt_loop() -> None: """plan 模式 → agent 调 request_plan_approval → interrupt → resume。""" print("\n[4] 中断/恢复闭环") with httpx.Client(timeout=240) as client: with client.stream( "POST", f"{BASE}/api/chat", data={ "message": "请制定一个三步计划：把一句话总结成简报并写到 /workspace/brief.md。先提交计划等我批准。", "user_name": USER, "mode": "plan", }, ) as res: events = sse_events(res) sid = None for e in events: sid = e.get("session_id") or sid types = [e["type"] for e in events] print(" 事件:", types) if "interrupt" not in types: check("触发 interrupt", False, f"agent 未请求审批；事件={types}") return check("触发 interrupt", True) intr = next(e for e in events if e["type"] == "interrupt") check("interrupt 带 tool_approval", bool(intr.get("tool_approval")), json.dumps(intr, ensure_ascii=False)[:150]) # 刷新场景：只读查询挂起的中断 r = client.get(f"{BASE}/api/chat/interrupt", params={"session_id": sid, "user_name": USER}) check("GET /chat/interrupt → 200", r.status_code == 200, str(r.status_code)) if r.status_code == 200: check("interrupts 非空", len(r.json().get("interrupts", [])) > 0) # 批准并续跑 r = client.post(f"{BASE}/api/chat/resume", json={ "session_id": sid, "thread_id": sid, "user_name": USER, "approved": True, "status": "answered", }) check("resume 200", r.status_code == 200, str(r.status_code)) if r.status_code == 200: texts = [] for line in r.iter_lines(): if line.startswith("data: "): try: texts.append(json.loads(line[6:])) except json.JSONDecodeError: pass rtypes = [e["type"] for e in texts] print(" resume 事件:", rtypes) check("恢复后有输出", "token" in rtypes or "tool_call" in rtypes, str(rtypes)) check("恢复后正常收尾", rtypes[-1] == "done" and "error" not in rtypes, str([e.get("message", "")[:60] for e in texts if e["type"] == "error"])) # 已恢复：再查应无挂起 r = client.get(f"{BASE}/api/chat/interrupt", params={"session_id": sid, "user_name": USER}) check("恢复后无挂起（204）", r.status_code == 204, str(r.status_code)) def test_scheduler() -> None: print("\n[5] 定时任务") with httpx.Client(timeout=240) as client: r = client.post(f"{BASE}/api/schedules", json={ "title": "e2e 任务", "cron": "0 9 * * *", "prompt": "用一句话说明今天要做什么（这是定时任务自检）。", "user_name": USER, }) check("创建任务 200", r.status_code == 200, r.text[:200]) if r.status_code != 200: return sid_task = r.json()["id"] r = client.post(f"{BASE}/api/schedules/{sid_task}/run") check("手动触发 200", r.status_code == 200, r.text[:250]) if r.status_code == 200: body = r.json() check("触发有输出", bool((body.get("output") or "").strip()), str(body.get("output"))[:80]) check("有独立会话", bool(body.get("session_id"))) check("记录了 token", int(body.get("input_tokens") or 0) > 0, str(body.get("input_tokens"))) r = client.get(f"{BASE}/api/schedules/{sid_task}/runs") check("执行记录 200", r.status_code == 200) runs = r.json().get("items", []) check("有成功记录", any(x["status"] == "success" for x in runs), str([x["status"] for x in runs])) r = client.post(f"{BASE}/api/schedules", json={ "title": "坏 cron", "cron": "not a cron", "prompt": "x", "user_name": USER, }) check("非法 cron → 400", r.status_code == 400, str(r.status_code)) r = client.post(f"{BASE}/api/schedules/{sid_task}/pause") check("暂停 200", r.status_code == 200 and r.json().get("enabled") is False) r = client.delete(f"{BASE}/api/schedules/{sid_task}") check("删除 200", r.status_code == 200 and r.json().get("deleted") is True) def main() -> int: started = time.time() print(f"目标 {BASE}") sid = test_stream_basic() if sid: test_persist_and_history(sid) test_404_and_errors() test_interrupt_loop() test_scheduler() print(f"\n{'=' * 56}") print(f"通过 {len(PASSED)} / {len(PASSED) + len(FAILED)} 耗时 {time.time() - started:.1f}s") if FAILED: print("失败项：") for name in FAILED: print(" -", name) return 1 print("全部通过") return 0 if __name__ == "__main__": sys.exit(main())