api_spec_e2e.py

#!/usr/bin/env python3
"""API Spec Studio E2E: app API + AI stream tool calls.

Usage: python3 api_spec_e2e.py
Prereq: app on 127.0.0.1:8094, harness on 127.0.0.1:8090,
        plugin dsh-java-api-plugin installed & activated.
"""
import json
import sys
import time
import urllib.request

APP = "http://127.0.0.1:8094"

SPEC_V1 = {
    "openapi": "3.0.0",
    "info": {"title": "Order Service", "version": "1.0.0"},
    "paths": {
        "/api/orders": {
            "get": {
                "operationId": "listOrders",
                "summary": "订单列表",
                "tags": ["orders"],
                "responses": {"200": {"description": "ok"}}
            },
            "post": {
                "operationId": "createOrder",
                "summary": "创建订单",
                "tags": ["orders"],
                "requestBody": {
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "object",
                                "required": ["skuId", "quantity"],
                                "properties": {
                                    "skuId": {"type": "string", "example": "SKU-001"},
                                    "quantity": {"type": "integer"},
                                    "coupon": {"type": "string"}
                                }
                            }
                        }
                    }
                },
                "responses": {"200": {"description": "ok"}, "400": {"description": "bad request"}}
            }
        },
        "/api/orders/{id}": {
            "get": {
                "operationId": "getOrder",
                "summary": "订单详情",
                "tags": ["orders"],
                "parameters": [
                    {"name": "id", "in": "path", "required": True, "schema": {"type": "string"}},
                    {"name": "verbose", "in": "query", "schema": {"type": "boolean", "enum": [True, False]}}
                ],
                "responses": {"200": {"description": "ok"}}
            },
            "delete": {
                "operationId": "deleteOrder",
                "summary": "删除订单",
                "tags": ["orders"],
                "responses": {"204": {"description": "deleted"}}
            }
        },
        "/api/users/me": {
            "get": {"summary": "当前用户", "tags": ["users"], "responses": {"200": {"description": "ok"}}}
        }
    }
}

SPEC_V2 = {
    "openapi": "3.0.0",
    "info": {"title": "Order Service", "version": "2.0.0"},
    "paths": {
        "/api/orders": {
            "get": {
                "operationId": "listOrders",
                "summary": "订单列表",
                "tags": ["orders"],
                "responses": {"200": {"description": "ok"}}
            },
            "post": {
                "operationId": "createOrder",
                "summary": "创建订单",
                "tags": ["orders"],
                "requestBody": {
                    "content": {
                        "application/json": {
                            "schema": {
                                "type": "object",
                                "required": ["skuId", "quantity", "channel"],
                                "properties": {
                                    "skuId": {"type": "string"},
                                    "quantity": {"type": "integer"},
                                    "channel": {"type": "string"}
                                }
                            }
                        }
                    }
                },
                "responses": {"200": {"description": "ok"}}
            }
        },
        "/api/orders/{id}": {
            "get": {
                "operationId": "getOrder",
                "summary": "订单详情",
                "tags": ["orders"],
                "deprecated": True,
                "parameters": [
                    {"name": "id", "in": "path", "required": True, "schema": {"type": "string"}}
                ],
                "responses": {"200": {"description": "ok"}}
            }
        },
        "/api/users/me": {
            "get": {"summary": "当前用户", "tags": ["users"], "responses": {"200": {"description": "ok"}}}
        },
        "/api/refunds": {
            "post": {"summary": "退款", "tags": ["refunds"], "responses": {"200": {"description": "ok"}}}
        }
    }
}

results = []


