"""Tests for Phase 4.5 — ``muse/plugins/knowtation/prime_context.py``. Coverage tiers (Rule #0) ------------------------ 1. **Unit** — ``_safe_str``, ``_extract_note_paths_from_delta``, ``ConsolidationRecord.to_dict``, ``HotNote.to_dict``, ``PrimeContext.to_dict``, constants, ``__all__``. 2. **Integration** — ``build_prime_context`` against a real Muse repo with commits written via the store layer. 3. **End-to-end** — ``build_prime_context`` on a repo that contains consolidation commits and hot notes; verify the full payload round-trips through JSON. 4. **Stress** — 200 commits, each touching a random note, with one consolidation inserted mid-stream; verify the walk completes in < 30 s and surfaces the correct hot notes. 5. **Data-integrity** — ``PrimeContext.to_dict()`` is lossless: every field survives a round-trip through ``json.dumps`` / ``json.loads``. 6. **Performance** — 1 000 calls to ``_extract_note_paths_from_delta`` and 1 000 calls to ``_safe_str`` complete in < 500 ms. 7. **Security** — Control-character injection in commit fields is stripped by ``_safe_str``; path traversal in delta addresses is not treated as a note path; ``../`` prefixes are kept verbatim (they come from the commit graph, not user input, but must be safe-stringified). """ from __future__ import annotations import datetime import json import pathlib import time import unittest.mock as mock from typing import Any import pytest from muse.plugins.knowtation.prime_context import ( PRIME_CONTEXT_SCHEMA_VERSION, ConsolidationRecord, HotNote, PrimeContext, _CONSOLIDATION_KINDS, _extract_note_paths_from_delta, _safe_str, build_prime_context, ) # ───────────────────────────────────────────────────────────────────────────── # Helpers / factories # ───────────────────────────────────────────────────────────────────────────── def _make_commit( commit_id: str = "abc123", message: str = "test commit", event_type: str = "", agent_id: str = "", model_id: str = "", structured_delta: dict[str, Any] | None = None, ) -> mock.MagicMock: """Build a minimal mock CommitRecord for unit/integration tests.""" c = mock.MagicMock() c.commit_id = commit_id c.message = message c.agent_id = agent_id c.model_id = model_id c.committed_at = datetime.datetime(2026, 1, 1, 12, 0, 0, tzinfo=datetime.timezone.utc) c.metadata = {"event_type": event_type} if event_type else {} c.structured_delta = structured_delta return c def _delta_with_ops(*addresses: str) -> dict[str, Any]: """Build a minimal StructuredDelta dict with ops for the given addresses.""" return {"ops": [{"address": addr, "kind": "modify"} for addr in addresses]} # ───────────────────────────────────────────────────────────────────────────── # Tier 1 — Unit tests # ───────────────────────────────────────────────────────────────────────────── class TestSafeStr: """Unit tests for ``_safe_str``.""" def test_none_returns_empty_string(self) -> None: assert _safe_str(None) == "" def test_plain_string_passthrough(self) -> None: assert _safe_str("hello world") == "hello world" def test_strips_null_byte(self) -> None: assert _safe_str("abc\x00def") == "abcdef" def test_strips_c0_control_chars(self) -> None: # \x01 through \x1f are C0 assert _safe_str("a\x01b\x1fc") == "abc" def test_strips_del(self) -> None: assert _safe_str("a\x7fb") == "ab" def test_strips_c1_control_chars(self) -> None: assert _safe_str("a\x80b\x9fc") == "abc" def test_strips_multiple_control_chars(self) -> None: assert _safe_str("\x00\x01\x1f\x7f\x80\x9f") == "" def test_non_string_converted_to_string(self) -> None: assert _safe_str(42) == "42" def test_non_string_list(self) -> None: assert _safe_str([1, 2]) == "[1, 2]" def test_unicode_preserved(self) -> None: assert _safe_str("café résumé") == "café résumé" def test_newline_stripped(self) -> None: # \n is \x0a (C0) assert _safe_str("line1\nline2") == "line1line2" def test_tab_stripped(self) -> None: assert _safe_str("col1\tcol2") == "col1col2" class TestExtractNotePathsFromDelta: """Unit tests for ``_extract_note_paths_from_delta``.""" def test_none_returns_empty(self) -> None: assert _extract_note_paths_from_delta(None) == [] def test_empty_ops(self) -> None: assert _extract_note_paths_from_delta({"ops": []}) == [] def test_plain_md_address(self) -> None: result = _extract_note_paths_from_delta(_delta_with_ops("notes/session.md")) assert result == ["notes/session.md"] def test_section_address_extracts_file_path(self) -> None: result = _extract_note_paths_from_delta( _delta_with_ops("notes/session.md::section:1:Intro#0") ) assert result == ["notes/session.md"] def test_non_md_address_ignored(self) -> None: result = _extract_note_paths_from_delta(_delta_with_ops("README.txt")) assert result == [] def test_non_md_section_address_ignored(self) -> None: result = _extract_note_paths_from_delta( _delta_with_ops("code/app.py::section:1:main#0") ) assert result == [] def test_deduplicates_same_path(self) -> None: result = _extract_note_paths_from_delta( _delta_with_ops("notes/a.md", "notes/a.md::section:1:A#0") ) assert result == ["notes/a.md"] def test_multiple_distinct_paths(self) -> None: result = set( _extract_note_paths_from_delta( _delta_with_ops( "notes/a.md", "notes/b.md::section:1:B#0", "notes/c.md", ) ) ) assert result == {"notes/a.md", "notes/b.md", "notes/c.md"} def test_non_string_address_ignored(self) -> None: delta = {"ops": [{"address": 123}, {"address": None}]} assert _extract_note_paths_from_delta(delta) == [] def test_missing_ops_key(self) -> None: assert _extract_note_paths_from_delta({"other": "key"}) == [] def test_control_char_in_path_is_stripped(self) -> None: # Control chars in addresses are sanitised via _safe_str result = _extract_note_paths_from_delta( _delta_with_ops("notes/file\x00.md") ) assert result == ["notes/file.md"] class TestConsolidationRecord: """Unit tests for :class:`ConsolidationRecord`.""" def _make(self, **kwargs: str) -> ConsolidationRecord: defaults = dict( commit_id="sha256:abc", committed_at="2026-01-01T12:00:00+00:00", message="consolidate vault", agent_id="agent-1", model_id="claude-opus-4", ) defaults.update(kwargs) return ConsolidationRecord(**defaults) def test_to_dict_keys(self) -> None: d = self._make().to_dict() assert set(d.keys()) == { "commit_id", "committed_at", "message", "agent_id", "model_id" } def test_to_dict_values_round_trip(self) -> None: rec = self._make() d = rec.to_dict() assert d["commit_id"] == rec.commit_id assert d["committed_at"] == rec.committed_at assert d["message"] == rec.message assert d["agent_id"] == rec.agent_id assert d["model_id"] == rec.model_id def test_frozen(self) -> None: rec = self._make() with pytest.raises((AttributeError, TypeError)): rec.commit_id = "other" # type: ignore[misc] def test_to_dict_is_json_serialisable(self) -> None: d = self._make().to_dict() assert json.loads(json.dumps(d)) == d class TestHotNote: """Unit tests for :class:`HotNote`.""" def test_to_dict(self) -> None: hn = HotNote(path="notes/session.md", edits=7) assert hn.to_dict() == {"path": "notes/session.md", "edits": 7} def test_frozen(self) -> None: hn = HotNote(path="notes/a.md", edits=1) with pytest.raises((AttributeError, TypeError)): hn.path = "other" # type: ignore[misc] def test_to_dict_json_serialisable(self) -> None: d = HotNote(path="notes/x.md", edits=3).to_dict() assert json.loads(json.dumps(d)) == d class TestPrimeContext: """Unit tests for :class:`PrimeContext`.""" def _make(self, **kwargs: Any) -> PrimeContext: defaults: dict[str, Any] = dict( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=5, last_consolidation=None, hot_notes=[], ) defaults.update(kwargs) return PrimeContext(**defaults) def test_to_dict_keys(self) -> None: d = self._make().to_dict() assert set(d.keys()) == { "schema_version", "source", "commits_scanned", "last_consolidation", "hot_notes" } def test_source_default(self) -> None: pc = self._make() assert pc.source == "muse-commit-graph" assert pc.to_dict()["source"] == "muse-commit-graph" def test_last_consolidation_none_serialises_as_null(self) -> None: d = self._make(last_consolidation=None).to_dict() assert d["last_consolidation"] is None def test_last_consolidation_serialised(self) -> None: rec = ConsolidationRecord( commit_id="sha256:abc", committed_at="2026-01-01T12:00:00+00:00", message="consolidate", agent_id="agent-1", model_id="opus", ) d = self._make(last_consolidation=rec).to_dict() assert d["last_consolidation"]["commit_id"] == "sha256:abc" def test_hot_notes_serialised(self) -> None: hot = [HotNote(path="notes/a.md", edits=5), HotNote(path="notes/b.md", edits=3)] d = self._make(hot_notes=hot).to_dict() assert d["hot_notes"] == [ {"path": "notes/a.md", "edits": 5}, {"path": "notes/b.md", "edits": 3}, ] def test_schema_version_constant(self) -> None: pc = self._make() assert pc.schema_version == PRIME_CONTEXT_SCHEMA_VERSION def test_frozen(self) -> None: pc = self._make() with pytest.raises((AttributeError, TypeError)): pc.commits_scanned = 999 # type: ignore[misc] class TestConsolidationKindsConstant: """Unit tests for the ``_CONSOLIDATION_KINDS`` constant.""" def test_consolidation_in_kinds(self) -> None: assert "consolidation" in _CONSOLIDATION_KINDS def test_consolidation_pass_in_kinds(self) -> None: assert "consolidation_pass" in _CONSOLIDATION_KINDS def test_other_kinds_not_in_set(self) -> None: assert "note_write" not in _CONSOLIDATION_KINDS assert "agent_decision" not in _CONSOLIDATION_KINDS def test_is_frozenset(self) -> None: assert isinstance(_CONSOLIDATION_KINDS, frozenset) class TestSchemaVersion: """Unit tests for schema version constant.""" def test_schema_version_format(self) -> None: parts = PRIME_CONTEXT_SCHEMA_VERSION.split(".") assert len(parts) == 3 assert all(p.isdigit() for p in parts) class TestAllExports: """Unit test for ``__all__`` completeness.""" def test_all_exports_importable(self) -> None: import muse.plugins.knowtation.prime_context as mod for name in mod.__all__: assert hasattr(mod, name), f"__all__ lists {name!r} but it is not defined" # ───────────────────────────────────────────────────────────────────────────── # Tier 2 — Integration tests # ───────────────────────────────────────────────────────────────────────────── class TestBuildPrimeContextIntegration: """Integration tests for ``build_prime_context`` against mocked store.""" def _run( self, commits: list[mock.MagicMock], max_commits: int = 100, hot_note_count: int = 10, ) -> PrimeContext: """Invoke build_prime_context with a mocked store layer.""" root = pathlib.Path("/fake/repo") with ( mock.patch( "muse.plugins.knowtation.prime_context.read_current_branch", return_value="feat/test", ), mock.patch( "muse.plugins.knowtation.prime_context.get_head_commit_id", return_value="sha256:head", ), mock.patch( "muse.plugins.knowtation.prime_context.walk_commits_bfs", return_value=(commits, False), ), ): return build_prime_context(root, max_commits=max_commits, hot_note_count=hot_note_count) def test_empty_repo_returns_empty_context(self) -> None: result = self._run([]) assert result.commits_scanned == 0 assert result.last_consolidation is None assert result.hot_notes == [] assert result.schema_version == PRIME_CONTEXT_SCHEMA_VERSION def test_commits_scanned_count(self) -> None: commits = [_make_commit(f"c{i}") for i in range(5)] result = self._run(commits) assert result.commits_scanned == 5 def test_consolidation_event_detected(self) -> None: commits = [ _make_commit("c1", event_type="note_write"), _make_commit("c2", event_type="consolidation", message="big cleanup", agent_id="agent-1", model_id="opus"), _make_commit("c3", event_type="note_read"), ] result = self._run(commits) assert result.last_consolidation is not None assert result.last_consolidation.commit_id == "c2" assert result.last_consolidation.message == "big cleanup" assert result.last_consolidation.agent_id == "agent-1" assert result.last_consolidation.model_id == "opus" def test_consolidation_pass_event_detected(self) -> None: commits = [_make_commit("c1", event_type="consolidation_pass")] result = self._run(commits) assert result.last_consolidation is not None def test_first_consolidation_wins(self) -> None: """The most-recent (first in BFS order) consolidation is captured.""" commits = [ _make_commit("c1", event_type="consolidation", message="newer"), _make_commit("c2", event_type="consolidation", message="older"), ] result = self._run(commits) assert result.last_consolidation is not None assert result.last_consolidation.commit_id == "c1" def test_no_consolidation_returns_none(self) -> None: commits = [_make_commit("c1", event_type="note_write")] result = self._run(commits) assert result.last_consolidation is None def test_hot_notes_sorted_by_edit_count_desc(self) -> None: commits = [ _make_commit("c1", structured_delta=_delta_with_ops("notes/a.md")), _make_commit("c2", structured_delta=_delta_with_ops("notes/a.md")), _make_commit("c3", structured_delta=_delta_with_ops("notes/b.md")), _make_commit("c4", structured_delta=_delta_with_ops("notes/a.md", "notes/b.md")), ] result = self._run(commits) paths = [n.path for n in result.hot_notes] assert paths[0] == "notes/a.md" assert paths[1] == "notes/b.md" def test_hot_note_count_limit(self) -> None: commits = [ _make_commit(f"c{i}", structured_delta=_delta_with_ops(f"notes/{i}.md")) for i in range(20) ] result = self._run(commits, hot_note_count=5) assert len(result.hot_notes) <= 5 def test_no_delta_no_hot_notes(self) -> None: commits = [_make_commit("c1", structured_delta=None)] result = self._run(commits) assert result.hot_notes == [] def test_head_none_returns_zero_scanned(self) -> None: root = pathlib.Path("/fake/repo") with ( mock.patch( "muse.plugins.knowtation.prime_context.read_current_branch", return_value="feat/test", ), mock.patch( "muse.plugins.knowtation.prime_context.get_head_commit_id", return_value=None, ), ): result = build_prime_context(root) assert result.commits_scanned == 0 assert result.last_consolidation is None def test_store_error_returns_empty_context(self) -> None: root = pathlib.Path("/fake/repo") with mock.patch( "muse.plugins.knowtation.prime_context.read_current_branch", side_effect=OSError("not a repo"), ): result = build_prime_context(root) assert result.commits_scanned == 0 assert result.last_consolidation is None # ───────────────────────────────────────────────────────────────────────────── # Tier 3 — End-to-end tests # ───────────────────────────────────────────────────────────────────────────── class TestBuildPrimeContextEndToEnd: """End-to-end tests: JSON payload round-trip and realistic vault scenario.""" def _run(self, commits: list[mock.MagicMock], **kw: Any) -> PrimeContext: root = pathlib.Path("/fake/repo") with ( mock.patch("muse.plugins.knowtation.prime_context.read_current_branch", return_value="main"), mock.patch("muse.plugins.knowtation.prime_context.get_head_commit_id", return_value="sha256:head"), mock.patch("muse.plugins.knowtation.prime_context.walk_commits_bfs", return_value=(commits, False)), ): return build_prime_context(root, **kw) def test_full_payload_json_round_trip(self) -> None: """The full PrimeContext.to_dict() must survive json.dumps/json.loads.""" commits = [ _make_commit( "c1", event_type="consolidation", message="quarterly cleanup", agent_id="agent-42", model_id="claude-opus", structured_delta=_delta_with_ops("notes/session.md"), ), _make_commit( "c2", structured_delta=_delta_with_ops( "notes/session.md", "notes/session.md::section:1:A#0", "notes/ideas.md", ), ), ] result = self._run(commits) payload = result.to_dict() serialised = json.dumps(payload) restored = json.loads(serialised) assert restored["schema_version"] == PRIME_CONTEXT_SCHEMA_VERSION assert restored["source"] == "muse-commit-graph" assert restored["commits_scanned"] == 2 assert restored["last_consolidation"]["commit_id"] == "c1" assert restored["last_consolidation"]["agent_id"] == "agent-42" hot_paths = {n["path"] for n in restored["hot_notes"]} assert "notes/session.md" in hot_paths assert "notes/ideas.md" in hot_paths def test_session_md_most_edited_in_payload(self) -> None: """notes/session.md touched in every commit should rank first in hot_notes.""" commits = [] for i in range(10): addresses = ["notes/session.md"] if i % 3 == 0: addresses.append("notes/ideas.md") commits.append(_make_commit(f"c{i}", structured_delta=_delta_with_ops(*addresses))) result = self._run(commits) assert result.hot_notes[0].path == "notes/session.md" assert result.hot_notes[0].edits == 10 def test_schema_version_in_payload(self) -> None: result = self._run([]) assert result.to_dict()["schema_version"] == "1.0.0" # ───────────────────────────────────────────────────────────────────────────── # Tier 4 — Stress tests # ───────────────────────────────────────────────────────────────────────────── class TestBuildPrimeContextStress: """Stress tests: 200 commits, consolidation mid-stream, < 30 s budget.""" def test_200_commits_with_consolidation(self) -> None: """200 commits with varied deltas; verify correctness and timing.""" commits = [] for i in range(200): event_type = "" if i == 100: event_type = "consolidation" notes = [f"notes/note_{i % 20}.md"] if i % 5 == 0: notes.append(f"notes/bonus_{i % 3}.md") commits.append( _make_commit( f"c{i}", event_type=event_type, structured_delta=_delta_with_ops(*notes), ) ) root = pathlib.Path("/fake/repo") with ( mock.patch("muse.plugins.knowtation.prime_context.read_current_branch", return_value="main"), mock.patch("muse.plugins.knowtation.prime_context.get_head_commit_id", return_value="sha256:head"), mock.patch("muse.plugins.knowtation.prime_context.walk_commits_bfs", return_value=(commits, True)), ): t0 = time.monotonic() result = build_prime_context(root, max_commits=200, hot_note_count=10) elapsed = time.monotonic() - t0 assert elapsed < 30.0, f"build_prime_context took {elapsed:.2f}s for 200 commits" assert result.commits_scanned == 200 assert result.last_consolidation is not None assert result.last_consolidation.commit_id == "c100" assert len(result.hot_notes) <= 10 def test_no_consolidation_among_200_commits(self) -> None: commits = [_make_commit(f"c{i}", event_type="note_write") for i in range(200)] root = pathlib.Path("/fake/repo") with ( mock.patch("muse.plugins.knowtation.prime_context.read_current_branch", return_value="main"), mock.patch("muse.plugins.knowtation.prime_context.get_head_commit_id", return_value="sha256:head"), mock.patch("muse.plugins.knowtation.prime_context.walk_commits_bfs", return_value=(commits, False)), ): result = build_prime_context(root, max_commits=200) assert result.last_consolidation is None assert result.commits_scanned == 200 # ───────────────────────────────────────────────────────────────────────────── # Tier 5 — Data-integrity tests # ───────────────────────────────────────────────────────────────────────────── class TestPrimeContextDataIntegrity: """Data-integrity tests: all field values survive serialisation.""" def test_consolidation_record_to_dict_lossless(self) -> None: rec = ConsolidationRecord( commit_id="sha256:abcdef1234567890", committed_at="2026-05-13T14:30:00+00:00", message="quarterly vault consolidation — 2026-Q2", agent_id="cursor-agent-abc", model_id="claude-opus-4-7-thinking-xhigh", ) d = rec.to_dict() serialised = json.dumps(d) restored = json.loads(serialised) assert restored["commit_id"] == rec.commit_id assert restored["committed_at"] == rec.committed_at assert restored["message"] == rec.message assert restored["agent_id"] == rec.agent_id assert restored["model_id"] == rec.model_id def test_hot_note_to_dict_lossless(self) -> None: hn = HotNote(path="notes/复杂路径/session.md", edits=999) restored = json.loads(json.dumps(hn.to_dict())) assert restored["path"] == hn.path assert restored["edits"] == hn.edits def test_prime_context_to_dict_lossless(self) -> None: rec = ConsolidationRecord( commit_id="c1", committed_at="2026-01-01T00:00:00+00:00", message="clean", agent_id="a", model_id="m", ) hot = [HotNote("notes/a.md", 5), HotNote("notes/b.md", 3)] pc = PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=42, last_consolidation=rec, hot_notes=hot, ) restored = json.loads(json.dumps(pc.to_dict())) assert restored["commits_scanned"] == 42 assert restored["last_consolidation"]["commit_id"] == "c1" assert len(restored["hot_notes"]) == 2 assert restored["hot_notes"][0]["edits"] == 5 def test_empty_context_serialises_cleanly(self) -> None: pc = PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=0, last_consolidation=None, ) restored = json.loads(json.dumps(pc.to_dict())) assert restored["last_consolidation"] is None assert restored["hot_notes"] == [] def test_hot_notes_ordering_preserved(self) -> None: """to_dict() must preserve the hot_notes list order.""" notes = [HotNote(f"notes/{chr(65 + i)}.md", 10 - i) for i in range(5)] pc = PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=5, last_consolidation=None, hot_notes=notes, ) d = pc.to_dict() assert [n["path"] for n in d["hot_notes"]] == [n.path for n in notes] # ───────────────────────────────────────────────────────────────────────────── # Tier 6 — Performance tests # ───────────────────────────────────────────────────────────────────────────── class TestPrimeContextPerformance: """Performance tests: 1 000 calls to helpers complete in < 500 ms.""" def test_safe_str_1000_calls(self) -> None: payloads = [f"commit message {i}\x00\x01\x1f" for i in range(1000)] t0 = time.monotonic() results = [_safe_str(p) for p in payloads] elapsed = time.monotonic() - t0 assert elapsed < 0.5, f"_safe_str 1000 calls took {elapsed:.3f}s" assert all(r for r in results) def test_extract_note_paths_1000_calls(self) -> None: delta = _delta_with_ops( "notes/a.md", "notes/b.md::section:1:B#0", "notes/c.md", "other/file.txt", ) t0 = time.monotonic() for _ in range(1000): paths = _extract_note_paths_from_delta(delta) elapsed = time.monotonic() - t0 assert elapsed < 0.5, f"_extract_note_paths_from_delta 1000 calls took {elapsed:.3f}s" assert set(paths) == {"notes/a.md", "notes/b.md", "notes/c.md"} def test_prime_context_to_dict_1000_calls(self) -> None: rec = ConsolidationRecord( commit_id="c1", committed_at="2026-01-01T00:00:00+00:00", message="clean", agent_id="a", model_id="m", ) pc = PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=50, last_consolidation=rec, hot_notes=[HotNote("notes/a.md", 10)], ) t0 = time.monotonic() for _ in range(1000): pc.to_dict() elapsed = time.monotonic() - t0 assert elapsed < 0.5, f"PrimeContext.to_dict() 1000 calls took {elapsed:.3f}s" # ───────────────────────────────────────────────────────────────────────────── # Tier 7 — Security tests # ───────────────────────────────────────────────────────────────────────────── class TestPrimeContextSecurity: """Security tests: control-char injection, path traversal, malicious deltas.""" def test_control_char_in_commit_message_stripped(self) -> None: """Commit message with control chars must be sanitised in ConsolidationRecord.""" commits = [ _make_commit( "c1", event_type="consolidation", message="clean\x00\x01up\x1f\x7fnow", ) ] root = pathlib.Path("/fake/repo") with ( mock.patch("muse.plugins.knowtation.prime_context.read_current_branch", return_value="main"), mock.patch("muse.plugins.knowtation.prime_context.get_head_commit_id", return_value="sha256:head"), mock.patch("muse.plugins.knowtation.prime_context.walk_commits_bfs", return_value=(commits, False)), ): result = build_prime_context(root) assert result.last_consolidation is not None msg = result.last_consolidation.message # No control characters should remain import re assert not re.search(r"[\x00-\x1f\x7f\x80-\x9f]", msg) assert "cleanup" in msg or "cleanupnow" in msg def test_control_char_in_agent_id_stripped(self) -> None: commits = [_make_commit("c1", event_type="consolidation", agent_id="agent\x00evil")] root = pathlib.Path("/fake/repo") with ( mock.patch("muse.plugins.knowtation.prime_context.read_current_branch", return_value="main"), mock.patch("muse.plugins.knowtation.prime_context.get_head_commit_id", return_value="sha256:head"), mock.patch("muse.plugins.knowtation.prime_context.walk_commits_bfs", return_value=(commits, False)), ): result = build_prime_context(root) assert result.last_consolidation is not None assert "\x00" not in result.last_consolidation.agent_id def test_control_char_in_note_path_stripped(self) -> None: """Note paths extracted from delta ops must have control chars stripped.""" paths = _extract_note_paths_from_delta( _delta_with_ops("notes/evil\x00file.md") ) for p in paths: import re assert not re.search(r"[\x00-\x1f\x7f\x80-\x9f]", p) def test_non_md_traversal_path_not_collected(self) -> None: """Non-.md paths (e.g., ``../../etc/passwd``) are not collected as notes.""" delta = _delta_with_ops("../../etc/passwd", "/etc/hosts", "windows\\system32\\cmd.exe") paths = _extract_note_paths_from_delta(delta) assert paths == [] def test_md_traversal_path_not_escalated(self) -> None: """A *.md path with ``../`` is collected verbatim but safe-stringified. This is acceptable: the path comes from the commit graph (trusted data), and it will be sanitised for display. The key property is that it does not trigger path resolution or filesystem access here. """ delta = _delta_with_ops("../../../vault/private.md") paths = _extract_note_paths_from_delta(delta) # Collected — the function does not do filesystem ops, so no escalation. assert paths == ["../../../vault/private.md"] def test_extremely_long_path_handled(self) -> None: """Very long address strings must not cause errors or buffer overflows.""" long_addr = "notes/" + "a" * 10_000 + ".md" delta = _delta_with_ops(long_addr) paths = _extract_note_paths_from_delta(delta) assert len(paths) == 1 assert paths[0].endswith(".md") def test_deeply_nested_ops_list_handled(self) -> None: """Ops that are not dicts are silently ignored (no AttributeError).""" delta = {"ops": [None, 42, "string", {"address": "notes/valid.md"}]} paths = _extract_note_paths_from_delta(delta) # None, 42, "string" must not raise; "notes/valid.md" is returned assert "notes/valid.md" in paths def test_store_exception_does_not_leak_details(self) -> None: """An OSError from the store must not propagate; empty context returned.""" root = pathlib.Path("/fake/repo") with mock.patch( "muse.plugins.knowtation.prime_context.read_current_branch", side_effect=PermissionError("access denied"), ): result = build_prime_context(root) # Must return a valid (empty) context — no exception escapes. assert isinstance(result, PrimeContext) assert result.commits_scanned == 0 def test_json_output_no_control_chars(self) -> None: """Final JSON output from to_dict() must contain no raw control characters.""" rec = ConsolidationRecord( commit_id="sha256:abc", committed_at="2026-01-01T00:00:00+00:00", message="clean", agent_id="agent-1", model_id="opus", ) pc = PrimeContext( schema_version=PRIME_CONTEXT_SCHEMA_VERSION, commits_scanned=1, last_consolidation=rec, hot_notes=[HotNote("notes/a.md", 1)], ) serialised = json.dumps(pc.to_dict()) import re # json.dumps encodes control chars as \uXXXX, so raw bytes won't appear assert not re.search(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]", serialised)