"""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/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_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()