gabriel / muse public
test_cmd_remaining.py python
517 lines 19.2 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Comprehensive tests for the remaining 8 commands:
2 domain-info, show-ref, verify-object, symbolic-ref,
3 for-each-ref, name-rev, check-ref-format, verify-pack.
4
5 (check-ignore and check-attr are covered via attribute/ignore tests elsewhere.)
6
7 Coverage tiers
8 --------------
9 - Integration: core functionality, JSON/text formats, key flags
10 - Security: errors to stderr, no traceback on bad input
11 - Stress: 100+ object verify, 200 ref iterations
12 """
13 from __future__ import annotations
14
15 import datetime
16 import json
17 import pathlib
18
19 import msgpack
20
21 from muse.core.errors import ExitCode
22 from muse.core.object_store import write_object
23 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
24 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
25 from muse.core._types import Manifest, blob_id
26 from tests.cli_test_helper import CliRunner, InvokeResult
27
28 runner = CliRunner()
29
30
31 # ---------------------------------------------------------------------------
32 # Shared helpers
33 # ---------------------------------------------------------------------------
34
35 def _make_repo(tmp_path: pathlib.Path, domain: str = "code") -> pathlib.Path:
36 repo = tmp_path / "repo"
37 muse = repo / ".muse"
38 for sub in ("objects", "commits", "snapshots", "refs/heads"):
39 (muse / sub).mkdir(parents=True)
40 (muse / "HEAD").write_text("ref: refs/heads/main")
41 (muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo", "domain": domain}))
42 return repo
43
44
45 _TS = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
46
47
48 def _snap(repo: pathlib.Path, manifest: Manifest | None = None) -> str:
49 sid = compute_snapshot_id(manifest or {})
50 write_snapshot(repo, SnapshotRecord(
51 snapshot_id=sid,
52 manifest=manifest or {},
53 created_at=_TS,
54 ))
55 return sid
56
57
58 def _commit(
59 repo: pathlib.Path,
60 snap_id: str,
61 *,
62 branch: str = "main",
63 parent: str | None = None,
64 message: str = "test",
65 ) -> str:
66 parents = [parent] if parent else []
67 cid = compute_commit_id(
68 repo_id="test-repo",
69 parent_ids=parents,
70 snapshot_id=snap_id,
71 message=message,
72 committed_at_iso=_TS.isoformat(),
73 )
74 write_commit(repo, CommitRecord(
75 commit_id=cid,
76 repo_id="test-repo",
77 created_on_branch=branch,
78 snapshot_id=snap_id,
79 message=message,
80 committed_at=_TS,
81 parent_commit_id=parent,
82 ))
83 return cid
84
85
86 def _set_head(repo: pathlib.Path, branch: str, commit_id: str) -> None:
87 ref = repo / ".muse" / "refs" / "heads" / branch
88 ref.parent.mkdir(parents=True, exist_ok=True)
89 ref.write_text(commit_id)
90 (repo / ".muse" / "HEAD").write_text(f"ref: refs/heads/{branch}")
91
92
93 def _invoke(cmd: str, repo: pathlib.Path, *args: str, input: bytes | None = None) -> InvokeResult:
94 from muse.cli.app import main as cli
95 return runner.invoke(
96 cli,
97 [cmd, *args],
98 env={"MUSE_REPO_ROOT": str(repo)},
99 input=input,
100 )
101
102
103 def _fake_oid(n: int) -> str:
104 return format(n, "064x")
105
106
107 def _write_obj(repo: pathlib.Path, data: bytes) -> str:
108 """Write bytes to the object store; returns the prefixed SHA-256 object ID."""
109 oid = blob_id(data)
110 write_object(repo, oid, data)
111 return oid
112
113
114 # ===========================================================================
115 # domain-info
116 # ===========================================================================
117
118
119 class TestDomainInfo:
120 def test_all_domains_returns_list(self, tmp_path: pathlib.Path) -> None:
121 repo = _make_repo(tmp_path)
122 result = _invoke("domain-info", repo, "--all-domains", "--json")
123 assert result.exit_code == 0
124 data = json.loads(result.output)
125 assert "registered_domains" in data
126 assert isinstance(data["registered_domains"], list)
127
128 def test_all_domains_text_format(self, tmp_path: pathlib.Path) -> None:
129 repo = _make_repo(tmp_path)
130 result = _invoke("domain-info", repo, "--all-domains")
131 assert result.exit_code == 0
132 assert len(result.output.strip()) > 0
133
134 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
135 repo = _make_repo(tmp_path)
136 result = _invoke("domain-info", repo, "--all-domains", "--json")
137 assert result.exit_code == 0
138 assert "registered_domains" in json.loads(result.output)
139
140 def test_no_traceback_on_bad_domain(self, tmp_path: pathlib.Path) -> None:
141 repo = _make_repo(tmp_path, domain="nonexistent-domain")
142 result = _invoke("domain-info", repo)
143 assert "Traceback" not in result.output
144
145
146 # ===========================================================================
147 # show-ref
148 # ===========================================================================
149
150
151 class TestShowRef:
152 def test_shows_branches(self, tmp_path: pathlib.Path) -> None:
153 repo = _make_repo(tmp_path)
154 sid = _snap(repo)
155 cid = _commit(repo, sid)
156 _set_head(repo, "main", cid)
157 result = _invoke("show-ref", repo, "--json")
158 assert result.exit_code == 0
159 data = json.loads(result.output)
160 assert data["count"] == 1
161 assert any(r["ref"] == "refs/heads/main" for r in data["refs"])
162
163 def test_head_only_flag(self, tmp_path: pathlib.Path) -> None:
164 repo = _make_repo(tmp_path)
165 sid = _snap(repo)
166 cid = _commit(repo, sid)
167 _set_head(repo, "main", cid)
168 result = _invoke("show-ref", repo, "--head", "--json")
169 assert result.exit_code == 0
170 data = json.loads(result.output)
171 assert data["head"] is not None
172 assert data["head"]["commit_id"] == cid
173
174 def test_head_text_format(self, tmp_path: pathlib.Path) -> None:
175 repo = _make_repo(tmp_path)
176 sid = _snap(repo)
177 cid = _commit(repo, sid)
178 _set_head(repo, "main", cid)
179 result = _invoke("show-ref", repo)
180 assert result.exit_code == 0
181 assert cid in result.output
182 assert "main" in result.output
183
184 def test_verify_existing_ref(self, tmp_path: pathlib.Path) -> None:
185 repo = _make_repo(tmp_path)
186 sid = _snap(repo)
187 cid = _commit(repo, sid)
188 _set_head(repo, "main", cid)
189 result = _invoke("show-ref", repo, "--verify", "refs/heads/main")
190 assert result.exit_code == 0
191
192 def test_verify_nonexistent_ref_exits_1(self, tmp_path: pathlib.Path) -> None:
193 repo = _make_repo(tmp_path)
194 result = _invoke("show-ref", repo, "--verify", "ghost")
195 assert result.exit_code == ExitCode.USER_ERROR
196
197 def test_empty_repo_no_refs(self, tmp_path: pathlib.Path) -> None:
198 repo = _make_repo(tmp_path)
199 result = _invoke("show-ref", repo, "--json")
200 assert result.exit_code == 0
201 data = json.loads(result.output)
202 assert data["count"] == 0
203
204 def test_200_sequential_calls(self, tmp_path: pathlib.Path) -> None:
205 repo = _make_repo(tmp_path)
206 sid = _snap(repo)
207 cid = _commit(repo, sid)
208 _set_head(repo, "main", cid)
209 for i in range(200):
210 result = _invoke("show-ref", repo)
211 assert result.exit_code == 0, f"failed at {i}"
212
213
214 # ===========================================================================
215 # verify-object
216 # ===========================================================================
217
218
219 class TestVerifyObject:
220 def test_valid_object_ok(self, tmp_path: pathlib.Path) -> None:
221 repo = _make_repo(tmp_path)
222 oid = _write_obj(repo, b"hello world")
223 result = _invoke("verify-object", repo, "--json", oid)
224 assert result.exit_code == 0
225 data = json.loads(result.output)
226 assert data["all_ok"] is True
227 assert data["results"][0]["ok"] is True
228
229 def test_missing_object_fails(self, tmp_path: pathlib.Path) -> None:
230 repo = _make_repo(tmp_path)
231 result = _invoke("verify-object", repo, "--json", "dead" + "beef" * 15)
232 assert result.exit_code == ExitCode.USER_ERROR
233 data = json.loads(result.output)
234 assert data["all_ok"] is False
235 assert not data["results"][0]["ok"]
236
237 def test_invalid_object_id_fails(self, tmp_path: pathlib.Path) -> None:
238 repo = _make_repo(tmp_path)
239 result = _invoke("verify-object", repo, "--json", "not-hex")
240 assert result.exit_code == ExitCode.USER_ERROR
241 data = json.loads(result.output)
242 assert data["all_ok"] is False
243
244 def test_text_format(self, tmp_path: pathlib.Path) -> None:
245 repo = _make_repo(tmp_path)
246 oid = _write_obj(repo, b"text test")
247 result = _invoke("verify-object", repo, oid)
248 assert result.exit_code == 0
249 assert "OK" in result.output
250
251 def test_quiet_mode_exit_code(self, tmp_path: pathlib.Path) -> None:
252 repo = _make_repo(tmp_path)
253 oid = _write_obj(repo, b"quiet")
254 result = _invoke("verify-object", repo, "--quiet", oid)
255 assert result.exit_code == 0
256
257 def test_multiple_objects(self, tmp_path: pathlib.Path) -> None:
258 repo = _make_repo(tmp_path)
259 oid1 = _write_obj(repo, b"obj1")
260 oid2 = _write_obj(repo, b"obj2")
261 result = _invoke("verify-object", repo, "--json", oid1, oid2)
262 assert result.exit_code == 0
263 data = json.loads(result.output)
264 assert data["checked"] == 2
265 assert data["failed"] == 0
266
267 def test_100_objects(self, tmp_path: pathlib.Path) -> None:
268 repo = _make_repo(tmp_path)
269 oids = [_write_obj(repo, f"obj-{i}".encode()) for i in range(100)]
270 result = _invoke("verify-object", repo, "--json", *oids)
271 assert result.exit_code == 0
272 data = json.loads(result.output)
273 assert data["checked"] == 100
274 assert data["failed"] == 0
275
276 def test_no_traceback_on_bad_id(self, tmp_path: pathlib.Path) -> None:
277 repo = _make_repo(tmp_path)
278 result = _invoke("verify-object", repo, "bad")
279 assert "Traceback" not in result.output
280
281
282 # ===========================================================================
283 # symbolic-ref
284 # ===========================================================================
285
286
287 class TestSymbolicRef:
288 def test_reads_head_branch(self, tmp_path: pathlib.Path) -> None:
289 repo = _make_repo(tmp_path)
290 sid = _snap(repo)
291 cid = _commit(repo, sid)
292 _set_head(repo, "main", cid)
293 result = _invoke("symbolic-ref", repo, "--json", "HEAD")
294 assert result.exit_code == 0
295 data = json.loads(result.output)
296 assert data["branch"] == "main"
297 assert "refs/heads/main" in data["symbolic_target"]
298
299 def test_short_flag(self, tmp_path: pathlib.Path) -> None:
300 repo = _make_repo(tmp_path)
301 sid = _snap(repo)
302 cid = _commit(repo, sid)
303 _set_head(repo, "main", cid)
304 result = _invoke("symbolic-ref", repo, "--short", "HEAD")
305 assert result.exit_code == 0
306 assert result.output.strip() == "main"
307
308 def test_set_changes_head(self, tmp_path: pathlib.Path) -> None:
309 repo = _make_repo(tmp_path)
310 sid = _snap(repo)
311 cid = _commit(repo, sid, branch="dev")
312 _set_head(repo, "dev", cid)
313 result = _invoke("symbolic-ref", repo, "--set", "dev", "HEAD")
314 assert result.exit_code == 0
315 assert (repo / ".muse" / "HEAD").read_text().strip() == "ref: refs/heads/dev"
316
317 def test_unsupported_ref_errors(self, tmp_path: pathlib.Path) -> None:
318 repo = _make_repo(tmp_path)
319 result = _invoke("symbolic-ref", repo, "refs/heads/main")
320 assert result.exit_code == ExitCode.USER_ERROR
321
322 def test_no_traceback_on_bad_ref(self, tmp_path: pathlib.Path) -> None:
323 repo = _make_repo(tmp_path)
324 result = _invoke("symbolic-ref", repo, "bad-ref")
325 assert "Traceback" not in result.output
326
327
328 # ===========================================================================
329 # for-each-ref
330 # ===========================================================================
331
332
333 class TestForEachRef:
334 def test_lists_all_refs(self, tmp_path: pathlib.Path) -> None:
335 repo = _make_repo(tmp_path)
336 sid = _snap(repo)
337 cid1 = _commit(repo, sid, branch="main", message="init-main")
338 cid2 = _commit(repo, sid, branch="dev", message="init-dev")
339 _set_head(repo, "main", cid1)
340 _set_head(repo, "dev", cid2)
341 result = _invoke("for-each-ref", repo, "--json")
342 assert result.exit_code == 0
343 data = json.loads(result.output)
344 branch_names = {r["branch"] for r in data["refs"]}
345 assert "main" in branch_names
346 assert "dev" in branch_names
347
348 def test_text_format(self, tmp_path: pathlib.Path) -> None:
349 repo = _make_repo(tmp_path)
350 sid = _snap(repo)
351 cid = _commit(repo, sid)
352 _set_head(repo, "main", cid)
353 result = _invoke("for-each-ref", repo)
354 assert result.exit_code == 0
355 assert "main" in result.output
356
357 def test_count_limit(self, tmp_path: pathlib.Path) -> None:
358 repo = _make_repo(tmp_path)
359 sid = _snap(repo)
360 for i in range(5):
361 cid = _commit(repo, sid, branch=f"branch-{i}", message=f"branch-{i}")
362 _set_head(repo, f"branch-{i}", cid)
363 result = _invoke("for-each-ref", repo, "--count", "3", "--json")
364 assert result.exit_code == 0
365 data = json.loads(result.output)
366 assert len(data["refs"]) == 3
367
368 def test_empty_repo(self, tmp_path: pathlib.Path) -> None:
369 repo = _make_repo(tmp_path)
370 result = _invoke("for-each-ref", repo, "--json")
371 assert result.exit_code == 0
372 data = json.loads(result.output)
373 assert data["refs"] == []
374
375
376 # ===========================================================================
377 # name-rev
378 # ===========================================================================
379
380
381 class TestNameRev:
382 def test_tip_commit_names_to_branch(self, tmp_path: pathlib.Path) -> None:
383 repo = _make_repo(tmp_path)
384 sid = _snap(repo)
385 cid = _commit(repo, sid, branch="main")
386 _set_head(repo, "main", cid)
387 result = _invoke("name-rev", repo, "--json", cid)
388 assert result.exit_code == 0
389 data = json.loads(result.output)
390 assert len(data["results"]) == 1
391 assert data["results"][0]["commit_id"] == cid
392 assert "main" in data["results"][0]["name"]
393
394 def test_parent_commit_named_with_tilde(self, tmp_path: pathlib.Path) -> None:
395 repo = _make_repo(tmp_path)
396 sid = _snap(repo)
397 c1 = _commit(repo, sid, branch="main", message="c1-msg")
398 c2 = _commit(repo, sid, branch="main", parent=c1, message="c2-msg")
399 _set_head(repo, "main", c2)
400 result = _invoke("name-rev", repo, "--json", c1)
401 assert result.exit_code == 0
402 data = json.loads(result.output)
403 name = data["results"][0]["name"]
404 assert "~" in name or "main" in name
405
406 def test_no_commit_ids_errors(self, tmp_path: pathlib.Path) -> None:
407 repo = _make_repo(tmp_path)
408 result = _invoke("name-rev", repo)
409 assert result.exit_code == ExitCode.USER_ERROR
410
411 def test_no_traceback_on_empty_input(self, tmp_path: pathlib.Path) -> None:
412 repo = _make_repo(tmp_path)
413 result = _invoke("name-rev", repo)
414 assert "Traceback" not in result.output
415
416
417 # ===========================================================================
418 # check-ref-format
419 # ===========================================================================
420
421
422 class TestCheckRefFormat:
423 def test_valid_branch_name(self, tmp_path: pathlib.Path) -> None:
424 repo = _make_repo(tmp_path)
425 result = _invoke("check-ref-format", repo, "--json", "main")
426 assert result.exit_code == 0
427 data = json.loads(result.output)
428 assert data["all_valid"] is True
429
430 def test_valid_feature_branch(self, tmp_path: pathlib.Path) -> None:
431 repo = _make_repo(tmp_path)
432 result = _invoke("check-ref-format", repo, "--json", "feat/add-melody")
433 assert result.exit_code == 0
434 data = json.loads(result.output)
435 assert data["all_valid"] is True
436
437 def test_invalid_null_byte_rejected(self, tmp_path: pathlib.Path) -> None:
438 repo = _make_repo(tmp_path)
439 result = _invoke("check-ref-format", repo, "--json", "bad\x00branch")
440 assert result.exit_code == ExitCode.USER_ERROR
441 data = json.loads(result.output)
442 assert data["all_valid"] is False
443
444 def test_multiple_names_mixed_validity(self, tmp_path: pathlib.Path) -> None:
445 repo = _make_repo(tmp_path)
446 result = _invoke("check-ref-format", repo, "--json", "main", "bad\x00branch")
447 assert result.exit_code == ExitCode.USER_ERROR
448 data = json.loads(result.output)
449 assert data["all_valid"] is False
450 valid = {r["name"]: r["valid"] for r in data["results"]}
451 assert valid["main"] is True
452 assert valid["bad\x00branch"] is False
453
454 def test_text_format(self, tmp_path: pathlib.Path) -> None:
455 repo = _make_repo(tmp_path)
456 result = _invoke("check-ref-format", repo, "main")
457 assert result.exit_code == 0
458 assert "ok" in result.output.lower()
459
460 def test_quiet_mode(self, tmp_path: pathlib.Path) -> None:
461 repo = _make_repo(tmp_path)
462 result = _invoke("check-ref-format", repo, "--quiet", "main")
463 assert result.exit_code == 0
464 assert result.output.strip() == ""
465
466 def test_no_args_errors(self, tmp_path: pathlib.Path) -> None:
467 repo = _make_repo(tmp_path)
468 result = _invoke("check-ref-format", repo)
469 assert result.exit_code == ExitCode.USER_ERROR
470
471 def test_no_traceback_on_bad_name(self, tmp_path: pathlib.Path) -> None:
472 repo = _make_repo(tmp_path)
473 result = _invoke("check-ref-format", repo, "bad\x00branch")
474 assert "Traceback" not in result.output
475
476
477 # ===========================================================================
478 # verify-pack
479 # ===========================================================================
480
481
482 class TestVerifyPack:
483 def _make_pack_bytes(self, repo: pathlib.Path, cid: str, sid: str) -> bytes:
484 from muse.cli.app import main as cli
485 result = runner.invoke(
486 cli,
487 ["pack-objects", cid],
488 env={"MUSE_REPO_ROOT": str(repo)},
489 )
490 assert result.exit_code == 0
491 return result.stdout_bytes
492
493 def test_valid_pack_from_stdin(self, tmp_path: pathlib.Path) -> None:
494 repo = _make_repo(tmp_path)
495 sid = _snap(repo)
496 cid = _commit(repo, sid, message="test-pack")
497 pack_bytes = self._make_pack_bytes(repo, cid, sid)
498 result = _invoke("verify-pack", repo, "--json", input=pack_bytes)
499 assert result.exit_code == 0
500 data = json.loads(result.output)
501 assert data["all_ok"] is True
502 assert data["commits_checked"] >= 1
503
504 def test_empty_stdin_errors(self, tmp_path: pathlib.Path) -> None:
505 repo = _make_repo(tmp_path)
506 result = _invoke("verify-pack", repo, input=b"")
507 assert result.exit_code == ExitCode.USER_ERROR
508
509 def test_corrupted_msgpack_errors(self, tmp_path: pathlib.Path) -> None:
510 repo = _make_repo(tmp_path)
511 result = _invoke("verify-pack", repo, input=b"\xff\xfe corrupted")
512 assert result.exit_code == ExitCode.USER_ERROR
513
514 def test_no_traceback_on_bad_input(self, tmp_path: pathlib.Path) -> None:
515 repo = _make_repo(tmp_path)
516 result = _invoke("verify-pack", repo, input=b"not msgpack at all")
517 assert "Traceback" not in result.output
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 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 140 days ago