204 lines
8.5 KiB
Python
204 lines
8.5 KiB
Python
"""cloud mcp_server 单测：协议与 REST 翻译，不联网、不碰真实 key。
|
|
|
|
transport 用假函数注入，返回 (status, payload)；测试里出现的
|
|
key、URL 均为占位值。
|
|
"""
|
|
|
|
import io
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
try:
|
|
from cloud.mcp_server import MCPServer, TOOLS
|
|
except ImportError:
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
from cloud.mcp_server import MCPServer, TOOLS
|
|
|
|
|
|
class FakeTransport:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def __call__(self, method, path, body=None):
|
|
self.calls.append((method, path, body))
|
|
if path == "/health":
|
|
return 200, {"ok": True, "commit": "test"}
|
|
if path == "/projects":
|
|
return 200, {"projects": [{"id": "p1"}]}
|
|
if path == "/tasks?project_id=p1":
|
|
return 200, {"tasks": [{"id": "t1", "status": "completed"}]}
|
|
if path == "/tasks/t1":
|
|
return 200, {"id": "t1", "status": "completed"}
|
|
if path == "/tasks" and method == "POST":
|
|
return 201, {"id": "t2", "status": "queued"}
|
|
if path == "/tasks/t1/cancel":
|
|
return 200, {"id": "t1", "status": "interrupted"}
|
|
if path == "/tasks/t1/cost":
|
|
return 200, {"total_tokens": 10}
|
|
if path == "/tasks/t1/trace":
|
|
return 200, {"entries": []}
|
|
if path == "/tasks/t1/files":
|
|
return 200, {"files": [{"path": "a.py"}]}
|
|
if path == "/tasks/t1/files?path=a.py":
|
|
return 200, {"content": "print(1)"}
|
|
if path == "/tasks/t1/result-full":
|
|
return 200, {"result_text_full": "full-result"}
|
|
if path == "/tasks/t1/retry":
|
|
return 200, {"id": "t1", "status": "queued"}
|
|
if path == "/tasks/t1/merge":
|
|
return 200, {"id": "t1", "status": "merged"}
|
|
if path == "/tasks/t1/pause":
|
|
return 200, {"id": "t1", "status": "paused"}
|
|
if path == "/tasks/t1/goal-resume":
|
|
return 200, {"id": "t1", "status": "queued"}
|
|
if path == "/tasks/t1/followups" and method == "POST":
|
|
return 200, {"id": "t1", "mode": "queued"}
|
|
if path == "/tasks/t1/agents":
|
|
return 200, {"agents": []}
|
|
if path == "/workers":
|
|
return 200, {"workers_total": 2}
|
|
if path == "/metrics":
|
|
return 200, {"total_tasks": 3}
|
|
if path == "/projects/p1/members":
|
|
return 200, {"members": []}
|
|
if path == "/models":
|
|
return 200, {"models": []}
|
|
if path == "/tasks/nope":
|
|
return 404, {"error": "not found"}
|
|
return 400, {"error": "bad request"}
|
|
|
|
|
|
class MCPServerTestCase(unittest.TestCase):
|
|
def setUp(self):
|
|
self.transport = FakeTransport()
|
|
self.server = MCPServer("https://example.invalid", "test-key",
|
|
transport=self.transport)
|
|
|
|
def test_requires_api_key(self):
|
|
with self.assertRaises(ValueError):
|
|
MCPServer("https://example.invalid", "")
|
|
|
|
def test_initialize_and_tools_list(self):
|
|
resp = self.server.handle({"jsonrpc": "2.0", "id": 1,
|
|
"method": "initialize",
|
|
"params": {"protocolVersion": "2024-11-05"}})
|
|
self.assertEqual(resp["result"]["serverInfo"]["name"],
|
|
"codex-cloud-mcp")
|
|
resp = self.server.handle({"jsonrpc": "2.0", "id": 2,
|
|
"method": "tools/list"})
|
|
names = [t["name"] for t in resp["result"]["tools"]]
|
|
self.assertEqual(names, [t["name"] for t in TOOLS])
|
|
self.assertIn("create_task", names)
|
|
self.assertIn("cancel_task", names)
|
|
|
|
def test_notification_returns_none(self):
|
|
self.assertIsNone(self.server.handle(
|
|
{"jsonrpc": "2.0", "method": "notifications/initialized"}))
|
|
|
|
def test_unknown_method_and_tool(self):
|
|
resp = self.server.handle({"jsonrpc": "2.0", "id": 1,
|
|
"method": "resources/list"})
|
|
self.assertEqual(resp["error"]["code"], -32601)
|
|
is_error, payload = self.server.call_tool("nope", {})
|
|
self.assertTrue(is_error)
|
|
self.assertIn("unknown tool", payload["error"])
|
|
|
|
def test_list_and_get_translate_to_rest(self):
|
|
is_error, payload = self.server.call_tool(
|
|
"list_tasks", {"project_id": "p1"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["tasks"][0]["id"], "t1")
|
|
self.assertEqual(self.transport.calls[0][:2],
|
|
("GET", "/tasks?project_id=p1"))
|
|
is_error, payload = self.server.call_tool(
|
|
"get_task", {"task_id": "t1"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["status"], "completed")
|
|
|
|
def test_create_only_passes_known_fields(self):
|
|
is_error, payload = self.server.call_tool("create_task", {
|
|
"project_id": "p1", "prompt": "do it", "model": "m1",
|
|
"priority": 2, "injected": "drop-me"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["id"], "t2")
|
|
method, path, body = self.transport.calls[-1]
|
|
self.assertEqual((method, path), ("POST", "/tasks"))
|
|
self.assertNotIn("injected", body)
|
|
self.assertEqual(body["priority"], 2)
|
|
|
|
def test_rest_error_becomes_is_error(self):
|
|
resp = self.server.handle({"jsonrpc": "2.0", "id": 9,
|
|
"method": "tools/call",
|
|
"params": {"name": "get_task",
|
|
"arguments": {"task_id": "nope"}}})
|
|
self.assertTrue(resp["result"]["isError"])
|
|
payload = json.loads(resp["result"]["content"][0]["text"])
|
|
self.assertEqual(payload["http_status"], 404)
|
|
|
|
def test_new_tools_translate_to_rest(self):
|
|
is_error, payload = self.server.call_tool(
|
|
"get_task_files", {"task_id": "t1"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["files"][0]["path"], "a.py")
|
|
is_error, payload = self.server.call_tool(
|
|
"get_task_files", {"task_id": "t1", "path": "a.py"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["content"], "print(1)")
|
|
is_error, payload = self.server.call_tool(
|
|
"get_task_result_full", {"task_id": "t1"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["result_text_full"], "full-result")
|
|
is_error, payload = self.server.call_tool(
|
|
"retry_task", {"task_id": "t1"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["status"], "queued")
|
|
is_error, payload = self.server.call_tool(
|
|
"merge_task", {"task_id": "t1"})
|
|
self.assertFalse(is_error)
|
|
self.assertEqual(payload["status"], "merged")
|
|
|
|
def test_expanded_tools_translate_to_rest(self):
|
|
cases = [
|
|
("pause_task", {"task_id": "t1"}, ("POST", "/tasks/t1/pause")),
|
|
("goal_resume_task", {"task_id": "t1"},
|
|
("POST", "/tasks/t1/goal-resume")),
|
|
("create_followup", {"task_id": "t1", "body": "hi"},
|
|
("POST", "/tasks/t1/followups")),
|
|
("get_task_agents", {"task_id": "t1"},
|
|
("GET", "/tasks/t1/agents")),
|
|
("list_workers", {}, ("GET", "/workers")),
|
|
("get_metrics", {}, ("GET", "/metrics")),
|
|
("list_project_members", {"project_id": "p1"},
|
|
("GET", "/projects/p1/members")),
|
|
("get_models", {}, ("GET", "/models")),
|
|
]
|
|
for name, args, expected in cases:
|
|
is_error, payload = self.server.call_tool(name, args)
|
|
self.assertFalse(is_error, name)
|
|
self.assertEqual(self.transport.calls[-1][:2], expected, name)
|
|
is_error, _ = self.server.call_tool("create_followup", {"task_id": "t1"})
|
|
self.assertTrue(is_error)
|
|
|
|
def test_missing_arg_is_error_not_rest_call(self):
|
|
before = len(self.transport.calls)
|
|
is_error, payload = self.server.call_tool("list_tasks", {})
|
|
self.assertTrue(is_error)
|
|
self.assertEqual(len(self.transport.calls), before)
|
|
|
|
def test_serve_loop_handles_bad_line(self):
|
|
instream = io.StringIO(
|
|
"not json\n"
|
|
'{"jsonrpc": "2.0", "id": 1, "method": "ping"}\n')
|
|
outstream = io.StringIO()
|
|
self.server.serve(instream, outstream)
|
|
lines = [json.loads(line) for line in outstream.getvalue().splitlines()]
|
|
self.assertEqual(lines[0]["error"]["code"], -32700)
|
|
self.assertEqual(lines[1]["id"], 1)
|
|
self.assertEqual(lines[1]["result"], {})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|