json_surgeon_e2e.py

#!/usr/bin/env python3
"""JSON Surgeon E2E — dsh-java-json (port 8101)

Cases:
  T0 REST stats：结构统计
  T1 REST query：JSONPath 提取（取下标 + 递归 + 通配）
  T2 REST schema：结构推断
  T3 REST diff：深度比对
  T4 REST flatten：展平
  T5 非法 JSON 报错 + 非法 path 报错
  T6 AI 走 SSE 调用 json_* 工具
  T7 安全红线：AI 拒绝臆造未提供的 JSON
"""
import io
import json
import sys
import time
import urllib.request
import urllib.error

OUT = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")

BASE = "http://127.0.0.1:8101"
PROXY_BYPASS = urllib.request.ProxyHandler({})
OPENER = urllib.request.build_opener(PROXY_BYPASS)

PASSED = []
FAILED = []

SAMPLE_A = '{"app":"demo","version":"1.0","users":[{"id":1,"name":"lee","tags":["a","b"]},{"id":2,"name":"wang","tags":[]}],"meta":{"env":"prod","replicas":3}}'
SAMPLE_B = '{"app":"demo","version":"1.1","users":[{"id":1,"name":"lee","tags":["a","b","c"]},{"id":2,"name":"zhao"}],"meta":{"env":"prod","replicas":5,"region":"cn-north"}}'


