cad-agent/verify_mcp_frame.py

"""验证基座 B 最小集改动（换行分帧 / 大报文 / string id / 通知 / isError）。

前提：CAD 已打开并加载了重新构建的 cad_mcp_frame.arx，8080 端口在监听。

用法：
    python verify_mcp_frame.py
"""

import json
import socket
import sys

HOST = "127.0.0.1"
PORT = 8080


def raw_send(payload_bytes, *, timeout=10.0):
    """发送一帧（已含 '\\n'），读回响应原始字节（读不到则为空）。"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(timeout)
    try:
        sock.connect((HOST, PORT))
        sock.sendall(payload_bytes)
        data = b""
        while True:
            try:
                chunk = sock.recv(65536)
            except socket.timeout:
                break
            if not chunk:
                break
            data += chunk
            if b"\n" in data:
                break
        return data
    finally:
        sock.close()


def _call(payload):
    return raw_send(json.dumps(payload).encode("utf-8") + b"\n")


def test_string_id():
    resp = json.loads(_call({"jsonrpc": "2.0", "id": "abc-123", "method": "tools/list"}).decode())
    assert resp.get("id") == "abc-123", f"id 未原样回显: {resp.get('id')!r}"
    count = len(resp.get("result", {}).get("tools", []))
    print(f"[PASS] string id 原样回显；tools={count}")


def test_large_request():
    """构造 >2KB 的 arguments；旧基座会截断成 -32700，新基座应正常返回 -32601。"""
    big = {"blob": "x" * 5000}
    resp = json.loads(
        _call(
            {
                "jsonrpc": "2.0",
                "id": 7,
                "method": "tools/call",
                "params": {"name": "__no_such_tool__", "arguments": big},
            }
        ).decode()
    )
    code = (resp.get("error") or {}).get("code")
    assert code == -32601, f"期望 -32601（工具不存在），实际 {resp}"
    print("[PASS] >2KB 请求未被截断（返回 -32601 而非 -32700）")


def test_notification_no_reply():
    data = raw_send(
        json.dumps({"jsonrpc": "2.0", "method": "notifications/initialized"}).encode() + b"\n",
        timeout=2.0,
    )
    assert data == b"", f"通知不应回包，却收到 {data[:120]!r}"
    print("[PASS] 无 id 的通知未回包")


def test_tool_error_shape():
    """用不存在的 handle 触发工具执行错误，应为 result.isError=true。"""
    resp = json.loads(
        _call(
            {
                "jsonrpc": "2.0",
                "id": 9,
                "method": "tools/call",
                "params": {"name": "get_entity_detail", "arguments": {"handle": "ZZZZZZ"}},
            }
        ).decode()
    )
    result = resp.get("result")
    if isinstance(result, dict) and result.get("isError") is True:
        print("[PASS] 工具执行错误 -> result.isError=true")
    elif resp.get("error"):
        print(f"[WARN] 仍是顶层 error（code={resp['error'].get('code')}），isError 映射未生效")
    else:
        print(f"[?] 未预期响应: {resp}")


def main():
    # 先探测端口
    probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    probe.settimeout(1.5)
    try:
        probe.connect((HOST, PORT))
    except OSError as exc:
        print(f"[X] {HOST}:{PORT} 未监听（{exc}）。请先启动已加载插件的 CAD。")
        return 1
    finally:
        probe.close()

    print(f"=== 验证基座 {HOST}:{PORT} ===")
    failed = 0
    for test in (test_string_id, test_large_request, test_notification_no_reply, test_tool_error_shape):
        try:
            test()
        except AssertionError as exc:
            print(f"[FAIL] {test.__name__}: {exc}")
            failed += 1
        except Exception as exc:  # noqa: BLE001
            print(f"[ERROR] {test.__name__}: {exc}")
            failed += 1
    print("=== 全部通过 ===" if failed == 0 else f"=== {failed} 项未通过 ===")
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())