dev_collab_e2e.py

#!/usr/bin/env python3
"""dev-collab E2E：通过业务应用 SSE 代理调用 DSH Agent，验证 6 个工具全链路。"""
import json, subprocess, sys

AGENT = "dev-collab"
URL = "http://127.0.0.1:18141/api/assistant/stream"

CASES = [
    ("T1 MR列表", "列出全部待修改状态的 MR，简洁回答，不要展开讲", ["MR-302", "MR-306", "待修改"]),
    ("T2 MR评审意见", "MR-302 的评审意见共识和分歧是什么？为什么卡住？简洁回答", ["MR-302", "拒绝策略", "DBA"]),
    ("T3 接口影响面", "改支付回调接口 /api/pay/notify 影响面有哪些调用方？简洁回答", ["微信支付", "云闪付", "调用"]),
    ("T4 技术债排序", "技术债按风险排个序，高风险有哪些？给偿还计划，简洁回答", ["支付回调", "幂等", "偿还"]),
    ("T5 接口检索", "查一下订单列表接口的定义和入参出参，简洁回答", ["groupByShop", "userId"]),
    ("T6 工作台总览", "研发协作总览怎么样？阻塞的 MR 有几个？简洁回答", ["阻塞", "MR"]),
]

def ask(message, timeout=170):
    payload = json.dumps({"message": message}, ensure_ascii=False)
    try:
        out = subprocess.run(
            ["curl", "-s", "--noproxy", "*", "-N", "-X", "POST", URL,
             "-H", "Content-Type: application/json", "-d", payload,
             "--max-time", str(timeout)],
            capture_output=True, text=True, timeout=timeout + 10).stdout
    except Exception as e:
        return "", f"curl 异常: {e}"
    text = []
    ev = ""
    for line in out.splitlines():
        line = line.rstrip("\r")
        if line.startswith("event:"):
            ev = line[6:].strip()
        elif line.startswith("data:"):
            s = line[5:].strip()
            if not s or s == "[DONE]" or ev != "chunk":
                continue
            try:
                j = json.loads(s)
                c = j.get("content", "")
                if c:
                    text.append(c)
            except Exception:
                pass
            ev = ""
    return "".join(text), out

def main():
    passed, failed = 0, []
    for name, q, keys in CASES:
        reply, raw = ask(q)
        ok = all(k in reply for k in keys)
        print(f"[{'PASS' if ok else 'FAIL'}] {name}\n  Q: {q}\n  A: {reply[:220]}")
        if ok:
            passed += 1
        else:
            failed.append(name)
            if not reply:
                print(f"  raw 首行: {raw.splitlines()[:3] if raw else '(空)'}")
    print(f"\n===== dev-collab E2E: {passed}/{len(CASES)} PASS =====")
    if failed:
        print("失败用例:", ", ".join(failed))
        sys.exit(1)

if __name__ == "__main__":
    main()