"""Tier 2 — integration tests against knowtation-vault-smoke.""" from __future__ import annotations import json import pathlib from unittest.mock import patch import pytest from muse.mcp.server import MuseMCPServer from muse.mcp.namespaces.knowtation import KnowtationNamespace SMOKE_OVERVIEW = "projects/kd-smoke/overview.md" @pytest.fixture def smoke_root() -> pathlib.Path: return pathlib.Path("/Users/aaronrenecarvajal/MUSE_HUB/knowtation-vault-smoke") @pytest.fixture def ns(smoke_root: pathlib.Path) -> KnowtationNamespace: return KnowtationNamespace(smoke_root) class TestToolIntegration: def test_get_note_overview(self, ns: KnowtationNamespace) -> None: out = ns.call_tool("knowtation/get-note", {"path": SMOKE_OVERVIEW}) assert out["exists"] is True assert out["path"] == SMOKE_OVERVIEW assert "content_hash" in out def test_prime_enriched(self, ns: KnowtationNamespace) -> None: out = ns.call_tool("knowtation/prime", {}) assert out["schema_version"] == "1.0.0" assert "active_projects" in out assert "suggested_next" in out def test_impact_and_dead(self, ns: KnowtationNamespace) -> None: impact = ns.call_tool("knowtation/impact", {"path": SMOKE_OVERVIEW}) assert impact["note"] == SMOKE_OVERVIEW dead = ns.call_tool("knowtation/dead", {"use_mtime": False}) assert "dead_count" in dead def test_consolidate_receipt(self, ns: KnowtationNamespace) -> None: out = ns.call_tool("knowtation/consolidate", {}) for key in ("hook_name", "fired", "out_of_band", "message", "dispatched_at", "hub_port"): assert key in out @patch("muse.plugins.knowtation.search.vector_search") def test_search_hybrid_fanout(self, mock_vec, ns: KnowtationNamespace) -> None: mock_vec.return_value = ( [{"path": SMOKE_OVERVIEW, "score": 0.9, "snippet": "hit"}], "ok", ) out = ns.call_tool("knowtation/search", {"query": "overview", "top_k": 5}) assert out["query"] == "overview" assert out["backends"]["lexical"] == "ok" assert out["backends"]["vector"] == "ok" @patch("muse.plugins.knowtation.mcp_helpers.urllib.request.urlopen") def test_propose_hub_stub(self, mock_urlopen, ns: KnowtationNamespace) -> None: class _Resp: def read(self, n: int = -1) -> bytes: return b'{"id":"prop-123"}' def __enter__(self): return self def __exit__(self, *a: object) -> None: return None mock_urlopen.return_value = _Resp() out = ns.call_tool( "knowtation/propose", {"path": SMOKE_OVERVIEW, "title": "Test proposal"}, ) assert out["status"] == "created" assert out["proposal_id"] == "prop-123" class TestResourceIntegration: def test_vault_graph_prime(self, ns: KnowtationNamespace) -> None: vault = ns.read_resource("knowtation://vault") assert "notes" in vault and "stats" in vault graph = ns.read_resource("knowtation://graph") assert "forward" in graph and "entities" in graph prime = ns.read_resource("knowtation://prime") assert prime["schema_version"] == "1.0.0" def test_memory_and_projects(self, ns: KnowtationNamespace) -> None: mem = ns.read_resource("knowtation://memory/recent") assert "events" in mem projects = ns.read_resource("knowtation://projects") assert "projects" in projects hot = ns.read_resource("knowtation://hotspots") assert "ranking" in hot class TestPromptIntegration: def test_search_synthesis(self, ns: KnowtationNamespace) -> None: with patch("muse.plugins.knowtation.search.vector_search", return_value=([], "unavailable")): p = ns.get_prompt("vault/search-synthesis", {"query": "overview"}) assert p["messages"] def test_memory_session(self, ns: KnowtationNamespace) -> None: p = ns.get_prompt("vault/memory-session", {}) assert "Prime" in p["messages"][0]["content"]["text"]