def call(path, payload=None, method=None):
    data = json.dumps(payload).encode() if payload is not None else None
    req = urllib.request.Request(APP + path, data=data, method=method or ("POST" if data else "GET"),
                                 headers={"Content-Type": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            return resp.status, json.loads(resp.read().decode() or "{}")
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read().decode() or "{}")


def stream_once(agent_id, message, spec_id=None):
    payload = {"agentId": agent_id, "message": message}
    if spec_id:
        payload["specId"] = spec_id
    body = json.dumps(payload).encode()
    req = urllib.request.Request(APP + "/api/assistant/stream", data=body, method="POST",
                                 headers={"Content-Type": "application/json"})
    content, tools, events = "", [], []
    try:
        with urllib.request.urlopen(req, timeout=200) as resp:
            event, data_buf = "message", []
            for raw in resp.read().decode("utf-8", "ignore").splitlines() + [""]:
                if raw.startswith("event:"):
                    event = raw[6:].strip()
                elif raw.startswith("data:"):
                    data_buf.append(raw[5:].strip())
                elif not raw and data_buf:
                    try:
                        payload = json.loads("\n".join(data_buf))
                        events.append(event)
                        if event == "chunk":
                            content += payload.get("content", "")
                        if event == "step_break" and payload.get("toolName"):
                            tools.append(payload["toolName"])
                    except json.JSONDecodeError:
                        pass
                    event, data_buf = "message", []
    except Exception as e:  # noqa: BLE001
        content += f"\nSTREAM-ERROR: {e}"
    return content, sorted(set(tools)), events


def case(name):
    def deco(fn):
        def wrapped():
            start = time.time()
            try:
                detail = fn()
                results.append((name, True, detail, time.time() - start))
                print(f"PASS  {name}  ({time.time()-start:.1f}s)  {detail}")
            except AssertionError as e:
                results.append((name, False, str(e), time.time() - start))
                print(f"FAIL  {name}  ({time.time()-start:.1f}s)  {e}")
            except Exception as e:  # noqa: BLE001
                results.append((name, False, f"{type(e).__name__}: {e}", time.time() - start))
                print(f"FAIL  {name}  ({time.time()-start:.1f}s)  {type(e).__name__}: {e}")
        return wrapped
    return deco


OLD_ID = NEW_ID = None


@case("T0 加载两份规范")
def t0():
    global OLD_ID, NEW_ID
    status, body = call("/api/specs", {"name": "order-service-v1", "json": json.dumps(SPEC_V1)})
    assert status == 200, f"load v1 {status}: {body}"
    OLD_ID = body["id"]
    status, body = call("/api/specs", {"name": "order-service-v2", "json": json.dumps(SPEC_V2)})
    assert status == 200, f"load v2 {status}: {body}"
    NEW_ID = body["id"]
    assert OLD_ID != NEW_ID
    return f"old={OLD_ID} new={NEW_ID}"


@case("T1 校验（operationCount）")
def t1():
    status, body = call(f"/api/specs/{OLD_ID}/validate", {})
    assert status == 200, f"validate {status}: {body}"
    assert body["operationCount"] == 5, f"operationCount={body['operationCount']}"
    return f"valid={body['valid']} errors={body['errorCount']} warns={body['warnCount']}"


@case("T2 端点清单")
def t2():
    status, body = call(f"/api/specs/{OLD_ID}/endpoints", {})
    assert status == 200, f"endpoints {status}: {body}"
    assert body["endpointCount"] == 5, f"count={body['endpointCount']}"
    methods = {e["method"] for e in body["endpoints"]}
    assert {"GET", "POST", "DELETE"} <= methods, f"methods={methods}"
    return f"endpoints={body['endpointCount']}"


@case("T3 按 tag 检索")
def t3():
    status, body = call(f"/api/specs/{OLD_ID}/find-by-tag", {"tag": "orders"})
    assert status == 200, f"tag {status}: {body}"
    assert body["endpointCount"] == 4, f"orders count={body['endpointCount']}"
    return f"orders={body['endpointCount']}"


@case("T4 破坏性变更对比")
def t4():
    status, body = call("/api/specs/diff", {"oldId": OLD_ID, "newId": NEW_ID})
    assert status == 200, f"diff {status}: {body}"
    assert body["breakingCount"] >= 3, f"breaking={body['breakingCount']} changes={body['changes']}"
    joined = json.dumps(body, ensure_ascii=False)
    assert "endpoint-removed" in joined, "deleteOrder removal missing"
    assert "deprecated" in joined, "deprecated change missing"
    assert "required-field-added" in joined, "channel required field missing"
    return f"changes={body['changeCount']} breaking={body['breakingCount']}"


@case("T5 Mock 提示")
def t5():
    status, body = call(f"/api/specs/{OLD_ID}/mock-hint", {"method": "GET", "path": "/api/orders/{id}"})
    assert status == 200, f"mock {status}: {body}"
    params = {p["name"]: p for p in body["params"]}
    assert "id" in params and params["id"]["required"], f"params={list(params)}"
    assert "verbose" in params and params["verbose"].get("enum"), "verbose enum missing"
    assert "curlDraft" in body and "curl -X GET" in body["curlDraft"], body.get("curlDraft", "")
    return f"params={list(params)}"


@case("T6 AI 对话调用 API 工具")
def t6():
    content, tools, events = stream_once("e2e-api-t6", "这份规范里有几个端点？哪些是 orders 标签的？", OLD_ID)
    assert any("api_" in t for t in tools), f"no api_ tool called, tools={tools}"
    assert "chunk" in events, "no chunk events"
    assert len(content) > 20, f"content too short: {content[:80]}"
    return f"tools={tools} chars={len(content)}"


@case("T7 非法 JSON 拒绝")
def t7():
    status, body = call("/api/specs", {"name": "bad", "json": "{not json"})
    assert status == 400, f"should reject: {status} {body}"
    return "invalid JSON rejected with 400"


def main():
    for fn in (t0, t1, t2, t3, t4, t5, t6, t7):
        fn()
    passed = sum(1 for _, ok, _, _ in results if ok)
    print(f"\n{passed}/{len(results)} PASS")
    sys.exit(0 if passed == len(results) else 1)


if __name__ == "__main__":
    main()