"""Tests for muse/plugins/knowtation/ — Phase 1.1 skeleton. Test tiers covered (per Rule #0 and plan §0.2): - **Unit** — every public method and helper, isolated with tmp_path fixtures. - **Data-integrity** — snapshot round-trips a fixture vault byte-for-byte; manifest hash is stable across repeated builds on the same content. Tiers deferred to later phases: - Integration — CLI invocation via subprocess (Phase 1.4+, needs full init). - End-to-end — Playwright / CLI smoke test (Phase 5.4). - Stress — 10k-note vault rebuild (Phase 3.4). - Performance — p95 diff < 5s on 2k-note vault (Phase 2.1). - Security — injection attacks on frontmatter, mist IDs (Phase 2.2 + 4.2). Structure --------- Each ``class Test`` groups related assertions. Fixtures are ``tmp_path``-scoped so every test runs in an isolated directory. All fixture vaults are minimal in-memory constructions that do **not** require a real ``muse init`` — the plugin's ``snapshot()`` only needs the workdir path; the registry tests only need a ``.muse/repo.json``. """ from __future__ import annotations import hashlib import json import pathlib import pytest from muse._version import __version__ from muse.core.errors import MuseCLIError from muse.core.schema import DomainSchema, SetSchema, SequenceSchema, TreeSchema, MapSchema from muse.domain import MuseDomainPlugin, MergeResult, DriftReport from muse.plugins.knowtation.plugin import KnowtationPlugin, _DOMAIN_NAME, _ALWAYS_IGNORE_DIRS from muse.plugins.knowtation._query import ( file_type_of, group_by_project, is_note, is_valid_mist_id, notes_in_manifest, non_notes_in_manifest, vault_note_count, NOTE_EXTENSIONS, ) from muse.plugins.knowtation.manifest import ( NoteEntry, ProjectManifest, VaultManifest, NoteFileDiff, _count_sections, _is_markdown, _note_type_of, _parse_frontmatter_hash, build_vault_manifest, diff_vault_manifests, read_vault_manifest, write_vault_manifest, ) from muse.plugins.registry import ( registered_domains, resolve_plugin, resolve_plugin_by_domain, schema_for, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _PLUGIN = KnowtationPlugin() _FRONTMATTER_NOTE = """\ --- title: My Research Note date: 2026-05-12 project: AI-Safety tags: [research, alignment] entities: [OpenAI, Anthropic] attachments: [Abc123DefGhi] --- # Introduction This is the first section. ## Methods Details of the approach. ### Sub-method More granular details. """.encode() _PLAIN_NOTE = """\ # A Plain Note No frontmatter here. ## Section One Content. """.encode() _EMPTY_NOTE = b"" _SECRET_NOTE = b"api_key = sk-abc123deadbeef\n# should be ignored by secret patterns" def _make_vault(tmp_path: pathlib.Path) -> pathlib.Path: """Scaffold a minimal vault with a few notes and a config dir.""" (tmp_path / "notes").mkdir() (tmp_path / "notes" / "hello.md").write_bytes(_PLAIN_NOTE) (tmp_path / "notes" / "research.md").write_bytes(_FRONTMATTER_NOTE) (tmp_path / "assets").mkdir() (tmp_path / "assets" / "diagram.svg").write_text("", encoding="utf-8") # Obsidian config dir — must be ignored (tmp_path / ".obsidian").mkdir() (tmp_path / ".obsidian" / "app.json").write_text("{}", encoding="utf-8") # Knowtation index dir — must be ignored (tmp_path / ".knowtation").mkdir() (tmp_path / ".knowtation" / "index.bin").write_text("binary", encoding="utf-8") return tmp_path def _make_repo_json(tmp_path: pathlib.Path, domain: str = "knowtation") -> pathlib.Path: """Create a minimal .muse/repo.json so the registry helpers work.""" muse_dir = tmp_path / ".muse" muse_dir.mkdir(exist_ok=True) (muse_dir / "repo.json").write_text( json.dumps({"repo_id": "test-knowtation", "schema_version": __version__, "domain": domain}), encoding="utf-8", ) return tmp_path def _sha256(content: bytes) -> str: return hashlib.sha256(content).hexdigest() # =========================================================================== # Registry tests # =========================================================================== class TestRegistryResolvesKnowtation: """The plugin registry must return a KnowtationPlugin for domain='knowtation'.""" def test_knowtation_in_registered_domains(self) -> None: assert "knowtation" in registered_domains() def test_registered_domains_sorted(self) -> None: domains = registered_domains() assert domains == sorted(domains) def test_resolve_plugin_returns_knowtation_plugin(self, tmp_path: pathlib.Path) -> None: root = _make_repo_json(tmp_path, domain="knowtation") plugin = resolve_plugin(root) assert isinstance(plugin, KnowtationPlugin) def test_resolve_plugin_by_domain_name(self) -> None: plugin = resolve_plugin_by_domain("knowtation") assert isinstance(plugin, KnowtationPlugin) def test_resolve_plugin_satisfies_protocol(self, tmp_path: pathlib.Path) -> None: root = _make_repo_json(tmp_path, domain="knowtation") plugin = resolve_plugin(root) assert isinstance(plugin, MuseDomainPlugin) def test_schema_for_returns_domain_schema(self) -> None: schema = schema_for("knowtation") assert schema is not None assert schema["domain"] == "knowtation" def test_unknown_domain_still_raises(self, tmp_path: pathlib.Path) -> None: root = _make_repo_json(tmp_path, domain="nonexistent-xyz") with pytest.raises(MuseCLIError, match="nonexistent-xyz"): resolve_plugin(root) # =========================================================================== # Schema tests # =========================================================================== class TestKnowtationSchema: """schema() must return a valid, fully-populated DomainSchema.""" schema = _PLUGIN.schema() def test_domain_is_knowtation(self) -> None: assert self.schema["domain"] == _DOMAIN_NAME def test_merge_mode_is_three_way(self) -> None: assert self.schema["merge_mode"] == "three_way" def test_schema_version_present(self) -> None: assert self.schema["schema_version"] == __version__ def test_description_is_non_empty(self) -> None: assert len(self.schema["description"]) > 0 def test_top_level_is_tree_schema(self) -> None: tl = self.schema["top_level"] assert tl["kind"] == "tree" assert tl["node_type"] == "vault" def test_has_six_dimensions(self) -> None: assert len(self.schema["dimensions"]) == 6 def test_dimension_names(self) -> None: names = [d["name"] for d in self.schema["dimensions"]] assert set(names) == {"notes", "frontmatter", "sections", "links", "entities", "attachments"} def test_notes_dimension_is_tree(self) -> None: notes_dim = next(d for d in self.schema["dimensions"] if d["name"] == "notes") assert notes_dim["schema"]["kind"] == "tree" def test_frontmatter_dimension_is_map(self) -> None: fm_dim = next(d for d in self.schema["dimensions"] if d["name"] == "frontmatter") assert fm_dim["schema"]["kind"] == "map" def test_sections_dimension_is_sequence(self) -> None: sec_dim = next(d for d in self.schema["dimensions"] if d["name"] == "sections") assert sec_dim["schema"]["kind"] == "sequence" assert sec_dim["schema"]["diff_algorithm"] == "myers" def test_links_dimension_is_set(self) -> None: links_dim = next(d for d in self.schema["dimensions"] if d["name"] == "links") assert links_dim["schema"]["kind"] == "set" def test_entities_dimension_is_set(self) -> None: ent_dim = next(d for d in self.schema["dimensions"] if d["name"] == "entities") assert ent_dim["schema"]["kind"] == "set" def test_attachments_dimension_is_set(self) -> None: att_dim = next(d for d in self.schema["dimensions"] if d["name"] == "attachments") assert att_dim["schema"]["kind"] == "set" assert att_dim["schema"]["element_type"] == "mist_id" def test_schema_is_json_serialisable(self) -> None: # DomainSchema must be JSON-round-trippable (TypedDict contract). serialised = json.dumps(self.schema) roundtripped = json.loads(serialised) assert roundtripped["domain"] == "knowtation" def test_independent_merge_flags(self) -> None: # notes dimension is NOT independently mergeable (structural changes # must block); all other five dimensions ARE independent. by_name = {d["name"]: d for d in self.schema["dimensions"]} assert by_name["notes"]["independent_merge"] is False for name in ("frontmatter", "sections", "links", "entities", "attachments"): assert by_name[name]["independent_merge"] is True, f"{name} should be independent_merge=True" def test_schema_stable_across_calls(self) -> None: schema_a = _PLUGIN.schema() schema_b = _PLUGIN.schema() assert json.dumps(schema_a, sort_keys=True) == json.dumps(schema_b, sort_keys=True) # =========================================================================== # Snapshot tests # =========================================================================== class TestKnowtationSnapshot: """snapshot() must walk the vault, respect ignore rules, and return a valid manifest.""" def test_snapshot_returns_snapshot_manifest_type(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert "files" in snap assert "domain" in snap assert snap["domain"] == "knowtation" def test_snapshot_contains_markdown_notes(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert "notes/hello.md" in snap["files"] assert "notes/research.md" in snap["files"] def test_snapshot_contains_non_markdown_assets(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert "assets/diagram.svg" in snap["files"] def test_snapshot_excludes_obsidian_dir(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert not any(p.startswith(".obsidian") for p in snap["files"]) def test_snapshot_excludes_knowtation_index_dir(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert not any(p.startswith(".knowtation") for p in snap["files"]) def test_snapshot_excludes_git_dir(self, tmp_path: pathlib.Path) -> None: (tmp_path / ".git").mkdir() (tmp_path / ".git" / "HEAD").write_text("ref: refs/heads/main") vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert not any(p.startswith(".git") for p in snap["files"]) def test_snapshot_excludes_muse_dir(self, tmp_path: pathlib.Path) -> None: (tmp_path / ".muse").mkdir() (tmp_path / ".muse" / "repo.json").write_text("{}") vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert not any(p.startswith(".muse") for p in snap["files"]) def test_snapshot_hashes_are_sha256(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) for path, content_hash in snap["files"].items(): assert len(content_hash) == 64, f"Hash for {path!r} is not 64 chars" assert all(c in "0123456789abcdef" for c in content_hash), f"Hash for {path!r} is not hex" def test_snapshot_hash_matches_file_content(self, tmp_path: pathlib.Path) -> None: (tmp_path / "check.md").write_bytes(_PLAIN_NOTE) snap = _PLUGIN.snapshot(tmp_path) expected = _sha256(_PLAIN_NOTE) assert snap["files"]["check.md"] == expected def test_snapshot_of_manifest_dict_returns_same_dict(self) -> None: existing = {"files": {"a.md": "abc"}, "domain": "knowtation", "directories": []} result = _PLUGIN.snapshot(existing) # type: ignore[arg-type] assert result is existing def test_snapshot_empty_vault_returns_empty_files(self, tmp_path: pathlib.Path) -> None: snap = _PLUGIN.snapshot(tmp_path) assert snap["files"] == {} def test_snapshot_directories_populated(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) assert "notes" in snap["directories"] assert "assets" in snap["directories"] def test_snapshot_museignore_pattern_respected(self, tmp_path: pathlib.Path) -> None: (tmp_path / "private.md").write_text("secret", encoding="utf-8") (tmp_path / "public.md").write_text("ok", encoding="utf-8") (tmp_path / ".museignore").write_text('[global]\npatterns = ["private.md"]\n', encoding="utf-8") snap = _PLUGIN.snapshot(tmp_path) assert "public.md" in snap["files"] assert "private.md" not in snap["files"] def test_snapshot_domain_specific_museignore(self, tmp_path: pathlib.Path) -> None: (tmp_path / "keep.md").write_text("keep", encoding="utf-8") (tmp_path / "draft.md").write_text("draft", encoding="utf-8") (tmp_path / ".museignore").write_text( '[domain.knowtation]\npatterns = ["draft.md"]\n', encoding="utf-8" ) snap = _PLUGIN.snapshot(tmp_path) assert "keep.md" in snap["files"] assert "draft.md" not in snap["files"] # =========================================================================== # Diff tests # =========================================================================== class TestKnowtationDiff: """diff() must return a StructuredDelta with the correct ops and domain.""" def _snap(self, files: dict[str, str]) -> dict: return {"files": files, "domain": "knowtation", "directories": []} def test_diff_empty_to_empty_has_no_ops(self) -> None: delta = _PLUGIN.diff(self._snap({}), self._snap({})) assert delta["domain"] == "knowtation" assert delta.get("ops", []) == [] def test_diff_added_note(self) -> None: base = self._snap({}) target = self._snap({"notes/new.md": "abc123"}) delta = _PLUGIN.diff(base, target) assert len(delta["ops"]) == 1 assert delta["ops"][0]["op"] == "insert" def test_diff_removed_note(self) -> None: base = self._snap({"notes/old.md": "abc123"}) target = self._snap({}) delta = _PLUGIN.diff(base, target) assert len(delta["ops"]) == 1 assert delta["ops"][0]["op"] == "delete" def test_diff_modified_note_produces_replace(self) -> None: base = self._snap({"notes/a.md": "hash1"}) target = self._snap({"notes/a.md": "hash2"}) delta = _PLUGIN.diff(base, target) ops = delta["ops"] assert len(ops) == 1 assert ops[0]["op"] == "replace" def test_diff_domain_is_knowtation(self) -> None: delta = _PLUGIN.diff(self._snap({"a.md": "x"}), self._snap({"a.md": "y"})) assert delta["domain"] == "knowtation" def test_diff_unchanged_note_produces_no_ops(self) -> None: snap = self._snap({"notes/same.md": "stable_hash"}) delta = _PLUGIN.diff(snap, snap) assert delta.get("ops", []) == [] def test_diff_has_summary(self) -> None: base = self._snap({}) target = self._snap({"notes/new.md": "abc"}) delta = _PLUGIN.diff(base, target) assert isinstance(delta.get("summary", ""), str) # =========================================================================== # Merge tests # =========================================================================== class TestKnowtationMerge: """merge() must implement clean three-way logic at file granularity.""" def _snap(self, files: dict[str, str]) -> dict: return {"files": files, "domain": "knowtation", "directories": []} def test_merge_both_sides_agree(self) -> None: base = self._snap({"a.md": "v1"}) left = self._snap({"a.md": "v2"}) right = self._snap({"a.md": "v2"}) result = _PLUGIN.merge(base, left, right) assert result.conflicts == [] assert result.merged["files"]["a.md"] == "v2" def test_merge_only_left_changed(self) -> None: base = self._snap({"a.md": "v1"}) left = self._snap({"a.md": "v2"}) right = self._snap({"a.md": "v1"}) result = _PLUGIN.merge(base, left, right) assert result.conflicts == [] assert result.merged["files"]["a.md"] == "v2" def test_merge_only_right_changed(self) -> None: base = self._snap({"a.md": "v1"}) left = self._snap({"a.md": "v1"}) right = self._snap({"a.md": "v3"}) result = _PLUGIN.merge(base, left, right) assert result.conflicts == [] assert result.merged["files"]["a.md"] == "v3" def test_merge_both_changed_differently_is_conflict(self) -> None: base = self._snap({"a.md": "v1"}) left = self._snap({"a.md": "v2"}) right = self._snap({"a.md": "v3"}) result = _PLUGIN.merge(base, left, right) assert "a.md" in result.conflicts def test_merge_left_adds_right_unchanged_no_conflict(self) -> None: base = self._snap({}) left = self._snap({"new.md": "x"}) right = self._snap({}) result = _PLUGIN.merge(base, left, right) assert result.conflicts == [] assert "new.md" in result.merged["files"] def test_merge_both_delete_same_note(self) -> None: base = self._snap({"del.md": "v1"}) left = self._snap({}) right = self._snap({}) result = _PLUGIN.merge(base, left, right) assert result.conflicts == [] assert "del.md" not in result.merged["files"] def test_merge_result_domain_is_knowtation(self) -> None: base = self._snap({}) result = _PLUGIN.merge(base, base, base) assert result.merged["domain"] == "knowtation" def test_merge_returns_merge_result_instance(self) -> None: base = self._snap({"a.md": "v"}) result = _PLUGIN.merge(base, base, base) assert isinstance(result, MergeResult) def test_merge_is_clean_when_all_sides_identical(self) -> None: snap = self._snap({"note.md": "x"}) result = _PLUGIN.merge(snap, snap, snap) assert result.is_clean # =========================================================================== # Drift tests # =========================================================================== class TestKnowtationDrift: """drift() must correctly detect uncommitted changes in the vault.""" def test_drift_no_changes_returns_no_drift(self, tmp_path: pathlib.Path) -> None: (tmp_path / "note.md").write_bytes(_PLAIN_NOTE) snap = _PLUGIN.snapshot(tmp_path) report = _PLUGIN.drift(snap, tmp_path) assert not report.has_drift def test_drift_added_file_detected(self, tmp_path: pathlib.Path) -> None: snap = _PLUGIN.snapshot(tmp_path) # empty vault (tmp_path / "new.md").write_bytes(_PLAIN_NOTE) report = _PLUGIN.drift(snap, tmp_path) assert report.has_drift def test_drift_modified_file_detected(self, tmp_path: pathlib.Path) -> None: (tmp_path / "note.md").write_bytes(_PLAIN_NOTE) snap = _PLUGIN.snapshot(tmp_path) (tmp_path / "note.md").write_bytes(_FRONTMATTER_NOTE) report = _PLUGIN.drift(snap, tmp_path) assert report.has_drift def test_drift_deleted_file_detected(self, tmp_path: pathlib.Path) -> None: (tmp_path / "note.md").write_bytes(_PLAIN_NOTE) snap = _PLUGIN.snapshot(tmp_path) (tmp_path / "note.md").unlink() report = _PLUGIN.drift(snap, tmp_path) assert report.has_drift def test_drift_returns_drift_report(self, tmp_path: pathlib.Path) -> None: snap = _PLUGIN.snapshot(tmp_path) report = _PLUGIN.drift(snap, tmp_path) assert isinstance(report, DriftReport) def test_drift_summary_is_string(self, tmp_path: pathlib.Path) -> None: snap = _PLUGIN.snapshot(tmp_path) report = _PLUGIN.drift(snap, tmp_path) assert isinstance(report.summary, str) # =========================================================================== # Apply tests # =========================================================================== class TestKnowtationApply: """apply() must be a pass-through in Phase 1.1.""" def test_apply_returns_live_state_unchanged(self, tmp_path: pathlib.Path) -> None: delta = {"domain": "knowtation", "ops": [], "summary": ""} result = _PLUGIN.apply(delta, tmp_path) assert result is tmp_path def test_apply_returns_snapshot_manifest_unchanged(self) -> None: manifest = {"files": {}, "domain": "knowtation", "directories": []} result = _PLUGIN.apply({}, manifest) # type: ignore[arg-type] assert result is manifest # =========================================================================== # Data-integrity: snapshot round-trip # =========================================================================== class TestSnapshotRoundTrip: """Snapshot must be byte-for-byte reproducible on the same vault content. This is the data-integrity tier: verifies that repeated calls to ``snapshot()`` on an unmodified vault return identical manifests, and that the content hashes match the actual bytes on disk. """ def test_snapshot_is_deterministic(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap_a = _PLUGIN.snapshot(vault) snap_b = _PLUGIN.snapshot(vault) assert snap_a["files"] == snap_b["files"] assert snap_a["domain"] == snap_b["domain"] def test_snapshot_hashes_match_disk_bytes(self, tmp_path: pathlib.Path) -> None: vault = _make_vault(tmp_path) snap = _PLUGIN.snapshot(vault) for rel_path, recorded_hash in snap["files"].items(): disk_path = vault / rel_path assert disk_path.is_file(), f"{rel_path} tracked but not on disk" actual_hash = hashlib.sha256(disk_path.read_bytes()).hexdigest() assert actual_hash == recorded_hash, ( f"Hash mismatch for {rel_path}: " f"recorded={recorded_hash!r} actual={actual_hash!r}" ) def test_snapshot_after_write_reflects_new_content(self, tmp_path: pathlib.Path) -> None: note_path = tmp_path / "evolving.md" note_path.write_bytes(_PLAIN_NOTE) snap_before = _PLUGIN.snapshot(tmp_path) note_path.write_bytes(_FRONTMATTER_NOTE) snap_after = _PLUGIN.snapshot(tmp_path) assert snap_before["files"]["evolving.md"] != snap_after["files"]["evolving.md"] def test_snapshot_stable_for_five_note_fixture(self, tmp_path: pathlib.Path) -> None: notes = [ ("project-a/note1.md", b"# Note 1\n\nContent."), ("project-a/note2.md", b"# Note 2\n\nContent."), ("project-b/note3.md", b"---\ntitle: Note 3\n---\n# Note 3"), ("note4.md", b"# Root Note"), ("docs/readme.md", b"# README"), ] for rel, content in notes: p = tmp_path / rel p.parent.mkdir(parents=True, exist_ok=True) p.write_bytes(content) snapshots = [_PLUGIN.snapshot(tmp_path)["files"] for _ in range(3)] assert snapshots[0] == snapshots[1] == snapshots[2] # =========================================================================== # Data-integrity: VaultManifest # =========================================================================== class TestVaultManifestIntegrity: """Manifest builds and persistence must be byte-for-byte stable.""" def _make_flat_manifest(self) -> dict[str, str]: notes = { "notes/alpha.md": _sha256(_FRONTMATTER_NOTE), "notes/beta.md": _sha256(_PLAIN_NOTE), "assets/logo.svg": _sha256(b""), } return notes def test_build_is_deterministic(self, tmp_path: pathlib.Path) -> None: flat = self._make_flat_manifest() m1 = build_vault_manifest("snap-001", flat, tmp_path) m2 = build_vault_manifest("snap-001", flat, tmp_path) assert m1["manifest_hash"] == m2["manifest_hash"] def test_manifest_hash_changes_with_content(self, tmp_path: pathlib.Path) -> None: # build_vault_manifest validates SHA-256 IDs via the object store; use # real digests (the objects won't be found, which is fine — the builder # gracefully skips missing blobs and still produces a valid manifest). hash_a = _sha256(b"content-a") hash_b = _sha256(b"content-b") flat_a = {"notes/a.md": hash_a} flat_b = {"notes/a.md": hash_b} m_a = build_vault_manifest("s1", flat_a, tmp_path) m_b = build_vault_manifest("s1", flat_b, tmp_path) assert m_a["manifest_hash"] != m_b["manifest_hash"] def test_write_and_read_roundtrip(self, tmp_path: pathlib.Path) -> None: (tmp_path / ".muse").mkdir() flat = self._make_flat_manifest() m = build_vault_manifest("snap-abc", flat, tmp_path) write_vault_manifest(tmp_path, m) loaded = read_vault_manifest(tmp_path, m["manifest_hash"]) assert loaded is not None assert loaded["manifest_hash"] == m["manifest_hash"] assert loaded["total_notes"] == m["total_notes"] def test_read_returns_none_for_missing_hash(self, tmp_path: pathlib.Path) -> None: result = read_vault_manifest(tmp_path, "nonexistent" + "0" * 53) assert result is None def test_total_notes_count(self, tmp_path: pathlib.Path) -> None: flat = self._make_flat_manifest() m = build_vault_manifest("s", flat, tmp_path) assert m["total_notes"] == 3 def test_markdown_notes_count(self, tmp_path: pathlib.Path) -> None: flat = self._make_flat_manifest() m = build_vault_manifest("s", flat, tmp_path) assert m["markdown_notes"] == 2 # alpha.md + beta.md def test_diff_manifests_detects_added(self, tmp_path: pathlib.Path) -> None: base_flat: dict[str, str] = {} target_flat = {"notes/new.md": _sha256(_PLAIN_NOTE)} base_m = build_vault_manifest("s1", base_flat, tmp_path) target_m = build_vault_manifest("s2", target_flat, tmp_path) diffs = diff_vault_manifests(base_m, target_m) assert len(diffs) == 1 assert diffs[0]["change"] == "added" assert diffs[0]["path"] == "notes/new.md" def test_diff_manifests_detects_removed(self, tmp_path: pathlib.Path) -> None: base_flat = {"notes/old.md": _sha256(_PLAIN_NOTE)} target_flat: dict[str, str] = {} base_m = build_vault_manifest("s1", base_flat, tmp_path) target_m = build_vault_manifest("s2", target_flat, tmp_path) diffs = diff_vault_manifests(base_m, target_m) assert len(diffs) == 1 assert diffs[0]["change"] == "removed" def test_diff_manifests_unchanged_notes_omitted(self, tmp_path: pathlib.Path) -> None: flat = {"notes/stable.md": _sha256(_PLAIN_NOTE)} m = build_vault_manifest("s", flat, tmp_path) diffs = diff_vault_manifests(m, m) assert diffs == [] # =========================================================================== # _query.py unit tests # =========================================================================== class TestFileTypeOf: """file_type_of() must return the correct label for each extension.""" @pytest.mark.parametrize("path,expected", [ ("notes/hello.md", "Markdown"), ("notes/hello.markdown", "Markdown"), ("notes/hello.mdx", "Markdown"), ("assets/photo.png", "Image"), ("assets/photo.jpg", "Image"), ("assets/photo.webp", "Image"), ("data/config.yaml", "YAML"), ("data/config.yml", "YAML"), ("scripts/run.sh", "Shell"), ("docs/spec.rst", "reStructuredText"), ("data/index.json", "JSON"), ("config/settings.toml", "TOML"), ("pages/index.html", "HTML"), ("styles/main.css", "CSS"), ("docs/diagram.svg", "SVG"), ("media/audio.mp3", "Audio"), ("media/video.mp4", "Video"), ("doc/report.pdf", "PDF"), ("unknown/file.xyz", ".xyz"), ("no_ext", "(no ext)"), ]) def test_classification(self, path: str, expected: str) -> None: assert file_type_of(path) == expected class TestIsNote: """is_note() must return True only for Markdown extensions.""" @pytest.mark.parametrize("path,expected", [ ("notes/hello.md", True), ("notes/hello.markdown", True), ("notes/hello.mdx", True), ("notes/HELLO.MD", True), # case-insensitive ("image.png", False), ("script.py", False), ("data.json", False), ("archive.pdf", False), ("no_ext", False), ]) def test_classification(self, path: str, expected: bool) -> None: assert is_note(path) == expected class TestNotesInManifest: """notes_in_manifest() and non_notes_in_manifest() must filter correctly.""" _manifest = { "notes/a.md": "hash1", "notes/b.markdown": "hash2", "assets/img.png": "hash3", "scripts/run.sh": "hash4", } def test_notes_only_markdown(self) -> None: result = notes_in_manifest(self._manifest) assert set(result) == {"notes/a.md", "notes/b.markdown"} def test_non_notes_excludes_markdown(self) -> None: result = non_notes_in_manifest(self._manifest) assert set(result) == {"assets/img.png", "scripts/run.sh"} def test_vault_note_count(self) -> None: assert vault_note_count(self._manifest) == 2 def test_notes_empty_manifest(self) -> None: assert notes_in_manifest({}) == {} def test_non_notes_all_markdown(self) -> None: assert non_notes_in_manifest({"a.md": "x", "b.md": "y"}) == {} class TestGroupByProject: """group_by_project() must group notes by their top-level directory.""" def test_single_level_grouping(self) -> None: manifest = { "project-a/note1.md": "h1", "project-a/note2.md": "h2", "project-b/note3.md": "h3", } groups = group_by_project(manifest) assert set(groups["project-a"]) == {"project-a/note1.md", "project-a/note2.md"} assert groups["project-b"] == ["project-b/note3.md"] def test_root_notes_under_root_key(self) -> None: manifest = {"root-note.md": "h"} groups = group_by_project(manifest) assert "__root__" in groups assert "root-note.md" in groups["__root__"] def test_non_markdown_excluded(self) -> None: manifest = {"project-a/note.md": "h", "project-a/img.png": "x"} groups = group_by_project(manifest) assert "project-a/img.png" not in groups.get("project-a", []) def test_empty_manifest_returns_empty(self) -> None: assert group_by_project({}) == {} class TestIsValidMistId: """is_valid_mist_id() must accept exactly 12-char base58 strings.""" @pytest.mark.parametrize("mist_id,expected", [ ("Abc123DefGhi", True), # valid base58, 12 chars ("111111111111", True), # all '1' is valid base58 ("ZZZZZZZZZzzz", True), # mixed case valid chars ("", False), # too short ("Abc123DefGh", False), # 11 chars ("Abc123DefGhiX", False), # 13 chars ("Abc123DefGh0", False), # contains '0' — not in base58 ("Abc123DefGhO", False), # contains 'O' — not in base58 ("Abc123DefGhI", False), # contains 'I' — not in base58 ("Abc123DefGhl", False), # contains 'l' — not in base58 ]) def test_validation(self, mist_id: str, expected: bool) -> None: assert is_valid_mist_id(mist_id) == expected # =========================================================================== # manifest.py unit tests — frontmatter and section parsing # =========================================================================== class TestParseFrontmatterHash: """_parse_frontmatter_hash() must extract hash and project correctly.""" def test_note_with_frontmatter_returns_non_empty_hash(self) -> None: fm_hash, project = _parse_frontmatter_hash(_FRONTMATTER_NOTE) assert len(fm_hash) == 64 assert project == "AI-Safety" def test_plain_note_returns_empty_strings(self) -> None: fm_hash, project = _parse_frontmatter_hash(_PLAIN_NOTE) assert fm_hash == "" assert project == "" def test_empty_content_returns_empty_strings(self) -> None: fm_hash, project = _parse_frontmatter_hash(_EMPTY_NOTE) assert fm_hash == "" assert project == "" def test_hash_stable_for_same_content(self) -> None: h1, _ = _parse_frontmatter_hash(_FRONTMATTER_NOTE) h2, _ = _parse_frontmatter_hash(_FRONTMATTER_NOTE) assert h1 == h2 def test_different_frontmatter_produces_different_hash(self) -> None: note_a = b"---\ntitle: A\n---\n# Body" note_b = b"---\ntitle: B\n---\n# Body" h_a, _ = _parse_frontmatter_hash(note_a) h_b, _ = _parse_frontmatter_hash(note_b) assert h_a != h_b def test_project_extraction(self) -> None: note = b"---\nproject: My-Project\ntitle: Test\n---\n# Test" _, project = _parse_frontmatter_hash(note) assert project == "My-Project" def test_no_project_key_returns_empty_string(self) -> None: note = b"---\ntitle: No project here\n---\n# Test" _, project = _parse_frontmatter_hash(note) assert project == "" class TestCountSections: """_count_sections() must count ATX headings correctly.""" @pytest.mark.parametrize("content,expected", [ (_FRONTMATTER_NOTE, 3), # # Introduction, ## Methods, ### Sub-method (_PLAIN_NOTE, 2), # # A Plain Note, ## Section One (_EMPTY_NOTE, 0), (b"No headings here.\n", 0), (b"# H1\n## H2\n### H3\n", 3), (b"#NoSpace", 0), # '#NoSpace' is NOT a valid ATX heading (b" # Indented not heading\n", 0), # indented, not ATX (b"# H1\n## H2\n", 2), ]) def test_count(self, content: bytes, expected: int) -> None: assert _count_sections(content) == expected class TestNoteTypeOf: """_note_type_of() and _is_markdown() must classify files correctly.""" @pytest.mark.parametrize("path,expected_type,expected_md", [ ("notes/a.md", "markdown", True), ("notes/a.markdown", "markdown", True), ("notes/a.mdx", "markdown", True), ("notes/A.MD", "markdown", True), # case-insensitive ("image.png", "other", False), ("data.json", "other", False), ("script.py", "other", False), ("no_ext", "other", False), ]) def test_type_and_markdown_flag( self, path: str, expected_type: str, expected_md: bool ) -> None: assert _note_type_of(path) == expected_type assert _is_markdown(path) == expected_md # =========================================================================== # Always-ignore-dirs contract test # =========================================================================== class TestAlwaysIgnoreDirs: """_ALWAYS_IGNORE_DIRS must contain the vault-specific dirs.""" @pytest.mark.parametrize("dirname", [ ".obsidian", ".logseq", ".trash", ".knowtation", ".git", ".muse", "node_modules", "__pycache__", ]) def test_always_ignored(self, dirname: str) -> None: assert dirname in _ALWAYS_IGNORE_DIRS, ( f"{dirname!r} should be in _ALWAYS_IGNORE_DIRS" )