gabriel / muse public
test_cmd_read_commit.py python
434 lines 17.8 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 146 days ago
1 """Comprehensive tests for ``muse read-commit``.
2
3 Coverage tiers
4 --------------
5 - Unit: _ALL_FIELDS completeness
6 - Integration: JSON/text format, prefix resolution, --fields filter, parent chain
7 - Security: ANSI in branch/author/message stripped in text mode
8 - Stress: 200 sequential reads, --fields on large schema
9 """
10 from __future__ import annotations
11
12 import datetime
13 import json
14 import pathlib
15
16 from muse.core.errors import ExitCode
17 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
18 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
19 from tests.cli_test_helper import CliRunner, InvokeResult
20
21 runner = CliRunner()
22
23 # Module-level constants so every test uses the same deterministic inputs.
24 _SNAP_ID: str = compute_snapshot_id({})
25 _COMMITTED_AT: datetime.datetime = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
26
27
28 # ---------------------------------------------------------------------------
29 # Helpers
30 # ---------------------------------------------------------------------------
31
32 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
33 repo = tmp_path / "repo"
34 muse = repo / ".muse"
35 for sub in ("objects", "commits", "snapshots", "refs/heads"):
36 (muse / sub).mkdir(parents=True)
37 (muse / "HEAD").write_text("ref: refs/heads/main")
38 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
39 return repo
40
41
42 def _commit(
43 repo: pathlib.Path,
44 *,
45 branch: str = "main",
46 message: str = "test commit",
47 author: str = "tester",
48 parent: str | None = None,
49 agent_id: str = "",
50 model_id: str = "",
51 ) -> str:
52 """Write a commit with a real content-addressed ID; return the commit_id."""
53 parent_ids: list[str] = [parent] if parent else []
54 commit_id = compute_commit_id(parent_ids, _SNAP_ID, message, _COMMITTED_AT.isoformat())
55 write_snapshot(repo, SnapshotRecord(
56 snapshot_id=_SNAP_ID,
57 manifest={},
58 created_at=_COMMITTED_AT,
59 ))
60 rec = CommitRecord(
61 commit_id=commit_id,
62 repo_id="test-repo",
63 branch=branch,
64 snapshot_id=_SNAP_ID,
65 message=message,
66 committed_at=_COMMITTED_AT,
67 author=author,
68 parent_commit_id=parent,
69 agent_id=agent_id,
70 model_id=model_id,
71 )
72 write_commit(repo, rec)
73 return commit_id
74
75
76 def _rc(repo: pathlib.Path, *args: str) -> InvokeResult:
77 from muse.cli.app import main as cli
78 return runner.invoke(
79 cli,
80 ["read-commit", *args],
81 env={"MUSE_REPO_ROOT": str(repo)},
82 )
83
84
85 # ---------------------------------------------------------------------------
86 # Unit — _ALL_FIELDS
87 # ---------------------------------------------------------------------------
88
89
90 class TestAllFields:
91 def test_all_fields_matches_commitdict_annotations(self) -> None:
92 """_ALL_FIELDS must be exactly the keys in CommitDict.__annotations__."""
93 from muse.cli.commands.read_commit import _ALL_FIELDS
94 from muse.core.store import CommitDict
95 assert _ALL_FIELDS == frozenset(CommitDict.__annotations__.keys())
96
97 def test_required_fields_present(self) -> None:
98 from muse.cli.commands.read_commit import _ALL_FIELDS
99 for field in ("commit_id", "branch", "message", "committed_at",
100 "agent_id", "model_id", "format_version", "reviewed_by"):
101 assert field in _ALL_FIELDS
102
103
104 # ---------------------------------------------------------------------------
105 # Integration — JSON format
106 # ---------------------------------------------------------------------------
107
108
109 class TestJsonFormat:
110 def test_full_schema_returned(self, tmp_path: pathlib.Path) -> None:
111 repo = _make_repo(tmp_path)
112 cid = _commit(repo, message="hello world")
113 result = _rc(repo, cid)
114 assert result.exit_code == 0
115 data = json.loads(result.output)
116 assert data["commit_id"] == cid
117 assert data["message"] == "hello world"
118 assert data["branch"] == "main"
119 assert "format_version" in data
120
121 def test_json_flag_shorthand(self, tmp_path: pathlib.Path) -> None:
122 repo = _make_repo(tmp_path)
123 cid = _commit(repo, message="shorthand test")
124 result = _rc(repo, "--json", cid)
125 assert result.exit_code == 0
126 assert "commit_id" in json.loads(result.output)
127
128 def test_agent_provenance_fields_present(self, tmp_path: pathlib.Path) -> None:
129 repo = _make_repo(tmp_path)
130 cid = _commit(repo, agent_id="my-agent", model_id="claude-4")
131 data = json.loads(_rc(repo, cid).output)
132 assert data["agent_id"] == "my-agent"
133 assert data["model_id"] == "claude-4"
134
135 def test_parent_commit_id_null_for_root(self, tmp_path: pathlib.Path) -> None:
136 repo = _make_repo(tmp_path)
137 cid = _commit(repo, message="root commit")
138 data = json.loads(_rc(repo, cid).output)
139 assert data["parent_commit_id"] is None
140
141 def test_parent_commit_id_set_for_child(self, tmp_path: pathlib.Path) -> None:
142 repo = _make_repo(tmp_path)
143 parent = _commit(repo, message="parent commit")
144 child = _commit(repo, message="child commit", parent=parent)
145 data = json.loads(_rc(repo, child).output)
146 assert data["parent_commit_id"] == parent
147
148 def test_committed_at_is_iso8601(self, tmp_path: pathlib.Path) -> None:
149 repo = _make_repo(tmp_path)
150 cid = _commit(repo, message="iso date test")
151 data = json.loads(_rc(repo, cid).output)
152 # Should parse without error
153 datetime.datetime.fromisoformat(data["committed_at"])
154
155 def test_snapshot_id_in_output(self, tmp_path: pathlib.Path) -> None:
156 repo = _make_repo(tmp_path)
157 cid = _commit(repo, message="snapshot test")
158 data = json.loads(_rc(repo, cid).output)
159 import re
160 assert re.fullmatch(r"sha256:[0-9a-f]{64}", data["snapshot_id"])
161
162
163 # ---------------------------------------------------------------------------
164 # Integration — text format
165 # ---------------------------------------------------------------------------
166
167
168 class TestTextFormat:
169 def test_text_format_contains_commit_prefix(self, tmp_path: pathlib.Path) -> None:
170 repo = _make_repo(tmp_path)
171 cid = _commit(repo, message="text test")
172 result = _rc(repo, "--format", "text", cid)
173 assert result.exit_code == 0
174 line = result.output.strip()
175 assert cid[:12] in line
176
177 def test_text_format_contains_branch(self, tmp_path: pathlib.Path) -> None:
178 repo = _make_repo(tmp_path)
179 cid = _commit(repo, branch="main", message="branch test")
180 result = _rc(repo, "--format", "text", cid)
181 assert "main" in result.output
182
183 def test_text_format_contains_message(self, tmp_path: pathlib.Path) -> None:
184 repo = _make_repo(tmp_path)
185 cid = _commit(repo, message="my commit message")
186 result = _rc(repo, "--format", "text", cid)
187 assert "my commit message" in result.output
188
189 def test_text_multiline_message_flattened(self, tmp_path: pathlib.Path) -> None:
190 repo = _make_repo(tmp_path)
191 cid = _commit(repo, message="line one\nline two")
192 result = _rc(repo, "--format", "text", cid)
193 # Newline replaced with space — output stays on one line
194 assert "\n" not in result.output.strip()
195 assert "line one" in result.output
196
197
198 # ---------------------------------------------------------------------------
199 # Integration — prefix resolution
200 # ---------------------------------------------------------------------------
201
202
203 class TestPrefixResolution:
204 def test_sha256_short_prefix_resolves(self, tmp_path: pathlib.Path) -> None:
205 """sha256:<8-hex> prefix form resolves to the full commit."""
206 repo = _make_repo(tmp_path)
207 cid = _commit(repo, message="prefix resolve test")
208 # cid is "sha256:<64-hex>"; take "sha256:" + first 8 hex chars = 15 chars
209 short_ref = cid[:15]
210 result = _rc(repo, short_ref)
211 assert result.exit_code == 0
212 assert json.loads(result.output)["commit_id"] == cid
213
214 def test_sha256_full_id_resolves(self, tmp_path: pathlib.Path) -> None:
215 """Full sha256:<64-hex> canonical form resolves directly."""
216 repo = _make_repo(tmp_path)
217 cid = _commit(repo, message="full id resolve test")
218 result = _rc(repo, cid)
219 assert result.exit_code == 0
220 assert json.loads(result.output)["commit_id"] == cid
221
222 def test_ambiguous_prefix_errors(self, tmp_path: pathlib.Path) -> None:
223 repo = _make_repo(tmp_path)
224 # "msg 121" and "msg 127" produce commit IDs sharing the "3f47" 4-char
225 # hex prefix when hashed with an empty manifest and 2026-01-01T00:00:00+00:00.
226 # Verified by precomputation; changing _SNAP_ID or _COMMITTED_AT requires
227 # updating these message strings.
228 cid1 = _commit(repo, message="msg 121")
229 cid2 = _commit(repo, message="msg 127")
230 result = _rc(repo, "sha256:3f47")
231 assert result.exit_code == ExitCode.USER_ERROR
232 data = json.loads(result.output)
233 assert "ambiguous" in data["error"]
234 assert set(data["candidates"]) == {cid1, cid2}
235
236 def test_missing_commit_errors(self, tmp_path: pathlib.Path) -> None:
237 repo = _make_repo(tmp_path)
238 # Valid canonical ID that doesn't exist in the store
239 result = _rc(repo, "sha256:" + "dead" + "beef" * 15)
240 assert result.exit_code == ExitCode.USER_ERROR
241 data = json.loads(result.output)
242 assert "not found" in data["error"]
243
244 def test_bare_hex_rejected(self, tmp_path: pathlib.Path) -> None:
245 """Bare hex without sha256: prefix is rejected with a clear error."""
246 repo = _make_repo(tmp_path)
247 result = _rc(repo, "a" * 64)
248 assert result.exit_code == ExitCode.USER_ERROR
249 data = json.loads(result.output)
250 assert "sha256:" in data["error"]
251
252 def test_bare_short_hex_rejected(self, tmp_path: pathlib.Path) -> None:
253 """Short bare hex prefix is rejected — sha256:<hex> form required."""
254 repo = _make_repo(tmp_path)
255 result = _rc(repo, "deadbeef")
256 assert result.exit_code == ExitCode.USER_ERROR
257 data = json.loads(result.output)
258 assert "sha256:" in data["error"]
259
260 def test_invalid_commit_id_errors(self, tmp_path: pathlib.Path) -> None:
261 repo = _make_repo(tmp_path)
262 result = _rc(repo, "ZZZZ" + "a" * 60)
263 assert result.exit_code == ExitCode.USER_ERROR
264
265
266 class TestSymbolicRefResolution:
267 def test_head_resolves(self, tmp_path: pathlib.Path) -> None:
268 """HEAD resolves to the tip of the current branch."""
269 repo = _make_repo(tmp_path)
270 cid = _commit(repo, branch="main", message="head test")
271 # Write branch ref so HEAD resolves
272 (repo / ".muse" / "refs" / "heads" / "main").write_text(cid, encoding="utf-8")
273 result = _rc(repo, "HEAD")
274 assert result.exit_code == 0
275 assert json.loads(result.output)["commit_id"] == cid
276
277 def test_branch_name_resolves(self, tmp_path: pathlib.Path) -> None:
278 """A branch name resolves to the tip commit of that branch."""
279 repo = _make_repo(tmp_path)
280 cid = _commit(repo, branch="dev", message="branch ref test")
281 (repo / ".muse" / "refs" / "heads" / "dev").write_text(cid, encoding="utf-8")
282 result = _rc(repo, "dev")
283 assert result.exit_code == 0
284 assert json.loads(result.output)["commit_id"] == cid
285
286 def test_tilde_notation_resolves(self, tmp_path: pathlib.Path) -> None:
287 """HEAD~1 resolves to the parent of the HEAD commit."""
288 repo = _make_repo(tmp_path)
289 parent_cid = _commit(repo, branch="main", message="parent")
290 child_cid = _commit(repo, branch="main", message="child", parent=parent_cid)
291 (repo / ".muse" / "refs" / "heads" / "main").write_text(child_cid, encoding="utf-8")
292 result = _rc(repo, "HEAD~1")
293 assert result.exit_code == 0
294 assert json.loads(result.output)["commit_id"] == parent_cid
295
296
297 # ---------------------------------------------------------------------------
298 # Integration — --fields filter
299 # ---------------------------------------------------------------------------
300
301
302 class TestFieldsFilter:
303 def test_single_field(self, tmp_path: pathlib.Path) -> None:
304 repo = _make_repo(tmp_path)
305 cid = _commit(repo, message="filtered")
306 result = _rc(repo, "--fields", "message", cid)
307 assert result.exit_code == 0
308 data = json.loads(result.output)
309 assert list(data.keys()) == ["message"]
310 assert data["message"] == "filtered"
311
312 def test_multiple_fields(self, tmp_path: pathlib.Path) -> None:
313 repo = _make_repo(tmp_path)
314 cid = _commit(repo, branch="dev", message="multi field test")
315 result = _rc(repo, "--fields", "commit_id,branch,message", cid)
316 data = json.loads(result.output)
317 assert set(data.keys()) == {"commit_id", "branch", "message"}
318 assert data["commit_id"] == cid
319 assert data["branch"] == "dev"
320
321 def test_agent_fields_filter(self, tmp_path: pathlib.Path) -> None:
322 """Agents extracting provenance fields should get exactly what they asked for."""
323 repo = _make_repo(tmp_path)
324 cid = _commit(repo, agent_id="audit-bot", model_id="claude-4")
325 result = _rc(repo, "--fields", "agent_id,model_id,format_version", cid)
326 data = json.loads(result.output)
327 assert set(data.keys()) == {"agent_id", "model_id", "format_version"}
328 assert data["agent_id"] == "audit-bot"
329 assert data["model_id"] == "claude-4"
330
331 def test_unknown_field_errors(self, tmp_path: pathlib.Path) -> None:
332 repo = _make_repo(tmp_path)
333 cid = _commit(repo, message="unknown field test")
334 result = _rc(repo, "--fields", "nonexistent_field", cid)
335 assert result.exit_code == ExitCode.USER_ERROR
336
337 def test_fields_with_text_format_errors(self, tmp_path: pathlib.Path) -> None:
338 repo = _make_repo(tmp_path)
339 cid = _commit(repo, message="fields text error test")
340 result = _rc(repo, "--fields", "commit_id", "--format", "text", cid)
341 assert result.exit_code == ExitCode.USER_ERROR
342
343 def test_fields_whitespace_trimmed(self, tmp_path: pathlib.Path) -> None:
344 repo = _make_repo(tmp_path)
345 cid = _commit(repo, message="whitespace trim test")
346 result = _rc(repo, "--fields", " commit_id , message ", cid)
347 assert result.exit_code == 0
348 data = json.loads(result.output)
349 assert "commit_id" in data
350 assert "message" in data
351
352
353 # ---------------------------------------------------------------------------
354 # Security
355 # ---------------------------------------------------------------------------
356
357
358 class TestSecurity:
359 def test_ansi_in_branch_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
360 repo = _make_repo(tmp_path)
361 _commit(repo, branch="main")
362 # Write an evil commit directly, bypassing the normal helper.
363 # The commit_id must be a real hash of the stored fields for read_commit
364 # to pass content-hash verification.
365 from muse.core.store import write_commit as _wc
366 evil_message = "test"
367 evil_cid = compute_commit_id([], _SNAP_ID, evil_message, _COMMITTED_AT.isoformat())
368 evil_rec = CommitRecord(
369 commit_id=evil_cid,
370 repo_id="test-repo",
371 branch="\x1b[31mevil\x1b[0m",
372 snapshot_id=_SNAP_ID,
373 message=evil_message,
374 committed_at=_COMMITTED_AT,
375 )
376 _wc(repo, evil_rec)
377 result = _rc(repo, "--format", "text", evil_cid)
378 assert result.exit_code == 0
379 assert "\x1b" not in result.output
380
381 def test_ansi_in_message_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
382 repo = _make_repo(tmp_path)
383 # snapshot_id does not need to be a valid hex string for read_commit to
384 # succeed — _verify_commit_id uses it as an opaque string in the hash.
385 evil_snap_id = "s" * 64
386 evil_message = "\x1b[31mmalicious\x1b[0m"
387 evil_committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
388 evil_cid = compute_commit_id([], evil_snap_id, evil_message, evil_committed_at.isoformat())
389 evil_rec = CommitRecord(
390 commit_id=evil_cid,
391 repo_id="test-repo",
392 branch="main",
393 snapshot_id=evil_snap_id,
394 message=evil_message,
395 committed_at=evil_committed_at,
396 )
397 write_commit(repo, evil_rec)
398 result = _rc(repo, "--format", "text", evil_cid)
399 assert result.exit_code == 0
400 assert "\x1b" not in result.output
401
402 def test_ansi_in_commit_id_rejected(self, tmp_path: pathlib.Path) -> None:
403 repo = _make_repo(tmp_path)
404 result = _rc(repo, "\x1b[31m" + "a" * 64)
405 assert result.exit_code == ExitCode.USER_ERROR
406
407 def test_no_traceback_on_bad_input(self, tmp_path: pathlib.Path) -> None:
408 repo = _make_repo(tmp_path)
409 result = _rc(repo, "not-valid")
410 assert "Traceback" not in result.output
411
412
413 # ---------------------------------------------------------------------------
414 # Stress
415 # ---------------------------------------------------------------------------
416
417
418 class TestStress:
419 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
420 repo = _make_repo(tmp_path)
421 cid = _commit(repo, message="stable")
422 for i in range(200):
423 result = _rc(repo, cid)
424 assert result.exit_code == 0, f"failed at iteration {i}"
425 assert json.loads(result.output)["message"] == "stable"
426
427 def test_fields_filter_200_iterations(self, tmp_path: pathlib.Path) -> None:
428 repo = _make_repo(tmp_path)
429 cid = _commit(repo, agent_id="bot")
430 for i in range(200):
431 result = _rc(repo, "--fields", "commit_id,agent_id", cid)
432 assert result.exit_code == 0, f"failed at iteration {i}"
433 data = json.loads(result.output)
434 assert data["agent_id"] == "bot"
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 146 days ago