82 lines
2.8 KiB
Python
82 lines
2.8 KiB
Python
"""端到端验证：SSE 订阅 + 任务提交，确认网关主动推送 task_done。"""
|
||
import asyncio
|
||
import json
|
||
import uuid
|
||
|
||
import httpx
|
||
|
||
BASE = "http://127.0.0.1:8000"
|
||
AUTH = "gw_Hz8Qp3Km9f"
|
||
SESSION = f"e2e-{uuid.uuid4().hex[:8]}"
|
||
|
||
|
||
async def main() -> None:
|
||
async with httpx.AsyncClient(timeout=30) as client:
|
||
# 1) 提交任务（携带 cli_session_id），任务类型用 agent 已注册的能力
|
||
resp = await client.post(
|
||
f"{BASE}/api/cli/tasks",
|
||
json={
|
||
"auth": AUTH,
|
||
"task_type": "dev",
|
||
"task_tags": ["code", "dev"],
|
||
"description": "SSE 通知验证：创建临时文件",
|
||
"cli_session_id": SESSION,
|
||
"payload": {
|
||
"type": "create_files",
|
||
"files": [{"path": "_sse_probe.txt", "content": "hello-sse"}],
|
||
},
|
||
"timeout": 120,
|
||
},
|
||
)
|
||
print("submit status:", resp.status_code)
|
||
task = resp.json()
|
||
request_id = task["request_id"]
|
||
print("request_id:", request_id, "initial status:", task["status"])
|
||
|
||
# 2) 订阅 SSE，等待 task_done
|
||
got = []
|
||
|
||
async def subscribe():
|
||
try:
|
||
async with client.stream(
|
||
"GET",
|
||
f"{BASE}/api/cli/events",
|
||
params={"cli_session_id": SESSION, "auth": AUTH},
|
||
) as stream:
|
||
async for line in stream.aiter_lines():
|
||
if line.startswith("data: "):
|
||
evt = json.loads(line[6:])
|
||
got.append(evt)
|
||
print("SSE EVENT:", json.dumps(evt, ensure_ascii=False))
|
||
if evt.get("request_id") == request_id:
|
||
return
|
||
except Exception as exc: # noqa: BLE001
|
||
print("SSE error:", exc)
|
||
|
||
# 3) 并行：订阅 + 轮询兜底打印（仅观察，不作为结论）
|
||
await asyncio.gather(
|
||
subscribe(),
|
||
_poll_until_done(client, request_id),
|
||
)
|
||
if any(e.get("request_id") == request_id for e in got):
|
||
print("\nRESULT: PASS - SSE task_done received without polling")
|
||
else:
|
||
print("\nRESULT: FAIL - no SSE event for this request_id")
|
||
|
||
|
||
async def _poll_until_done(client, request_id: str) -> None:
|
||
for _ in range(60):
|
||
r = await client.get(f"{BASE}/api/cli/tasks/{request_id}")
|
||
if r.status_code == 404:
|
||
await asyncio.sleep(2)
|
||
continue
|
||
t = r.json()
|
||
print("poll status:", t["status"])
|
||
if t["status"] in ("success", "failed"):
|
||
return
|
||
await asyncio.sleep(2)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|