test_knowtation_mcp_integration.py
python
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
13 days ago
| 1 | """Tier 2 — integration tests against knowtation-vault-smoke.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import json |
| 6 | import pathlib |
| 7 | from unittest.mock import patch |
| 8 | |
| 9 | import pytest |
| 10 | |
| 11 | from muse.mcp.server import MuseMCPServer |
| 12 | from muse.mcp.namespaces.knowtation import KnowtationNamespace |
| 13 | |
| 14 | SMOKE_OVERVIEW = "projects/kd-smoke/overview.md" |
| 15 | |
| 16 | |
| 17 | @pytest.fixture |
| 18 | def smoke_root() -> pathlib.Path: |
| 19 | return pathlib.Path("/Users/aaronrenecarvajal/MUSE_HUB/knowtation-vault-smoke") |
| 20 | |
| 21 | |
| 22 | @pytest.fixture |
| 23 | def ns(smoke_root: pathlib.Path) -> KnowtationNamespace: |
| 24 | return KnowtationNamespace(smoke_root) |
| 25 | |
| 26 | |
| 27 | class TestToolIntegration: |
| 28 | def test_get_note_overview(self, ns: KnowtationNamespace) -> None: |
| 29 | out = ns.call_tool("knowtation/get-note", {"path": SMOKE_OVERVIEW}) |
| 30 | assert out["exists"] is True |
| 31 | assert out["path"] == SMOKE_OVERVIEW |
| 32 | assert "content_hash" in out |
| 33 | |
| 34 | def test_prime_enriched(self, ns: KnowtationNamespace) -> None: |
| 35 | out = ns.call_tool("knowtation/prime", {}) |
| 36 | assert out["schema_version"] == "1.0.0" |
| 37 | assert "active_projects" in out |
| 38 | assert "suggested_next" in out |
| 39 | |
| 40 | def test_impact_and_dead(self, ns: KnowtationNamespace) -> None: |
| 41 | impact = ns.call_tool("knowtation/impact", {"path": SMOKE_OVERVIEW}) |
| 42 | assert impact["note"] == SMOKE_OVERVIEW |
| 43 | dead = ns.call_tool("knowtation/dead", {"use_mtime": False}) |
| 44 | assert "dead_count" in dead |
| 45 | |
| 46 | def test_consolidate_receipt(self, ns: KnowtationNamespace) -> None: |
| 47 | out = ns.call_tool("knowtation/consolidate", {}) |
| 48 | for key in ("hook_name", "fired", "out_of_band", "message", "dispatched_at", "hub_port"): |
| 49 | assert key in out |
| 50 | |
| 51 | @patch("muse.plugins.knowtation.search.vector_search") |
| 52 | def test_search_hybrid_fanout(self, mock_vec, ns: KnowtationNamespace) -> None: |
| 53 | mock_vec.return_value = ( |
| 54 | [{"path": SMOKE_OVERVIEW, "score": 0.9, "snippet": "hit"}], |
| 55 | "ok", |
| 56 | ) |
| 57 | out = ns.call_tool("knowtation/search", {"query": "overview", "top_k": 5}) |
| 58 | assert out["query"] == "overview" |
| 59 | assert out["backends"]["lexical"] == "ok" |
| 60 | assert out["backends"]["vector"] == "ok" |
| 61 | |
| 62 | @patch("muse.plugins.knowtation.mcp_helpers.urllib.request.urlopen") |
| 63 | def test_propose_hub_stub(self, mock_urlopen, ns: KnowtationNamespace) -> None: |
| 64 | class _Resp: |
| 65 | def read(self, n: int = -1) -> bytes: |
| 66 | return b'{"id":"prop-123"}' |
| 67 | |
| 68 | def __enter__(self): |
| 69 | return self |
| 70 | |
| 71 | def __exit__(self, *a: object) -> None: |
| 72 | return None |
| 73 | |
| 74 | mock_urlopen.return_value = _Resp() |
| 75 | out = ns.call_tool( |
| 76 | "knowtation/propose", |
| 77 | {"path": SMOKE_OVERVIEW, "title": "Test proposal"}, |
| 78 | ) |
| 79 | assert out["status"] == "created" |
| 80 | assert out["proposal_id"] == "prop-123" |
| 81 | |
| 82 | |
| 83 | class TestResourceIntegration: |
| 84 | def test_vault_graph_prime(self, ns: KnowtationNamespace) -> None: |
| 85 | vault = ns.read_resource("knowtation://vault") |
| 86 | assert "notes" in vault and "stats" in vault |
| 87 | graph = ns.read_resource("knowtation://graph") |
| 88 | assert "forward" in graph and "entities" in graph |
| 89 | prime = ns.read_resource("knowtation://prime") |
| 90 | assert prime["schema_version"] == "1.0.0" |
| 91 | |
| 92 | def test_memory_and_projects(self, ns: KnowtationNamespace) -> None: |
| 93 | mem = ns.read_resource("knowtation://memory/recent") |
| 94 | assert "events" in mem |
| 95 | projects = ns.read_resource("knowtation://projects") |
| 96 | assert "projects" in projects |
| 97 | hot = ns.read_resource("knowtation://hotspots") |
| 98 | assert "ranking" in hot |
| 99 | |
| 100 | |
| 101 | class TestPromptIntegration: |
| 102 | def test_search_synthesis(self, ns: KnowtationNamespace) -> None: |
| 103 | with patch("muse.plugins.knowtation.search.vector_search", return_value=([], "unavailable")): |
| 104 | p = ns.get_prompt("vault/search-synthesis", {"query": "overview"}) |
| 105 | assert p["messages"] |
| 106 | |
| 107 | def test_memory_session(self, ns: KnowtationNamespace) -> None: |
| 108 | p = ns.get_prompt("vault/memory-session", {}) |
| 109 | assert "Prime" in p["messages"][0]["content"]["text"] |
File History
1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8
docs(KD-STAGING): sync governance after KD-6b DONE
Human
13 days ago