gabriel / muse public
test_knowtation_plugin.py python
929 lines 36.0 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Tests for muse/plugins/knowtation/ — Phase 1.1 skeleton.
2
3 Test tiers covered (per Rule #0 and plan §0.2):
4
5 - **Unit** — every public method and helper, isolated with tmp_path fixtures.
6 - **Data-integrity** — snapshot round-trips a fixture vault byte-for-byte; manifest
7 hash is stable across repeated builds on the same content.
8
9 Tiers deferred to later phases:
10 - Integration — CLI invocation via subprocess (Phase 1.4+, needs full init).
11 - End-to-end — Playwright / CLI smoke test (Phase 5.4).
12 - Stress — 10k-note vault rebuild (Phase 3.4).
13 - Performance — p95 diff < 5s on 2k-note vault (Phase 2.1).
14 - Security — injection attacks on frontmatter, mist IDs (Phase 2.2 + 4.2).
15
16 Structure
17 ---------
18 Each ``class Test<Subject>`` groups related assertions. Fixtures are
19 ``tmp_path``-scoped so every test runs in an isolated directory.
20
21 All fixture vaults are minimal in-memory constructions that do **not** require
22 a real ``muse init`` — the plugin's ``snapshot()`` only needs the workdir path;
23 the registry tests only need a ``.muse/repo.json``.
24 """
25
26 from __future__ import annotations
27
28 import hashlib
29 import json
30 import pathlib
31
32 import pytest
33
34 from muse._version import __version__
35 from muse.core.errors import MuseCLIError
36 from muse.core.schema import DomainSchema, SetSchema, SequenceSchema, TreeSchema, MapSchema
37 from muse.domain import MuseDomainPlugin, MergeResult, DriftReport
38 from muse.plugins.knowtation.plugin import KnowtationPlugin, _DOMAIN_NAME, _ALWAYS_IGNORE_DIRS
39 from muse.plugins.knowtation._query import (
40 file_type_of,
41 group_by_project,
42 is_note,
43 is_valid_mist_id,
44 notes_in_manifest,
45 non_notes_in_manifest,
46 vault_note_count,
47 NOTE_EXTENSIONS,
48 )
49 from muse.plugins.knowtation.manifest import (
50 NoteEntry,
51 ProjectManifest,
52 VaultManifest,
53 NoteFileDiff,
54 _count_sections,
55 _is_markdown,
56 _note_type_of,
57 _parse_frontmatter_hash,
58 build_vault_manifest,
59 diff_vault_manifests,
60 read_vault_manifest,
61 write_vault_manifest,
62 )
63 from muse.plugins.registry import (
64 registered_domains,
65 resolve_plugin,
66 resolve_plugin_by_domain,
67 schema_for,
68 )
69
70
71 # ---------------------------------------------------------------------------
72 # Helpers
73 # ---------------------------------------------------------------------------
74
75 _PLUGIN = KnowtationPlugin()
76
77 _FRONTMATTER_NOTE = """\
78 ---
79 title: My Research Note
80 date: 2026-05-12
81 project: AI-Safety
82 tags: [research, alignment]
83 entities: [OpenAI, Anthropic]
84 attachments: [Abc123DefGhi]
85 ---
86
87 # Introduction
88
89 This is the first section.
90
91 ## Methods
92
93 Details of the approach.
94
95 ### Sub-method
96
97 More granular details.
98 """.encode()
99
100 _PLAIN_NOTE = """\
101 # A Plain Note
102
103 No frontmatter here.
104
105 ## Section One
106
107 Content.
108 """.encode()
109
110 _EMPTY_NOTE = b""
111
112 _SECRET_NOTE = b"api_key = sk-abc123deadbeef\n# should be ignored by secret patterns"
113
114
115 def _make_vault(tmp_path: pathlib.Path) -> pathlib.Path:
116 """Scaffold a minimal vault with a few notes and a config dir."""
117 (tmp_path / "notes").mkdir()
118 (tmp_path / "notes" / "hello.md").write_bytes(_PLAIN_NOTE)
119 (tmp_path / "notes" / "research.md").write_bytes(_FRONTMATTER_NOTE)
120 (tmp_path / "assets").mkdir()
121 (tmp_path / "assets" / "diagram.svg").write_text("<svg/>", encoding="utf-8")
122 # Obsidian config dir — must be ignored
123 (tmp_path / ".obsidian").mkdir()
124 (tmp_path / ".obsidian" / "app.json").write_text("{}", encoding="utf-8")
125 # Knowtation index dir — must be ignored
126 (tmp_path / ".knowtation").mkdir()
127 (tmp_path / ".knowtation" / "index.bin").write_text("binary", encoding="utf-8")
128 return tmp_path
129
130
131 def _make_repo_json(tmp_path: pathlib.Path, domain: str = "knowtation") -> pathlib.Path:
132 """Create a minimal .muse/repo.json so the registry helpers work."""
133 muse_dir = tmp_path / ".muse"
134 muse_dir.mkdir(exist_ok=True)
135 (muse_dir / "repo.json").write_text(
136 json.dumps({"repo_id": "test-knowtation", "schema_version": __version__, "domain": domain}),
137 encoding="utf-8",
138 )
139 return tmp_path
140
141
142 def _sha256(content: bytes) -> str:
143 return hashlib.sha256(content).hexdigest()
144
145
146 # ===========================================================================
147 # Registry tests
148 # ===========================================================================
149
150
151 class TestRegistryResolvesKnowtation:
152 """The plugin registry must return a KnowtationPlugin for domain='knowtation'."""
153
154 def test_knowtation_in_registered_domains(self) -> None:
155 assert "knowtation" in registered_domains()
156
157 def test_registered_domains_sorted(self) -> None:
158 domains = registered_domains()
159 assert domains == sorted(domains)
160
161 def test_resolve_plugin_returns_knowtation_plugin(self, tmp_path: pathlib.Path) -> None:
162 root = _make_repo_json(tmp_path, domain="knowtation")
163 plugin = resolve_plugin(root)
164 assert isinstance(plugin, KnowtationPlugin)
165
166 def test_resolve_plugin_by_domain_name(self) -> None:
167 plugin = resolve_plugin_by_domain("knowtation")
168 assert isinstance(plugin, KnowtationPlugin)
169
170 def test_resolve_plugin_satisfies_protocol(self, tmp_path: pathlib.Path) -> None:
171 root = _make_repo_json(tmp_path, domain="knowtation")
172 plugin = resolve_plugin(root)
173 assert isinstance(plugin, MuseDomainPlugin)
174
175 def test_schema_for_returns_domain_schema(self) -> None:
176 schema = schema_for("knowtation")
177 assert schema is not None
178 assert schema["domain"] == "knowtation"
179
180 def test_unknown_domain_still_raises(self, tmp_path: pathlib.Path) -> None:
181 root = _make_repo_json(tmp_path, domain="nonexistent-xyz")
182 with pytest.raises(MuseCLIError, match="nonexistent-xyz"):
183 resolve_plugin(root)
184
185
186 # ===========================================================================
187 # Schema tests
188 # ===========================================================================
189
190
191 class TestKnowtationSchema:
192 """schema() must return a valid, fully-populated DomainSchema."""
193
194 schema = _PLUGIN.schema()
195
196 def test_domain_is_knowtation(self) -> None:
197 assert self.schema["domain"] == _DOMAIN_NAME
198
199 def test_merge_mode_is_three_way(self) -> None:
200 assert self.schema["merge_mode"] == "three_way"
201
202 def test_schema_version_present(self) -> None:
203 assert self.schema["schema_version"] == __version__
204
205 def test_description_is_non_empty(self) -> None:
206 assert len(self.schema["description"]) > 0
207
208 def test_top_level_is_tree_schema(self) -> None:
209 tl = self.schema["top_level"]
210 assert tl["kind"] == "tree"
211 assert tl["node_type"] == "vault"
212
213 def test_has_six_dimensions(self) -> None:
214 assert len(self.schema["dimensions"]) == 6
215
216 def test_dimension_names(self) -> None:
217 names = [d["name"] for d in self.schema["dimensions"]]
218 assert set(names) == {"notes", "frontmatter", "sections", "links", "entities", "attachments"}
219
220 def test_notes_dimension_is_tree(self) -> None:
221 notes_dim = next(d for d in self.schema["dimensions"] if d["name"] == "notes")
222 assert notes_dim["schema"]["kind"] == "tree"
223
224 def test_frontmatter_dimension_is_map(self) -> None:
225 fm_dim = next(d for d in self.schema["dimensions"] if d["name"] == "frontmatter")
226 assert fm_dim["schema"]["kind"] == "map"
227
228 def test_sections_dimension_is_sequence(self) -> None:
229 sec_dim = next(d for d in self.schema["dimensions"] if d["name"] == "sections")
230 assert sec_dim["schema"]["kind"] == "sequence"
231 assert sec_dim["schema"]["diff_algorithm"] == "myers"
232
233 def test_links_dimension_is_set(self) -> None:
234 links_dim = next(d for d in self.schema["dimensions"] if d["name"] == "links")
235 assert links_dim["schema"]["kind"] == "set"
236
237 def test_entities_dimension_is_set(self) -> None:
238 ent_dim = next(d for d in self.schema["dimensions"] if d["name"] == "entities")
239 assert ent_dim["schema"]["kind"] == "set"
240
241 def test_attachments_dimension_is_set(self) -> None:
242 att_dim = next(d for d in self.schema["dimensions"] if d["name"] == "attachments")
243 assert att_dim["schema"]["kind"] == "set"
244 assert att_dim["schema"]["element_type"] == "mist_id"
245
246 def test_schema_is_json_serialisable(self) -> None:
247 # DomainSchema must be JSON-round-trippable (TypedDict contract).
248 serialised = json.dumps(self.schema)
249 roundtripped = json.loads(serialised)
250 assert roundtripped["domain"] == "knowtation"
251
252 def test_independent_merge_flags(self) -> None:
253 # notes dimension is NOT independently mergeable (structural changes
254 # must block); all other five dimensions ARE independent.
255 by_name = {d["name"]: d for d in self.schema["dimensions"]}
256 assert by_name["notes"]["independent_merge"] is False
257 for name in ("frontmatter", "sections", "links", "entities", "attachments"):
258 assert by_name[name]["independent_merge"] is True, f"{name} should be independent_merge=True"
259
260 def test_schema_stable_across_calls(self) -> None:
261 schema_a = _PLUGIN.schema()
262 schema_b = _PLUGIN.schema()
263 assert json.dumps(schema_a, sort_keys=True) == json.dumps(schema_b, sort_keys=True)
264
265
266 # ===========================================================================
267 # Snapshot tests
268 # ===========================================================================
269
270
271 class TestKnowtationSnapshot:
272 """snapshot() must walk the vault, respect ignore rules, and return a valid manifest."""
273
274 def test_snapshot_returns_snapshot_manifest_type(self, tmp_path: pathlib.Path) -> None:
275 vault = _make_vault(tmp_path)
276 snap = _PLUGIN.snapshot(vault)
277 assert "files" in snap
278 assert "domain" in snap
279 assert snap["domain"] == "knowtation"
280
281 def test_snapshot_contains_markdown_notes(self, tmp_path: pathlib.Path) -> None:
282 vault = _make_vault(tmp_path)
283 snap = _PLUGIN.snapshot(vault)
284 assert "notes/hello.md" in snap["files"]
285 assert "notes/research.md" in snap["files"]
286
287 def test_snapshot_contains_non_markdown_assets(self, tmp_path: pathlib.Path) -> None:
288 vault = _make_vault(tmp_path)
289 snap = _PLUGIN.snapshot(vault)
290 assert "assets/diagram.svg" in snap["files"]
291
292 def test_snapshot_excludes_obsidian_dir(self, tmp_path: pathlib.Path) -> None:
293 vault = _make_vault(tmp_path)
294 snap = _PLUGIN.snapshot(vault)
295 assert not any(p.startswith(".obsidian") for p in snap["files"])
296
297 def test_snapshot_excludes_knowtation_index_dir(self, tmp_path: pathlib.Path) -> None:
298 vault = _make_vault(tmp_path)
299 snap = _PLUGIN.snapshot(vault)
300 assert not any(p.startswith(".knowtation") for p in snap["files"])
301
302 def test_snapshot_excludes_git_dir(self, tmp_path: pathlib.Path) -> None:
303 (tmp_path / ".git").mkdir()
304 (tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/main")
305 vault = _make_vault(tmp_path)
306 snap = _PLUGIN.snapshot(vault)
307 assert not any(p.startswith(".git") for p in snap["files"])
308
309 def test_snapshot_excludes_muse_dir(self, tmp_path: pathlib.Path) -> None:
310 (tmp_path / ".muse").mkdir()
311 (tmp_path / ".muse" / "repo.json").write_text("{}")
312 vault = _make_vault(tmp_path)
313 snap = _PLUGIN.snapshot(vault)
314 assert not any(p.startswith(".muse") for p in snap["files"])
315
316 def test_snapshot_hashes_are_sha256(self, tmp_path: pathlib.Path) -> None:
317 vault = _make_vault(tmp_path)
318 snap = _PLUGIN.snapshot(vault)
319 for path, content_hash in snap["files"].items():
320 assert len(content_hash) == 64, f"Hash for {path!r} is not 64 chars"
321 assert all(c in "0123456789abcdef" for c in content_hash), f"Hash for {path!r} is not hex"
322
323 def test_snapshot_hash_matches_file_content(self, tmp_path: pathlib.Path) -> None:
324 (tmp_path / "check.md").write_bytes(_PLAIN_NOTE)
325 snap = _PLUGIN.snapshot(tmp_path)
326 expected = _sha256(_PLAIN_NOTE)
327 assert snap["files"]["check.md"] == expected
328
329 def test_snapshot_of_manifest_dict_returns_same_dict(self) -> None:
330 existing = {"files": {"a.md": "abc"}, "domain": "knowtation", "directories": []}
331 result = _PLUGIN.snapshot(existing) # type: ignore[arg-type]
332 assert result is existing
333
334 def test_snapshot_empty_vault_returns_empty_files(self, tmp_path: pathlib.Path) -> None:
335 snap = _PLUGIN.snapshot(tmp_path)
336 assert snap["files"] == {}
337
338 def test_snapshot_directories_populated(self, tmp_path: pathlib.Path) -> None:
339 vault = _make_vault(tmp_path)
340 snap = _PLUGIN.snapshot(vault)
341 assert "notes" in snap["directories"]
342 assert "assets" in snap["directories"]
343
344 def test_snapshot_museignore_pattern_respected(self, tmp_path: pathlib.Path) -> None:
345 (tmp_path / "private.md").write_text("secret", encoding="utf-8")
346 (tmp_path / "public.md").write_text("ok", encoding="utf-8")
347 (tmp_path / ".museignore").write_text('[global]\npatterns = ["private.md"]\n', encoding="utf-8")
348 snap = _PLUGIN.snapshot(tmp_path)
349 assert "public.md" in snap["files"]
350 assert "private.md" not in snap["files"]
351
352 def test_snapshot_domain_specific_museignore(self, tmp_path: pathlib.Path) -> None:
353 (tmp_path / "keep.md").write_text("keep", encoding="utf-8")
354 (tmp_path / "draft.md").write_text("draft", encoding="utf-8")
355 (tmp_path / ".museignore").write_text(
356 '[domain.knowtation]\npatterns = ["draft.md"]\n', encoding="utf-8"
357 )
358 snap = _PLUGIN.snapshot(tmp_path)
359 assert "keep.md" in snap["files"]
360 assert "draft.md" not in snap["files"]
361
362
363 # ===========================================================================
364 # Diff tests
365 # ===========================================================================
366
367
368 class TestKnowtationDiff:
369 """diff() must return a StructuredDelta with the correct ops and domain."""
370
371 def _snap(self, files: dict[str, str]) -> dict:
372 return {"files": files, "domain": "knowtation", "directories": []}
373
374 def test_diff_empty_to_empty_has_no_ops(self) -> None:
375 delta = _PLUGIN.diff(self._snap({}), self._snap({}))
376 assert delta["domain"] == "knowtation"
377 assert delta.get("ops", []) == []
378
379 def test_diff_added_note(self) -> None:
380 base = self._snap({})
381 target = self._snap({"notes/new.md": "abc123"})
382 delta = _PLUGIN.diff(base, target)
383 assert len(delta["ops"]) == 1
384 assert delta["ops"][0]["op"] == "insert"
385
386 def test_diff_removed_note(self) -> None:
387 base = self._snap({"notes/old.md": "abc123"})
388 target = self._snap({})
389 delta = _PLUGIN.diff(base, target)
390 assert len(delta["ops"]) == 1
391 assert delta["ops"][0]["op"] == "delete"
392
393 def test_diff_modified_note_produces_replace(self) -> None:
394 base = self._snap({"notes/a.md": "hash1"})
395 target = self._snap({"notes/a.md": "hash2"})
396 delta = _PLUGIN.diff(base, target)
397 ops = delta["ops"]
398 assert len(ops) == 1
399 assert ops[0]["op"] == "replace"
400
401 def test_diff_domain_is_knowtation(self) -> None:
402 delta = _PLUGIN.diff(self._snap({"a.md": "x"}), self._snap({"a.md": "y"}))
403 assert delta["domain"] == "knowtation"
404
405 def test_diff_unchanged_note_produces_no_ops(self) -> None:
406 snap = self._snap({"notes/same.md": "stable_hash"})
407 delta = _PLUGIN.diff(snap, snap)
408 assert delta.get("ops", []) == []
409
410 def test_diff_has_summary(self) -> None:
411 base = self._snap({})
412 target = self._snap({"notes/new.md": "abc"})
413 delta = _PLUGIN.diff(base, target)
414 assert isinstance(delta.get("summary", ""), str)
415
416
417 # ===========================================================================
418 # Merge tests
419 # ===========================================================================
420
421
422 class TestKnowtationMerge:
423 """merge() must implement clean three-way logic at file granularity."""
424
425 def _snap(self, files: dict[str, str]) -> dict:
426 return {"files": files, "domain": "knowtation", "directories": []}
427
428 def test_merge_both_sides_agree(self) -> None:
429 base = self._snap({"a.md": "v1"})
430 left = self._snap({"a.md": "v2"})
431 right = self._snap({"a.md": "v2"})
432 result = _PLUGIN.merge(base, left, right)
433 assert result.conflicts == []
434 assert result.merged["files"]["a.md"] == "v2"
435
436 def test_merge_only_left_changed(self) -> None:
437 base = self._snap({"a.md": "v1"})
438 left = self._snap({"a.md": "v2"})
439 right = self._snap({"a.md": "v1"})
440 result = _PLUGIN.merge(base, left, right)
441 assert result.conflicts == []
442 assert result.merged["files"]["a.md"] == "v2"
443
444 def test_merge_only_right_changed(self) -> None:
445 base = self._snap({"a.md": "v1"})
446 left = self._snap({"a.md": "v1"})
447 right = self._snap({"a.md": "v3"})
448 result = _PLUGIN.merge(base, left, right)
449 assert result.conflicts == []
450 assert result.merged["files"]["a.md"] == "v3"
451
452 def test_merge_both_changed_differently_is_conflict(self) -> None:
453 base = self._snap({"a.md": "v1"})
454 left = self._snap({"a.md": "v2"})
455 right = self._snap({"a.md": "v3"})
456 result = _PLUGIN.merge(base, left, right)
457 assert "a.md" in result.conflicts
458
459 def test_merge_left_adds_right_unchanged_no_conflict(self) -> None:
460 base = self._snap({})
461 left = self._snap({"new.md": "x"})
462 right = self._snap({})
463 result = _PLUGIN.merge(base, left, right)
464 assert result.conflicts == []
465 assert "new.md" in result.merged["files"]
466
467 def test_merge_both_delete_same_note(self) -> None:
468 base = self._snap({"del.md": "v1"})
469 left = self._snap({})
470 right = self._snap({})
471 result = _PLUGIN.merge(base, left, right)
472 assert result.conflicts == []
473 assert "del.md" not in result.merged["files"]
474
475 def test_merge_result_domain_is_knowtation(self) -> None:
476 base = self._snap({})
477 result = _PLUGIN.merge(base, base, base)
478 assert result.merged["domain"] == "knowtation"
479
480 def test_merge_returns_merge_result_instance(self) -> None:
481 base = self._snap({"a.md": "v"})
482 result = _PLUGIN.merge(base, base, base)
483 assert isinstance(result, MergeResult)
484
485 def test_merge_is_clean_when_all_sides_identical(self) -> None:
486 snap = self._snap({"note.md": "x"})
487 result = _PLUGIN.merge(snap, snap, snap)
488 assert result.is_clean
489
490
491 # ===========================================================================
492 # Drift tests
493 # ===========================================================================
494
495
496 class TestKnowtationDrift:
497 """drift() must correctly detect uncommitted changes in the vault."""
498
499 def test_drift_no_changes_returns_no_drift(self, tmp_path: pathlib.Path) -> None:
500 (tmp_path / "note.md").write_bytes(_PLAIN_NOTE)
501 snap = _PLUGIN.snapshot(tmp_path)
502 report = _PLUGIN.drift(snap, tmp_path)
503 assert not report.has_drift
504
505 def test_drift_added_file_detected(self, tmp_path: pathlib.Path) -> None:
506 snap = _PLUGIN.snapshot(tmp_path) # empty vault
507 (tmp_path / "new.md").write_bytes(_PLAIN_NOTE)
508 report = _PLUGIN.drift(snap, tmp_path)
509 assert report.has_drift
510
511 def test_drift_modified_file_detected(self, tmp_path: pathlib.Path) -> None:
512 (tmp_path / "note.md").write_bytes(_PLAIN_NOTE)
513 snap = _PLUGIN.snapshot(tmp_path)
514 (tmp_path / "note.md").write_bytes(_FRONTMATTER_NOTE)
515 report = _PLUGIN.drift(snap, tmp_path)
516 assert report.has_drift
517
518 def test_drift_deleted_file_detected(self, tmp_path: pathlib.Path) -> None:
519 (tmp_path / "note.md").write_bytes(_PLAIN_NOTE)
520 snap = _PLUGIN.snapshot(tmp_path)
521 (tmp_path / "note.md").unlink()
522 report = _PLUGIN.drift(snap, tmp_path)
523 assert report.has_drift
524
525 def test_drift_returns_drift_report(self, tmp_path: pathlib.Path) -> None:
526 snap = _PLUGIN.snapshot(tmp_path)
527 report = _PLUGIN.drift(snap, tmp_path)
528 assert isinstance(report, DriftReport)
529
530 def test_drift_summary_is_string(self, tmp_path: pathlib.Path) -> None:
531 snap = _PLUGIN.snapshot(tmp_path)
532 report = _PLUGIN.drift(snap, tmp_path)
533 assert isinstance(report.summary, str)
534
535
536 # ===========================================================================
537 # Apply tests
538 # ===========================================================================
539
540
541 class TestKnowtationApply:
542 """apply() must be a pass-through in Phase 1.1."""
543
544 def test_apply_returns_live_state_unchanged(self, tmp_path: pathlib.Path) -> None:
545 delta = {"domain": "knowtation", "ops": [], "summary": ""}
546 result = _PLUGIN.apply(delta, tmp_path)
547 assert result is tmp_path
548
549 def test_apply_returns_snapshot_manifest_unchanged(self) -> None:
550 manifest = {"files": {}, "domain": "knowtation", "directories": []}
551 result = _PLUGIN.apply({}, manifest) # type: ignore[arg-type]
552 assert result is manifest
553
554
555 # ===========================================================================
556 # Data-integrity: snapshot round-trip
557 # ===========================================================================
558
559
560 class TestSnapshotRoundTrip:
561 """Snapshot must be byte-for-byte reproducible on the same vault content.
562
563 This is the data-integrity tier: verifies that repeated calls to
564 ``snapshot()`` on an unmodified vault return identical manifests, and that
565 the content hashes match the actual bytes on disk.
566 """
567
568 def test_snapshot_is_deterministic(self, tmp_path: pathlib.Path) -> None:
569 vault = _make_vault(tmp_path)
570 snap_a = _PLUGIN.snapshot(vault)
571 snap_b = _PLUGIN.snapshot(vault)
572 assert snap_a["files"] == snap_b["files"]
573 assert snap_a["domain"] == snap_b["domain"]
574
575 def test_snapshot_hashes_match_disk_bytes(self, tmp_path: pathlib.Path) -> None:
576 vault = _make_vault(tmp_path)
577 snap = _PLUGIN.snapshot(vault)
578 for rel_path, recorded_hash in snap["files"].items():
579 disk_path = vault / rel_path
580 assert disk_path.is_file(), f"{rel_path} tracked but not on disk"
581 actual_hash = hashlib.sha256(disk_path.read_bytes()).hexdigest()
582 assert actual_hash == recorded_hash, (
583 f"Hash mismatch for {rel_path}: "
584 f"recorded={recorded_hash!r} actual={actual_hash!r}"
585 )
586
587 def test_snapshot_after_write_reflects_new_content(self, tmp_path: pathlib.Path) -> None:
588 note_path = tmp_path / "evolving.md"
589 note_path.write_bytes(_PLAIN_NOTE)
590 snap_before = _PLUGIN.snapshot(tmp_path)
591
592 note_path.write_bytes(_FRONTMATTER_NOTE)
593 snap_after = _PLUGIN.snapshot(tmp_path)
594
595 assert snap_before["files"]["evolving.md"] != snap_after["files"]["evolving.md"]
596
597 def test_snapshot_stable_for_five_note_fixture(self, tmp_path: pathlib.Path) -> None:
598 notes = [
599 ("project-a/note1.md", b"# Note 1\n\nContent."),
600 ("project-a/note2.md", b"# Note 2\n\nContent."),
601 ("project-b/note3.md", b"---\ntitle: Note 3\n---\n# Note 3"),
602 ("note4.md", b"# Root Note"),
603 ("docs/readme.md", b"# README"),
604 ]
605 for rel, content in notes:
606 p = tmp_path / rel
607 p.parent.mkdir(parents=True, exist_ok=True)
608 p.write_bytes(content)
609
610 snapshots = [_PLUGIN.snapshot(tmp_path)["files"] for _ in range(3)]
611 assert snapshots[0] == snapshots[1] == snapshots[2]
612
613
614 # ===========================================================================
615 # Data-integrity: VaultManifest
616 # ===========================================================================
617
618
619 class TestVaultManifestIntegrity:
620 """Manifest builds and persistence must be byte-for-byte stable."""
621
622 def _make_flat_manifest(self) -> dict[str, str]:
623 notes = {
624 "notes/alpha.md": _sha256(_FRONTMATTER_NOTE),
625 "notes/beta.md": _sha256(_PLAIN_NOTE),
626 "assets/logo.svg": _sha256(b"<svg/>"),
627 }
628 return notes
629
630 def test_build_is_deterministic(self, tmp_path: pathlib.Path) -> None:
631 flat = self._make_flat_manifest()
632 m1 = build_vault_manifest("snap-001", flat, tmp_path)
633 m2 = build_vault_manifest("snap-001", flat, tmp_path)
634 assert m1["manifest_hash"] == m2["manifest_hash"]
635
636 def test_manifest_hash_changes_with_content(self, tmp_path: pathlib.Path) -> None:
637 # build_vault_manifest validates SHA-256 IDs via the object store; use
638 # real digests (the objects won't be found, which is fine — the builder
639 # gracefully skips missing blobs and still produces a valid manifest).
640 hash_a = _sha256(b"content-a")
641 hash_b = _sha256(b"content-b")
642 flat_a = {"notes/a.md": hash_a}
643 flat_b = {"notes/a.md": hash_b}
644 m_a = build_vault_manifest("s1", flat_a, tmp_path)
645 m_b = build_vault_manifest("s1", flat_b, tmp_path)
646 assert m_a["manifest_hash"] != m_b["manifest_hash"]
647
648 def test_write_and_read_roundtrip(self, tmp_path: pathlib.Path) -> None:
649 (tmp_path / ".muse").mkdir()
650 flat = self._make_flat_manifest()
651 m = build_vault_manifest("snap-abc", flat, tmp_path)
652 write_vault_manifest(tmp_path, m)
653 loaded = read_vault_manifest(tmp_path, m["manifest_hash"])
654 assert loaded is not None
655 assert loaded["manifest_hash"] == m["manifest_hash"]
656 assert loaded["total_notes"] == m["total_notes"]
657
658 def test_read_returns_none_for_missing_hash(self, tmp_path: pathlib.Path) -> None:
659 result = read_vault_manifest(tmp_path, "nonexistent" + "0" * 53)
660 assert result is None
661
662 def test_total_notes_count(self, tmp_path: pathlib.Path) -> None:
663 flat = self._make_flat_manifest()
664 m = build_vault_manifest("s", flat, tmp_path)
665 assert m["total_notes"] == 3
666
667 def test_markdown_notes_count(self, tmp_path: pathlib.Path) -> None:
668 flat = self._make_flat_manifest()
669 m = build_vault_manifest("s", flat, tmp_path)
670 assert m["markdown_notes"] == 2 # alpha.md + beta.md
671
672 def test_diff_manifests_detects_added(self, tmp_path: pathlib.Path) -> None:
673 base_flat: dict[str, str] = {}
674 target_flat = {"notes/new.md": _sha256(_PLAIN_NOTE)}
675 base_m = build_vault_manifest("s1", base_flat, tmp_path)
676 target_m = build_vault_manifest("s2", target_flat, tmp_path)
677 diffs = diff_vault_manifests(base_m, target_m)
678 assert len(diffs) == 1
679 assert diffs[0]["change"] == "added"
680 assert diffs[0]["path"] == "notes/new.md"
681
682 def test_diff_manifests_detects_removed(self, tmp_path: pathlib.Path) -> None:
683 base_flat = {"notes/old.md": _sha256(_PLAIN_NOTE)}
684 target_flat: dict[str, str] = {}
685 base_m = build_vault_manifest("s1", base_flat, tmp_path)
686 target_m = build_vault_manifest("s2", target_flat, tmp_path)
687 diffs = diff_vault_manifests(base_m, target_m)
688 assert len(diffs) == 1
689 assert diffs[0]["change"] == "removed"
690
691 def test_diff_manifests_unchanged_notes_omitted(self, tmp_path: pathlib.Path) -> None:
692 flat = {"notes/stable.md": _sha256(_PLAIN_NOTE)}
693 m = build_vault_manifest("s", flat, tmp_path)
694 diffs = diff_vault_manifests(m, m)
695 assert diffs == []
696
697
698 # ===========================================================================
699 # _query.py unit tests
700 # ===========================================================================
701
702
703 class TestFileTypeOf:
704 """file_type_of() must return the correct label for each extension."""
705
706 @pytest.mark.parametrize("path,expected", [
707 ("notes/hello.md", "Markdown"),
708 ("notes/hello.markdown", "Markdown"),
709 ("notes/hello.mdx", "Markdown"),
710 ("assets/photo.png", "Image"),
711 ("assets/photo.jpg", "Image"),
712 ("assets/photo.webp", "Image"),
713 ("data/config.yaml", "YAML"),
714 ("data/config.yml", "YAML"),
715 ("scripts/run.sh", "Shell"),
716 ("docs/spec.rst", "reStructuredText"),
717 ("data/index.json", "JSON"),
718 ("config/settings.toml", "TOML"),
719 ("pages/index.html", "HTML"),
720 ("styles/main.css", "CSS"),
721 ("docs/diagram.svg", "SVG"),
722 ("media/audio.mp3", "Audio"),
723 ("media/video.mp4", "Video"),
724 ("doc/report.pdf", "PDF"),
725 ("unknown/file.xyz", ".xyz"),
726 ("no_ext", "(no ext)"),
727 ])
728 def test_classification(self, path: str, expected: str) -> None:
729 assert file_type_of(path) == expected
730
731
732 class TestIsNote:
733 """is_note() must return True only for Markdown extensions."""
734
735 @pytest.mark.parametrize("path,expected", [
736 ("notes/hello.md", True),
737 ("notes/hello.markdown", True),
738 ("notes/hello.mdx", True),
739 ("notes/HELLO.MD", True), # case-insensitive
740 ("image.png", False),
741 ("script.py", False),
742 ("data.json", False),
743 ("archive.pdf", False),
744 ("no_ext", False),
745 ])
746 def test_classification(self, path: str, expected: bool) -> None:
747 assert is_note(path) == expected
748
749
750 class TestNotesInManifest:
751 """notes_in_manifest() and non_notes_in_manifest() must filter correctly."""
752
753 _manifest = {
754 "notes/a.md": "hash1",
755 "notes/b.markdown": "hash2",
756 "assets/img.png": "hash3",
757 "scripts/run.sh": "hash4",
758 }
759
760 def test_notes_only_markdown(self) -> None:
761 result = notes_in_manifest(self._manifest)
762 assert set(result) == {"notes/a.md", "notes/b.markdown"}
763
764 def test_non_notes_excludes_markdown(self) -> None:
765 result = non_notes_in_manifest(self._manifest)
766 assert set(result) == {"assets/img.png", "scripts/run.sh"}
767
768 def test_vault_note_count(self) -> None:
769 assert vault_note_count(self._manifest) == 2
770
771 def test_notes_empty_manifest(self) -> None:
772 assert notes_in_manifest({}) == {}
773
774 def test_non_notes_all_markdown(self) -> None:
775 assert non_notes_in_manifest({"a.md": "x", "b.md": "y"}) == {}
776
777
778 class TestGroupByProject:
779 """group_by_project() must group notes by their top-level directory."""
780
781 def test_single_level_grouping(self) -> None:
782 manifest = {
783 "project-a/note1.md": "h1",
784 "project-a/note2.md": "h2",
785 "project-b/note3.md": "h3",
786 }
787 groups = group_by_project(manifest)
788 assert set(groups["project-a"]) == {"project-a/note1.md", "project-a/note2.md"}
789 assert groups["project-b"] == ["project-b/note3.md"]
790
791 def test_root_notes_under_root_key(self) -> None:
792 manifest = {"root-note.md": "h"}
793 groups = group_by_project(manifest)
794 assert "__root__" in groups
795 assert "root-note.md" in groups["__root__"]
796
797 def test_non_markdown_excluded(self) -> None:
798 manifest = {"project-a/note.md": "h", "project-a/img.png": "x"}
799 groups = group_by_project(manifest)
800 assert "project-a/img.png" not in groups.get("project-a", [])
801
802 def test_empty_manifest_returns_empty(self) -> None:
803 assert group_by_project({}) == {}
804
805
806 class TestIsValidMistId:
807 """is_valid_mist_id() must accept exactly 12-char base58 strings."""
808
809 @pytest.mark.parametrize("mist_id,expected", [
810 ("Abc123DefGhi", True), # valid base58, 12 chars
811 ("111111111111", True), # all '1' is valid base58
812 ("ZZZZZZZZZzzz", True), # mixed case valid chars
813 ("", False), # too short
814 ("Abc123DefGh", False), # 11 chars
815 ("Abc123DefGhiX", False), # 13 chars
816 ("Abc123DefGh0", False), # contains '0' — not in base58
817 ("Abc123DefGhO", False), # contains 'O' — not in base58
818 ("Abc123DefGhI", False), # contains 'I' — not in base58
819 ("Abc123DefGhl", False), # contains 'l' — not in base58
820 ])
821 def test_validation(self, mist_id: str, expected: bool) -> None:
822 assert is_valid_mist_id(mist_id) == expected
823
824
825 # ===========================================================================
826 # manifest.py unit tests — frontmatter and section parsing
827 # ===========================================================================
828
829
830 class TestParseFrontmatterHash:
831 """_parse_frontmatter_hash() must extract hash and project correctly."""
832
833 def test_note_with_frontmatter_returns_non_empty_hash(self) -> None:
834 fm_hash, project = _parse_frontmatter_hash(_FRONTMATTER_NOTE)
835 assert len(fm_hash) == 64
836 assert project == "AI-Safety"
837
838 def test_plain_note_returns_empty_strings(self) -> None:
839 fm_hash, project = _parse_frontmatter_hash(_PLAIN_NOTE)
840 assert fm_hash == ""
841 assert project == ""
842
843 def test_empty_content_returns_empty_strings(self) -> None:
844 fm_hash, project = _parse_frontmatter_hash(_EMPTY_NOTE)
845 assert fm_hash == ""
846 assert project == ""
847
848 def test_hash_stable_for_same_content(self) -> None:
849 h1, _ = _parse_frontmatter_hash(_FRONTMATTER_NOTE)
850 h2, _ = _parse_frontmatter_hash(_FRONTMATTER_NOTE)
851 assert h1 == h2
852
853 def test_different_frontmatter_produces_different_hash(self) -> None:
854 note_a = b"---\ntitle: A\n---\n# Body"
855 note_b = b"---\ntitle: B\n---\n# Body"
856 h_a, _ = _parse_frontmatter_hash(note_a)
857 h_b, _ = _parse_frontmatter_hash(note_b)
858 assert h_a != h_b
859
860 def test_project_extraction(self) -> None:
861 note = b"---\nproject: My-Project\ntitle: Test\n---\n# Test"
862 _, project = _parse_frontmatter_hash(note)
863 assert project == "My-Project"
864
865 def test_no_project_key_returns_empty_string(self) -> None:
866 note = b"---\ntitle: No project here\n---\n# Test"
867 _, project = _parse_frontmatter_hash(note)
868 assert project == ""
869
870
871 class TestCountSections:
872 """_count_sections() must count ATX headings correctly."""
873
874 @pytest.mark.parametrize("content,expected", [
875 (_FRONTMATTER_NOTE, 3), # # Introduction, ## Methods, ### Sub-method
876 (_PLAIN_NOTE, 2), # # A Plain Note, ## Section One
877 (_EMPTY_NOTE, 0),
878 (b"No headings here.\n", 0),
879 (b"# H1\n## H2\n### H3\n", 3),
880 (b"#NoSpace", 0), # '#NoSpace' is NOT a valid ATX heading
881 (b" # Indented not heading\n", 0), # indented, not ATX
882 (b"# H1\n## H2\n", 2),
883 ])
884 def test_count(self, content: bytes, expected: int) -> None:
885 assert _count_sections(content) == expected
886
887
888 class TestNoteTypeOf:
889 """_note_type_of() and _is_markdown() must classify files correctly."""
890
891 @pytest.mark.parametrize("path,expected_type,expected_md", [
892 ("notes/a.md", "markdown", True),
893 ("notes/a.markdown", "markdown", True),
894 ("notes/a.mdx", "markdown", True),
895 ("notes/A.MD", "markdown", True), # case-insensitive
896 ("image.png", "other", False),
897 ("data.json", "other", False),
898 ("script.py", "other", False),
899 ("no_ext", "other", False),
900 ])
901 def test_type_and_markdown_flag(
902 self, path: str, expected_type: str, expected_md: bool
903 ) -> None:
904 assert _note_type_of(path) == expected_type
905 assert _is_markdown(path) == expected_md
906
907
908 # ===========================================================================
909 # Always-ignore-dirs contract test
910 # ===========================================================================
911
912
913 class TestAlwaysIgnoreDirs:
914 """_ALWAYS_IGNORE_DIRS must contain the vault-specific dirs."""
915
916 @pytest.mark.parametrize("dirname", [
917 ".obsidian",
918 ".logseq",
919 ".trash",
920 ".knowtation",
921 ".git",
922 ".muse",
923 "node_modules",
924 "__pycache__",
925 ])
926 def test_always_ignored(self, dirname: str) -> None:
927 assert dirname in _ALWAYS_IGNORE_DIRS, (
928 f"{dirname!r} should be in _ALWAYS_IGNORE_DIRS"
929 )
File History 2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago