test_kd6_vault_ui.py
python
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