gabriel / muse public

test_knowtation_cli_hooks.py file-level

at sha256:6 · View file ↗ · Intel ↗

History
1 files
1 commits
0 hotspots
0 🧊 dead
0 💥 blast risk
sha256:0 test: validate staging push and MP workflow for aaronrene · · Jul 7, 2026
1 """Tests for Phase 3.3 — Knowtation CLI hooks (cli_hooks.py).
2
3 Covers the six runners that dispatch ``muse code`` commands to the
4 note-metrics helpers when the active domain is ``knowtation``.
5
6 Test tiers (Rule #0):
7 1. Unit — each runner handles missing-arg defaults via getattr fallbacks
8 2. Integration — JSON-mode output is shape-correct and stable
9 3. Security — runners survive corrupt manifests and missing snapshots
10
11 Notes
12 -----
13 These tests intentionally exercise the runners *without* a real Muse repo
14 context — they monkeypatch the snapshot/commit lookups to use synthetic data,
15 which keeps them fast and isolated from store layer changes.
16 """
17
18 from __future__ import annotations
19
20 import io
21 import json
22 import pathlib
23 import types
24 from contextlib import redirect_stdout
25
26 import pytest
27
28 from muse.plugins.knowtation import cli_hooks
29
30
31 # ============================================================================
32 # Helpers
33 # ============================================================================
34
35
36 def _ns(**kw) -> types.SimpleNamespace:
37 return types.SimpleNamespace(**kw)
38
39
40 def _make_vault(tmp_path: pathlib.Path, files: dict[str, bytes]) -> pathlib.Path:
41 for rel, content in files.items():
42 full = tmp_path / rel
43 full.parent.mkdir(parents=True, exist_ok=True)
44 full.write_bytes(content)
45 return tmp_path
46
47
48 def _capture_run(runner, root: pathlib.Path, **arg_kw) -> str:
49 """Run *runner* with stubbed args; return captured stdout."""
50 buf = io.StringIO()
51 args = _ns(**arg_kw)
52 with redirect_stdout(buf):
53 runner(root, args)
54 return buf.getvalue()
55
56
57 # ============================================================================
58 # TestHotspotsRunner
59 # ============================================================================
60
61
62 class TestHotspotsRunner:
63 def test_runs_with_empty_vault(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
64 monkeypatch.setattr(cli_hooks, "_walk_commits", lambda *_a, **_kw: [])
65 out = _capture_run(
66 cli_hooks.run_hotspots_for_vault, tmp_path,
67 top=20, min_changes=1, as_json=True,
68 )
69 payload = json.loads(out)
70 assert payload["domain"] == "knowtation"
71 assert payload["ranking"] == []
72
73 def test_text_mode_renders(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
74 monkeypatch.setattr(cli_hooks, "_walk_commits", lambda *_a, **_kw: [])
75 out = _capture_run(
76 cli_hooks.run_hotspots_for_vault, tmp_path,
77 top=20, min_changes=1, as_json=False,
78 )
79 assert "Note churn" in out
80
81
82 # ============================================================================
83 # TestGravityRunner
84 # ============================================================================
85
86
87 class TestGravityRunner:
88 def test_empty_vault_json(self, tmp_path: pathlib.Path) -> None:
89 out = _capture_run(
90 cli_hooks.run_gravity_for_vault, tmp_path,
91 top=20, as_json=True,
92 )
93 payload = json.loads(out)
94 assert payload["domain"] == "knowtation"
95 assert payload["ranking"] == []
96
97 def test_hub_vault(self, tmp_path: pathlib.Path) -> None:
98 _make_vault(tmp_path, {
99 "a.md": b"[[hub]]",
100 "b.md": b"[[hub]]",
101 "hub.md": b"# Hub",
102 })
103 out = _capture_run(
104 cli_hooks.run_gravity_for_vault, tmp_path,
105 top=20, as_json=True,
106 )
107 payload = json.loads(out)
108 assert payload["ranking"][0]["path"] == "hub.md"
109
110
111 # ============================================================================
112 # TestDeadRunner
113 # ============================================================================
114
115
116 class TestDeadRunner:
117 def test_isolated_notes_listed(self, tmp_path: pathlib.Path) -> None:
118 _make_vault(tmp_path, {
119 "alone.md": b"# Alone",
120 })
121 out = _capture_run(
122 cli_hooks.run_dead_for_vault, tmp_path,
123 use_mtime=False, mtime_days=180, top=None, as_json=True,
124 )
125 payload = json.loads(out)
126 paths = {n["path"] for n in payload["dead_notes"]}
127 assert "alone.md" in paths
128
129 def test_top_limit_applied(self, tmp_path: pathlib.Path) -> None:
130 for i in range(5):
131 (tmp_path / f"n_{i}.md").write_bytes(b"# Note")
132 out = _capture_run(
133 cli_hooks.run_dead_for_vault, tmp_path,
134 use_mtime=False, mtime_days=180, top=2, as_json=True,
135 )
136 payload = json.loads(out)
137 assert len(payload["dead_notes"]) == 2
138
139
140 # ============================================================================
141 # TestClonesRunner
142 # ============================================================================
143
144
145 class TestClonesRunner:
146 def test_clones_runs_with_stubbed_head(
147 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
148 ) -> None:
149 # Stub the commit/manifest path so we don't need a real repo.
150 from muse.core import store
151 monkeypatch.setattr(cli_hooks, "read_repo_id", lambda _r: "test_repo")
152 monkeypatch.setattr(cli_hooks, "read_current_branch", lambda _r: "main")
153
154 class _Fake:
155 commit_id = "abc"
156
157 monkeypatch.setattr(
158 cli_hooks, "resolve_commit_ref",
159 lambda _r, _id, _b, _ref: _Fake(),
160 )
161 monkeypatch.setattr(
162 cli_hooks, "get_commit_snapshot_manifest",
163 lambda _r, _cid: {"a.md": "h", "b.md": "h", "c.md": "h2"},
164 )
165 out = _capture_run(cli_hooks.run_clones_for_vault, tmp_path, as_json=True)
166 payload = json.loads(out)
167 assert payload["total_clones"] == 2 # a.md and b.md share hash "h"
168
169
170 # ============================================================================
171 # TestEntangleRunner
172 # ============================================================================
173
174
175 class TestEntangleRunner:
176 def test_empty_commits_safe(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
177 monkeypatch.setattr(cli_hooks, "_walk_commits", lambda *_a, **_kw: [])
178 out = _capture_run(
179 cli_hooks.run_entangle_for_vault, tmp_path,
180 top=20, min=2, as_json=True,
181 )
182 payload = json.loads(out)
183 assert payload["ranking"] == []
184
185
186 # ============================================================================
187 # TestVelocityRunner
188 # ============================================================================
189
190
191 class TestVelocityRunner:
192 def test_empty_commits_safe(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
193 monkeypatch.setattr(cli_hooks, "_walk_commits", lambda *_a, **_kw: [])
194 out = _capture_run(
195 cli_hooks.run_velocity_for_vault, tmp_path,
196 window=10, as_json=True,
197 )
198 payload = json.loads(out)
199 assert payload["window_size"] == 10
200 assert payload["acceleration"] == 0
201
202
203 # ============================================================================
204 # TestSnapshotLoaderFactory
205 # ============================================================================
206
207
208 class TestSnapshotLoaderFactory:
209 def test_loader_returns_empty_on_exception(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
210 def bad_loader(_r, _c):
211 raise RuntimeError("simulated")
212
213 monkeypatch.setattr(cli_hooks, "get_commit_snapshot_manifest", bad_loader)
214 loader = cli_hooks._snapshot_loader(tmp_path)
215 assert loader("anything") == {}
216
217
218 # ============================================================================
219 # TestWalkCommitsResilience
220 # ============================================================================
221
222
223 class TestWalkCommitsResilience:
224 def test_walk_returns_empty_on_failure(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
225 # tmp_path is not a Muse repo → read_repo_id will raise.
226 result = cli_hooks._walk_commits(tmp_path)
227 assert result == []