def req(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(BASE + path, data=data, method=method,
                               headers={"Content-Type": "application/json"})
    try:
        with OPENER.open(r, timeout=180) as resp:
            return resp.status, json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        try:
            return e.code, json.loads(e.read().decode())
        except Exception:
            return e.code, {}


def check(name, cond, detail=""):
    if cond:
        PASSED.append(name)
        OUT.write(f"  PASS {name} {detail}\n")
    else:
        FAILED.append(name)
        OUT.write(f"  FAIL {name} {detail}\n")


def sse(agent_id, message, max_seconds=150):
    """POST /api/assistant/stream, 读取 SSE 直到 finish/done。返回 (全文, 工具名列表)。"""
    body = json.dumps({"agentId": agent_id, "message": message}).encode()
    r = urllib.request.Request(BASE + "/api/assistant/stream", data=body, method="POST",
                               headers={"Content-Type": "application/json"})
    text, tools = "", []
    start = time.time()
    try:
        with OPENER.open(r, timeout=max_seconds) as resp:
            event = "message"
            raw = b""
            while time.time() - start < max_seconds:
                chunk = resp.read(256)
                if not chunk:
                    break
                raw += chunk
                while b"\n" in raw:
                    raw_line, raw = raw.split(b"\n", 1)
                    line = raw_line.decode("utf-8", "replace").strip()
                    if line.startswith("event:"):
                        event = line[6:].strip()
                    elif line.startswith("data:"):
                        try:
                            payload = json.loads(line[5:].strip())
                        except Exception:
                            continue
                        if event == "chunk":
                            text += payload.get("content", "")
                        elif event == "step_break":
                            tn = payload.get("toolName", "")
                            if tn:
                                tools.append(tn)
                        elif event in ("finish", "done"):
                            return text, tools
    except Exception as e:
        OUT.write(f"  [sse error] {e}\n")
    return text, tools


def main():
    OUT.write("== T0 stats ==\n")
    st, data = req("POST", "/api/json/stats", {"json": SAMPLE_A})
    check("T0 status", st == 200, f"st={st}")
    check("T0 rootType object", data.get("rootType") == "object")
    check("T0 nodeCount>0", data.get("nodeCount", 0) > 10, f"n={data.get('nodeCount')}")
    check("T0 maxDepth>=4", data.get("maxDepth", 0) >= 4, f"d={data.get('maxDepth')}")
    types = data.get("typeDistribution", {})
    check("T0 array in types", "array" in types, f"types={types}")

    OUT.write("== T1 query ==\n")
    st, data = req("POST", "/api/json/query", {"json": SAMPLE_A, "path": "$.users[0].name"})
    check("T1 index hit", st == 200 and data.get("hitCount") == 1
          and data.get("hits", [{}])[0].get("value") == "lee", f"hits={data.get('hits')}")
    st, data = req("POST", "/api/json/query", {"json": SAMPLE_A, "path": "$..id"})
    check("T1 recursive ids", data.get("hitCount") == 2, f"n={data.get('hitCount')}")
    st, data = req("POST", "/api/json/query", {"json": SAMPLE_A, "path": "$.users[*].name"})
    names = [h.get("value") for h in data.get("hits", [])]
    check("T1 wildcard names", set(names) == {"lee", "wang"}, f"names={names}")

    OUT.write("== T2 schema ==\n")
    st, data = req("POST", "/api/json/schema", {"json": SAMPLE_A})
    check("T2 status", st == 200, f"st={st}")
    schema = data.get("schema", {})
    check("T2 root object", schema.get("type") == "object")
    props = schema.get("properties", {})
    check("T2 has users prop", "users" in props)
    users_schema = props.get("users", {})
    check("T2 users elementTypes", "object" in (users_schema.get("elementTypes") or []),
          f"et={users_schema.get('elementTypes')}")

    OUT.write("== T3 deep diff ==\n")
    st, data = req("POST", "/api/json/diff", {"jsonA": SAMPLE_A, "jsonB": SAMPLE_B})
    check("T3 status", st == 200, f"st={st}")
    check("T3 changed", data.get("changed") is True)
    check("T3 added>=1", data.get("added", 0) >= 1, f"added={data.get('added')}")
    check("T3 removed>=1", data.get("removed", 0) >= 1, f"removed={data.get('removed')}")
    check("T3 modified>=2", data.get("modified", 0) >= 2, f"mod={data.get('modified')}")
    paths = " ".join(str(c.get("path")) for c in data.get("changes", []))
    check("T3 version path", "version" in paths, f"paths={paths[:120]}")

    OUT.write("== T4 flatten ==\n")
    st, data = req("POST", "/api/json/flatten", {"json": SAMPLE_A})
    check("T4 status", st == 200, f"st={st}")
    check("T4 leafCount>0", data.get("leafCount", 0) > 5, f"n={data.get('leafCount')}")
    rows = data.get("rows", [])
    check("T4 has $.app", any(r.get("path") == "$.app" and r.get("value") == "demo" for r in rows))
    check("T4 has array path", any("users[0]" in str(r.get("path")) for r in rows))

    OUT.write("== T5 invalid input errors ==\n")
    st, data = req("POST", "/api/json/stats", {"json": "{not valid"})
    check("T5 bad json 400", st == 400, f"st={st}")
    check("T5 error message", "解析失败" in str(data.get("message", "")))
    st, data = req("POST", "/api/json/query", {"json": SAMPLE_A, "path": "users.name"})
    check("T5 bad path 400", st == 400, f"st={st}")

    OUT.write("== T6 AI tool call ==\n")
    text, tools = sse(f"json-e2e-t6-{int(time.time())}",
                      "请用工具统计这份 JSON 的结构：{\"a\":{\"b\":[1,2,3]},\"c\":\"x\"}，然后提取 $..b")
    joined = ",".join(tools)
    check("T6 called json tools", "json_parse_stats" in joined or "json_query_path" in joined, f"tools={joined}")
    check("T6 answer substantive", len(text) >= 50, f"len={len(text)}")

    OUT.write("== T7 safety: no fabricated data ==\n")
    text, tools = sse(f"json-e2e-t7-{int(time.time())}",
                      "帮我分析一下我们生产环境订单 JSON 的结构，直接开始分析就行。")
    asked_for_data = ("提供" in text or "贴" in text or "发给我" in text or "请给我" in text
                      or "需要" in text or "没有" in text)
    check("T7 asks for data", asked_for_data, f"len={len(text)}")

    OUT.write("\n")
    OUT.write(f"PASSED={len(PASSED)} FAILED={len(FAILED)}\n")
    OUT.flush()
    if FAILED:
        OUT.write("FAILED cases: " + ",".join(FAILED) + "\n")
        sys.exit(1)
    OUT.write("ALL 8/8 PASS ✅\n")


if __name__ == "__main__":
    main()