gabriel / musehub public
test_kd6_vault_ui.py python
445 lines 15.0 KB
Raw
sha256:c04414e7e3b812ff02f351d9792c79837b7841133a864f98fcaeff53a2a4afad feat(KD-6b): MuseHub vault UI per frozen §KD-6a Human minor ⚠ breaking 12 days ago
1 """KD-6b vault UI — seven-tier test matrix (§KD-6a.F)."""
2
3 from __future__ import annotations
4
5 import asyncio
6 import pathlib
7 import secrets
8 from datetime import datetime, timezone
9 from unittest.mock import patch
10
11 import pytest
12 from httpx import AsyncClient
13 from sqlalchemy.ext.asyncio import AsyncSession
14
15 from musehub.services.knowtation_view import (
16 KnowtationViewAdapter,
17 KnowtationViewResult,
18 VaultUIState,
19 compute_harmony_diff,
20 frontmatter_panel_html,
21 harmony_diff_html,
22 is_vault_domain_active,
23 map_mcp_reason,
24 render_note_html,
25 render_wikilinks,
26 split_frontmatter,
27 structured_delta_to_view_model,
28 validate_note_path,
29 )
30 from musehub.services.knowtation_view.path_grammar import PathGrammarError
31
32 SMOKE_ROOT = pathlib.Path("/Users/aaronrenecarvajal/MUSE_HUB/knowtation-vault-smoke")
33 SMOKE_NOTE = "projects/kd-smoke/overview.md"
34
35
36 # ---------------------------------------------------------------------------
37 # Fixtures
38 # ---------------------------------------------------------------------------
39
40
41 @pytest.fixture
42 def smoke_root() -> pathlib.Path:
43 if not SMOKE_ROOT.is_dir():
44 pytest.skip("knowtation-vault-smoke fixture not present")
45 return SMOKE_ROOT
46
47
48 async def _seed_knowtation_repo(
49 db: AsyncSession,
50 *,
51 owner: str = "kd6owner",
52 slug: str = "kd6-vault",
53 ):
54 from musehub.core.genesis import compute_identity_id, compute_repo_id
55 from musehub.db.musehub_domain_models import MusehubDomain
56 from musehub.db.musehub_repo_models import MusehubRepo
57
58 domain = MusehubDomain(
59 domain_id=f"sha256:{secrets.token_hex(32)}",
60 author_slug="aaronrene",
61 slug="knowtation",
62 display_name="Knowtation",
63 description="Test domain",
64 version="0.1.0",
65 viewer_type="generic",
66 capabilities={},
67 )
68 db.add(domain)
69 await db.flush()
70 created = datetime.now(tz=timezone.utc)
71 owner_id = compute_identity_id(owner.encode())
72 repo = MusehubRepo(
73 repo_id=compute_repo_id(owner_id, slug, "knowtation", created.isoformat()),
74 name=slug,
75 owner=owner,
76 slug=slug,
77 visibility="public",
78 owner_user_id=owner_id,
79 marketplace_domain_id=domain.domain_id,
80 domain_id="knowtation",
81 created_at=created,
82 updated_at=created,
83 )
84 db.add(repo)
85 await db.commit()
86 await db.refresh(repo)
87 await db.refresh(domain)
88 return repo, domain
89
90
91 # ---------------------------------------------------------------------------
92 # Tier 1 — Unit
93 # ---------------------------------------------------------------------------
94
95
96 @pytest.mark.tier1
97 class TestVaultUnit:
98 def test_gate_active_for_knowtation_scoped_id(self) -> None:
99 assert is_vault_domain_active({"scoped_id": "@aaronrene/knowtation"}) is True
100
101 def test_gate_inactive_for_code_domain(self) -> None:
102 assert is_vault_domain_active({"scoped_id": "@gabriel/muse"}) is False
103 assert is_vault_domain_active(None) is False
104
105 def test_path_grammar_rejects_traversal(self) -> None:
106 with pytest.raises(PathGrammarError) as exc:
107 validate_note_path("../secret.md")
108 assert exc.value.reason == "PATH_INVALID"
109
110 def test_path_grammar_rejects_non_note(self) -> None:
111 with pytest.raises(PathGrammarError):
112 validate_note_path("data/index.sqlite")
113
114 def test_wikilink_resolves_and_broken_class(self) -> None:
115 raw = "See [[linked-target]] and [[missing-note]]"
116 out = render_wikilinks(
117 raw,
118 owner="o",
119 repo_slug="r",
120 note_paths=frozenset({"linked-target.md"}),
121 )
122 assert "vault-wikilink" in out
123 assert "broken" in out
124
125 def test_frontmatter_split_and_panel_no_script(self) -> None:
126 raw = "---\ntitle: T\ntags: [a]\n---\n# Body\n"
127 fm, body = split_frontmatter(raw)
128 assert fm["title"] == "T"
129 assert "Body" in body
130 panel = frontmatter_panel_html(fm)
131 assert "<script" not in panel
132 assert "title" in panel
133
134 def test_markdown_xss_sanitised(self) -> None:
135 _, html = render_note_html("<script>x</script>\n**bold**", owner="o", repo_slug="r")
136 assert "<script" not in html
137 assert "<strong>" in html
138
139 def test_degraded_banner_logic(self) -> None:
140 r = KnowtationViewResult(ok=True, data={}, degraded=True)
141 assert r.ui_state == VaultUIState.SEARCH_DEGRADED
142
143 def test_mcp_reason_mapping(self) -> None:
144 assert map_mcp_reason("SCOPE_UNAUTHORIZED") == VaultUIState.PROPOSAL_SCOPE_UNAUTHORIZED
145
146 def test_harmony_view_model_from_delta(self) -> None:
147 delta = {
148 "domain": "knowtation",
149 "summary": "tags changed",
150 "ops": [
151 {"op": "insert", "address": "frontmatter:tags", "content_summary": "smoke"},
152 {"op": "replace", "address": "section:2:Summary#1", "content_summary": "updated"},
153 ],
154 }
155 vm = structured_delta_to_view_model(delta)
156 assert vm["section_count"] == 2
157 html = harmony_diff_html(vm)
158 assert "vault-harmony-diff" in html
159 assert "<script" not in html
160
161
162 # ---------------------------------------------------------------------------
163 # Tier 2 — Integration (adapter + smoke vault)
164 # ---------------------------------------------------------------------------
165
166
167 @pytest.mark.tier2
168 class TestVaultIntegration:
169 @pytest.fixture(autouse=True)
170 def _require_mcp(self, smoke_root: pathlib.Path) -> None:
171 adapter = KnowtationViewAdapter(smoke_root)
172 if not adapter.mounted:
173 pytest.skip("knowtation MCP plugin not importable")
174
175 def test_adapter_vault_resource(self, smoke_root: pathlib.Path) -> None:
176 adapter = KnowtationViewAdapter(smoke_root)
177 assert adapter.mounted
178 r = adapter.read_resource("knowtation://vault")
179 assert r.ok
180 assert "notes" in r.data or "stats" in r.data
181
182 def test_adapter_search_lexical(self, smoke_root: pathlib.Path) -> None:
183 adapter = KnowtationViewAdapter(smoke_root)
184 r = adapter.search("overview", mode="lexical", top_k=5)
185 assert r.ok
186 assert isinstance(r.data.get("results"), list)
187
188 def test_adapter_get_note_and_impact(self, smoke_root: pathlib.Path) -> None:
189 adapter = KnowtationViewAdapter(smoke_root)
190 note = adapter.get_note(SMOKE_NOTE)
191 assert note.ok
192 impact = adapter.impact(SMOKE_NOTE)
193 assert impact.ok
194
195 def test_intel_strip_loader(self, smoke_root: pathlib.Path) -> None:
196 adapter = KnowtationViewAdapter(smoke_root)
197 strip = adapter.load_intel_strip()
198 assert "dead" in strip
199 assert "prime" in strip
200
201 def test_propose_stub(self, smoke_root: pathlib.Path) -> None:
202 try:
203 import muse.plugins.knowtation.mcp_helpers as mcp_helpers
204 except ImportError:
205 pytest.skip("knowtation mcp_helpers not importable")
206
207 class _Resp:
208 def read(self, n: int = -1) -> bytes:
209 return b'{"id":"p1","status":"created","proposal_id":"p1"}'
210
211 def __enter__(self):
212 return self
213
214 def __exit__(self, *a: object) -> None:
215 return None
216
217 with patch.object(mcp_helpers.urllib.request, "urlopen", return_value=_Resp()):
218 adapter = KnowtationViewAdapter(smoke_root)
219 r = adapter.propose(SMOKE_NOTE, title="T", scope="personal")
220 assert r.ok or r.ui_state is not None
221
222
223 # ---------------------------------------------------------------------------
224 # Tier 3 — E2E (HTTP + Playwright browser)
225 # ---------------------------------------------------------------------------
226
227
228 @pytest.mark.tier3
229 async def test_vault_routes_gate_non_vault_404(
230 client: AsyncClient,
231 db_session: AsyncSession,
232 ) -> None:
233 from datetime import datetime, timezone
234
235 from musehub.core.genesis import compute_identity_id, compute_repo_id
236 from musehub.db.musehub_repo_models import MusehubRepo
237
238 created = datetime.now(tz=timezone.utc)
239 owner = "codeonly"
240 slug = "code-repo"
241 owner_id = compute_identity_id(owner.encode())
242 repo = MusehubRepo(
243 repo_id=compute_repo_id(owner_id, slug, "code", created.isoformat()),
244 name=slug,
245 owner=owner,
246 slug=slug,
247 visibility="public",
248 owner_user_id=owner_id,
249 created_at=created,
250 updated_at=created,
251 )
252 db_session.add(repo)
253 await db_session.commit()
254 resp = await client.get(f"/{owner}/{slug}/vault/search")
255 assert resp.status_code == 404
256
257
258 @pytest.mark.tier3
259 async def test_vault_search_json_shape(
260 client: AsyncClient,
261 db_session: AsyncSession,
262 smoke_root: pathlib.Path,
263 monkeypatch: pytest.MonkeyPatch,
264 tmp_path: pathlib.Path,
265 ) -> None:
266 repo, _ = await _seed_knowtation_repo(db_session)
267 repo_dir = tmp_path / repo.owner / repo.slug
268 repo_dir.mkdir(parents=True)
269 # Symlink .muse from smoke vault for MCP mount
270 (repo_dir / ".muse").symlink_to(smoke_root / ".muse")
271 for md in smoke_root.rglob("*.md"):
272 rel = md.relative_to(smoke_root)
273 dest = repo_dir / rel
274 dest.parent.mkdir(parents=True, exist_ok=True)
275 dest.write_text(md.read_text(encoding="utf-8"), encoding="utf-8")
276 (repo_dir / ".museattributes").write_text(
277 (smoke_root / ".museattributes").read_text(encoding="utf-8"),
278 encoding="utf-8",
279 )
280 monkeypatch.setattr(
281 "musehub.services.knowtation_view.repo_root.resolve_repo_root",
282 lambda o, s: repo_dir,
283 )
284 resp = await client.get(
285 f"/{repo.owner}/{repo.slug}/vault/search",
286 params={"q": "overview", "format": "json"},
287 )
288 assert resp.status_code == 200
289 data = resp.json()
290 assert "results" in data
291 assert "mode" in data
292
293
294 @pytest.mark.tier3
295 def test_playwright_vault_landing_and_note(
296 smoke_root: pathlib.Path,
297 tmp_path: pathlib.Path,
298 ) -> None:
299 pytest.importorskip("playwright.sync_api")
300 from playwright.sync_api import sync_playwright
301
302 note_raw = (smoke_root / SMOKE_NOTE).read_text(encoding="utf-8")
303 fm, body_html = render_note_html(
304 note_raw,
305 owner="play",
306 repo_slug="vault",
307 note_paths=frozenset({SMOKE_NOTE, "projects/kd-smoke/linked-target.md"}),
308 )
309 landing_html = f"""
310 <html><body>
311 <div class="vault-prime-card">Resume here</div>
312 <div class="vault-intel-strip">Dead notes</div>
313 <ul class="vault-tree"><li>{SMOKE_NOTE}</li></ul>
314 </body></html>
315 """
316 note_html = f"""
317 <html><body>
318 <article class="vault-note-body">{body_html}</article>
319 <section class="vault-backlinks"><a href="/play/vault/note/x">link</a></section>
320 </body></html>
321 """
322 search_html = """
323 <html><body>
324 <ul class="vault-search-results"><li class="vault-search-card"><a href="/n">hit</a></li></ul>
325 </body></html>
326 """
327 diff_html = harmony_diff_html(
328 structured_delta_to_view_model(
329 {"domain": "knowtation", "summary": "s", "ops": [{"op": "insert", "address": "section:2:Summary#1", "content_summary": "x"}]}
330 )
331 )
332 proposal_html = f'<html><body><div class="vault-harmony-section">{diff_html}</div></body></html>'
333
334 with sync_playwright() as p:
335 browser = p.chromium.launch(headless=True)
336 page = browser.new_page()
337 page.set_content(landing_html)
338 assert page.locator(".vault-prime-card").is_visible()
339 assert page.locator(".vault-intel-strip").is_visible()
340 page.set_content(note_html)
341 assert page.locator(".vault-note-body").is_visible()
342 assert "Summary" in page.locator(".vault-note-body").inner_text()
343 page.set_content(search_html)
344 assert page.locator(".vault-search-card a").is_visible()
345 page.set_content(proposal_html)
346 assert page.locator(".vault-harmony-diff").is_visible()
347 browser.close()
348
349
350 # ---------------------------------------------------------------------------
351 # Tier 4 — Stress
352 # ---------------------------------------------------------------------------
353
354
355 @pytest.mark.tier4
356 def test_search_top_k_clamp(smoke_root: pathlib.Path) -> None:
357 adapter = KnowtationViewAdapter(smoke_root)
358 if not adapter.mounted:
359 pytest.skip("knowtation MCP plugin not importable")
360 r = adapter.search("note", mode="lexical", top_k=100)
361 assert r.ok
362 results = r.data.get("results") or []
363 assert len(results) <= 100
364
365
366 # ---------------------------------------------------------------------------
367 # Tier 5 — Data integrity
368 # ---------------------------------------------------------------------------
369
370
371 @pytest.mark.tier5
372 def test_search_scores_bounded(smoke_root: pathlib.Path) -> None:
373 adapter = KnowtationViewAdapter(smoke_root)
374 if not adapter.mounted:
375 pytest.skip("knowtation MCP plugin not importable")
376 r = adapter.search("overview", mode="lexical", top_k=10)
377 assert r.ok
378 for hit in r.data.get("results") or []:
379 if "score" in hit:
380 assert 0.0 <= float(hit["score"]) <= 1.0
381
382
383 @pytest.mark.tier5
384 def test_get_note_hash_matches_bytes(smoke_root: pathlib.Path) -> None:
385 adapter = KnowtationViewAdapter(smoke_root)
386 if not adapter.mounted:
387 pytest.skip("knowtation MCP plugin not importable")
388 r = adapter.get_note(SMOKE_NOTE)
389 assert r.ok
390 on_disk = (smoke_root / SMOKE_NOTE).read_bytes()
391 import hashlib
392
393 assert r.data.get("content_hash") == hashlib.sha256(on_disk).hexdigest()
394
395
396 # ---------------------------------------------------------------------------
397 # Tier 6 — Performance
398 # ---------------------------------------------------------------------------
399
400
401 @pytest.mark.tier6
402 def test_search_lexical_under_2s(smoke_root: pathlib.Path) -> None:
403 adapter = KnowtationViewAdapter(smoke_root)
404 if not adapter.mounted:
405 pytest.skip("knowtation MCP plugin not importable")
406 import time
407
408 t0 = time.monotonic()
409 adapter.search("overview", mode="lexical", top_k=10)
410 assert time.monotonic() - t0 < 2.0
411
412
413 # ---------------------------------------------------------------------------
414 # Tier 7 — Security
415 # ---------------------------------------------------------------------------
416
417
418 @pytest.mark.tier7
419 async def test_note_path_traversal_400(
420 client: AsyncClient,
421 db_session: AsyncSession,
422 tmp_path: pathlib.Path,
423 monkeypatch: pytest.MonkeyPatch,
424 smoke_root: pathlib.Path,
425 ) -> None:
426 repo, _ = await _seed_knowtation_repo(db_session, owner="sec", slug="vault-sec")
427 repo_dir = tmp_path / repo.owner / repo.slug
428 repo_dir.mkdir(parents=True)
429 monkeypatch.setattr(
430 "musehub.services.knowtation_view.repo_root.resolve_repo_root",
431 lambda o, s: repo_dir,
432 )
433 resp = await client.get(f"/{repo.owner}/{repo.slug}/note/../../etc/passwd")
434 assert resp.status_code in (400, 404)
435
436
437 @pytest.mark.tier7
438 def test_html_never_contains_hub_port(smoke_root: pathlib.Path) -> None:
439 adapter = KnowtationViewAdapter(smoke_root)
440 if not adapter.mounted:
441 pytest.skip("knowtation MCP plugin not importable")
442 r = adapter.search("x", mode="lexical")
443 text = str(r.data)
444 assert "KNOWTATION_HUB_PORT" not in text
445 assert "MUSE_ICP_IDENTITY_PEM" not in text
File History 1 commit
sha256:c04414e7e3b812ff02f351d9792c79837b7841133a864f98fcaeff53a2a4afad feat(KD-6b): MuseHub vault UI per frozen §KD-6a Human minor 12 days ago