gabriel / muse public
test_cmd_read_commit.py python
481 lines 19.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 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 from muse.core.paths import heads_dir, muse_dir
22
23 runner = CliRunner()
24
25 # Module-level constants so every test uses the same deterministic inputs.
26 _SNAP_ID: str = compute_snapshot_id({})
27 _COMMITTED_AT: datetime.datetime = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
28
29
30 # ---------------------------------------------------------------------------
31 # Helpers
32 # ---------------------------------------------------------------------------
33
34 def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
35 repo = tmp_path / "repo"
36 muse = muse_dir(repo)
37 for sub in ("objects", "commits", "snapshots", "refs/heads"):
38 (muse / sub).mkdir(parents=True)
39 (muse / "HEAD").write_text("ref: refs/heads/main")
40 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": "code"}))
41 return repo
42
43
44 def _commit(
45 repo: pathlib.Path,
46 *,
47 branch: str = "main",
48 message: str = "test commit",
49 author: str = "tester",
50 parent: str | None = None,
51 agent_id: str = "",
52 model_id: str = "",
53 ) -> str:
54 """Write a commit with a real content-addressed ID; return the commit_id."""
55 parent_ids: list[str] = [parent] if parent else []
56 commit_id = compute_commit_id(
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 repo_id="test-repo",
70 commit_id=commit_id,
71 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", "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["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 72" and "msg 329" produce commit IDs sharing the "72fe" 4-char
237 # hex prefix under the v2 formula (author="tester", empty manifest,
238 # 2026-01-01T00:00:00+00:00).
239 # Verified by precomputation; changing _SNAP_ID, _COMMITTED_AT, or
240 # author requires updating these message strings.
241 cid1 = _commit(repo, message="msg 72")
242 cid2 = _commit(repo, message="msg 329")
243 result = _rc(repo, "sha256:72fe")
244 assert result.exit_code == ExitCode.USER_ERROR
245 data = json.loads(result.stderr)
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(f"dead{'beef' * 15}"))
253 assert result.exit_code == ExitCode.USER_ERROR
254 data = json.loads(result.stderr)
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.stderr)
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.stderr)
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, f"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 (heads_dir(repo) / "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 (heads_dir(repo) / "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 (heads_dir(repo) / "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,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", "branch", "message"}
334 assert data["commit_id"] == cid
335 assert data["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 malicious 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 malicious_message = "test"
384 malicious_cid = compute_commit_id(
385 parent_ids=[],
386 snapshot_id=_SNAP_ID,
387 message=malicious_message,
388 committed_at_iso=_COMMITTED_AT.isoformat(),
389 )
390 malicious_rec = CommitRecord(
391 repo_id="test-repo",
392 commit_id=malicious_cid,
393 branch="\x1b[31mmalicious\x1b[0m",
394 snapshot_id=_SNAP_ID,
395 message=malicious_message,
396 committed_at=_COMMITTED_AT,
397 )
398 _wc(repo, malicious_rec)
399 result = _rc(repo, malicious_cid)
400 assert result.exit_code == 0
401 assert "\x1b" not in result.output
402
403 def test_ansi_in_message_stripped_in_text(self, tmp_path: pathlib.Path) -> None:
404 repo = _make_repo(tmp_path)
405 malicious_snap_id = fake_id("malicious-snap")
406 malicious_message = "\x1b[31mmalicious\x1b[0m"
407 malicious_committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
408 malicious_cid = compute_commit_id(
409 parent_ids=[],
410 snapshot_id=malicious_snap_id,
411 message=malicious_message,
412 committed_at_iso=malicious_committed_at.isoformat(),
413 )
414 malicious_rec = CommitRecord(
415 repo_id="test-repo",
416 commit_id=malicious_cid,
417 branch="main",
418 snapshot_id=malicious_snap_id,
419 message=malicious_message,
420 committed_at=malicious_committed_at,
421 )
422 write_commit(repo, malicious_rec)
423 result = _rc(repo, malicious_cid)
424 assert result.exit_code == 0
425 assert "\x1b" not in result.output
426
427 def test_ansi_in_commit_id_rejected(self, tmp_path: pathlib.Path) -> None:
428 repo = _make_repo(tmp_path)
429 result = _rc(repo, f"\x1b[31m{'a' * 64}")
430 assert result.exit_code == ExitCode.USER_ERROR
431
432 def test_no_traceback_on_bad_input(self, tmp_path: pathlib.Path) -> None:
433 repo = _make_repo(tmp_path)
434 result = _rc(repo, "not-valid")
435 assert "Traceback" not in result.output
436
437
438 # ---------------------------------------------------------------------------
439 # Stress
440 # ---------------------------------------------------------------------------
441
442
443 class TestStress:
444 def test_200_sequential_reads(self, tmp_path: pathlib.Path) -> None:
445 repo = _make_repo(tmp_path)
446 cid = _commit(repo, message="stable")
447 for i in range(200):
448 result = _rcj(repo, cid)
449 assert result.exit_code == 0, f"failed at iteration {i}"
450 assert json.loads(result.output)["message"] == "stable"
451
452 def test_fields_filter_200_iterations(self, tmp_path: pathlib.Path) -> None:
453 repo = _make_repo(tmp_path)
454 cid = _commit(repo, agent_id="bot")
455 for i in range(200):
456 result = _rcj(repo, "--fields", "commit_id,agent_id", cid)
457 assert result.exit_code == 0, f"failed at iteration {i}"
458 data = json.loads(result.output)
459 assert data["agent_id"] == "bot"
460
461
462 class TestRegisterFlags:
463 def _parse(self, *args: str) -> "argparse.Namespace":
464 import argparse
465 from muse.cli.commands.read_commit import register
466 p = argparse.ArgumentParser()
467 subs = p.add_subparsers()
468 register(subs)
469 return p.parse_args(["read-commit", fake_id("a"), *args])
470
471 def test_json_short_flag(self) -> None:
472 args = self._parse("-j")
473 assert args.json_out is True
474
475 def test_json_long_flag(self) -> None:
476 args = self._parse("--json")
477 assert args.json_out is True
478
479 def test_default_no_json(self) -> None:
480 args = self._parse()
481 assert args.json_out is False
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago