test_knowtation_mcp_e2e.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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 |