127 lines
3.9 KiB
Python
127 lines
3.9 KiB
Python
|
"""验证基座 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())
|