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