aaronrene / gabriel-muse public
test_knowtation_mcp_e2e.py python
136 lines 4.6 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """Tier 3 — JSON-RPC E2E + Claude Desktop integration evidence."""
2
3 from __future__ import annotations
4
5 import json
6 import pathlib
7 import subprocess
8 import sys
9
10 import pytest
11
12 from muse.mcp.server import MuseMCPServer
13
14
15 SMOKE = pathlib.Path("/Users/aaronrenecarvajal/MUSE_HUB/knowtation-vault-smoke")
16
17
18 def _rpc(server: MuseMCPServer, method: str, params: dict | None = None, req_id: int = 1) -> dict:
19 raw = {"jsonrpc": "2.0", "id": req_id, "method": method}
20 if params is not None:
21 raw["params"] = params
22 resp = server.handle_request(raw)
23 assert resp is not None
24 return resp
25
26
27 class TestJsonRpcE2E:
28 @pytest.fixture
29 def server(self) -> MuseMCPServer:
30 return MuseMCPServer(SMOKE)
31
32 def test_initialize(self, server: MuseMCPServer) -> None:
33 resp = _rpc(server, "initialize")
34 assert resp["result"]["serverInfo"]["name"] == "muse-mcp"
35
36 def test_tools_call_search_and_resource_prime(self, server: MuseMCPServer) -> None:
37 tools = _rpc(server, "tools/list")["result"]["tools"]
38 assert len(tools) == 9
39 resources = _rpc(server, "resources/list")["result"]["resources"]
40 assert len(resources) >= 6
41 prompts = _rpc(server, "prompts/list")["result"]["prompts"]
42 assert len(prompts) == 4
43
44 search = _rpc(
45 server,
46 "tools/call",
47 {"name": "knowtation/search", "arguments": {"query": "overview", "mode": "lexical"}},
48 req_id=2,
49 )
50 payload = json.loads(search["result"]["content"][0]["text"])
51 assert payload["mode"] == "lexical"
52
53 prime = _rpc(server, "resources/read", {"uri": "knowtation://prime"}, req_id=3)
54 prime_body = json.loads(prime["result"]["contents"][0]["text"])
55 assert prime_body["schema_version"] == "1.0.0"
56
57 prompt = _rpc(
58 server,
59 "prompts/get",
60 {"name": "vault/daily-brief", "arguments": {"since_hours": 24}},
61 req_id=4,
62 )
63 assert prompt["result"]["messages"]
64
65
66 class TestClaudeDesktopIntegration:
67 """Evidence matrix: stdio subprocess bound to knowtation-vault-smoke."""
68
69 def test_stdio_subprocess_list_and_roundtrip(self) -> None:
70 init = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "tools/list"})
71 proc = subprocess.run(
72 [sys.executable, "-m", "muse.mcp.stdio_server"],
73 input=init + "\n",
74 cwd=str(SMOKE),
75 capture_output=True,
76 text=True,
77 timeout=30,
78 check=False,
79 )
80 assert proc.returncode == 0, proc.stderr
81 line = proc.stdout.strip().splitlines()[0]
82 data = json.loads(line)
83 names = {t["name"] for t in data["result"]["tools"]}
84 assert len(names) == 9
85
86 call = json.dumps(
87 {
88 "jsonrpc": "2.0",
89 "id": 2,
90 "method": "tools/call",
91 "params": {
92 "name": "knowtation/search",
93 "arguments": {"query": "kd-smoke", "mode": "lexical"},
94 },
95 }
96 )
97 proc2 = subprocess.run(
98 [sys.executable, "-m", "muse.mcp.stdio_server"],
99 input=init + "\n" + call + "\n",
100 cwd=str(SMOKE),
101 capture_output=True,
102 text=True,
103 timeout=30,
104 check=False,
105 )
106 assert proc2.returncode == 0, proc2.stderr
107 lines = [json.loads(ln) for ln in proc2.stdout.strip().splitlines() if ln.strip()]
108 search_resp = next(r for r in lines if r.get("id") == 2)
109 body = json.loads(search_resp["result"]["content"][0]["text"])
110 assert body["query"] == "kd-smoke"
111
112 res_list = json.dumps({"jsonrpc": "2.0", "id": 3, "method": "resources/list"})
113 proc3 = subprocess.run(
114 [sys.executable, "-m", "muse.mcp.stdio_server"],
115 input=res_list + "\n",
116 cwd=str(SMOKE),
117 capture_output=True,
118 text=True,
119 timeout=30,
120 check=False,
121 )
122 resources = json.loads(proc3.stdout.strip())["result"]["resources"]
123 assert any(r["uri"] == "knowtation://prime" for r in resources)
124
125 prompt_list = json.dumps({"jsonrpc": "2.0", "id": 4, "method": "prompts/list"})
126 proc4 = subprocess.run(
127 [sys.executable, "-m", "muse.mcp.stdio_server"],
128 input=prompt_list + "\n",
129 cwd=str(SMOKE),
130 capture_output=True,
131 text=True,
132 timeout=30,
133 check=False,
134 )
135 prompts = json.loads(proc4.stdout.strip())["result"]["prompts"]
136 assert len(prompts) == 4
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago