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