gabriel / muse public
test_release_supercharge.py python
498 lines 20.0 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 124 days ago
1 """Supercharge tests for ``muse release``.
2
3 Coverage tiers
4 --------------
5 - Unit: _short_id helper — sha256:-prefixed input
6 - Integration: duration_ms + exit_code on every JSON path
7 (add, list, read, push --dry-run, delete --dry-run, suggest)
8 - Data: short commit IDs in text output (not bare truncated hex)
9 - Data: list JSON wraps in {total, releases} envelope
10 - Security: no tracebacks; ANSI stripped in text
11 - Performance: suggest on 50 commits under 500ms
12 """
13 from __future__ import annotations
14
15 import datetime
16 import json
17 import pathlib
18 import time
19
20 from muse.core.errors import ExitCode
21 from tests.cli_test_helper import CliRunner, InvokeResult
22 from muse.core.types import content_hash, long_id, short_id as _short_id
23 from muse.core.paths import ref_path, muse_dir
24
25 runner = CliRunner()
26
27 _SHA256_SHORT_19 = __import__("re").compile(r"^sha256:[0-9a-f]{12}$")
28
29
30 # ---------------------------------------------------------------------------
31 # Repo + commit helpers (mirrors test_release.py)
32 # ---------------------------------------------------------------------------
33
34
35 def _make_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
36 dot_muse = muse_dir(tmp_path)
37 dot_muse.mkdir()
38 repo_id = content_hash({"path": str(tmp_path), "domain": "code", "created_at": "2025-01-01T00:00:00+00:00"})
39 (dot_muse / "repo.json").write_text(
40 json.dumps({"repo_id": repo_id, "domain": "code",
41 "default_branch": "main",
42 "created_at": "2025-01-01T00:00:00+00:00"}),
43 encoding="utf-8",
44 )
45 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8")
46 (dot_muse / "refs" / "heads").mkdir(parents=True)
47 (dot_muse / "snapshots").mkdir()
48 (dot_muse / "commits").mkdir()
49 (dot_muse / "objects").mkdir()
50 return tmp_path, repo_id
51
52
53 def _commit(
54 root: pathlib.Path,
55 repo_id: str,
56 *,
57 message: str = "feat: add something",
58 sem_ver_bump: str = "minor",
59 agent_id: str = "claude-code",
60 model_id: str = "claude-sonnet-4-6",
61 ) -> str:
62 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
63 from muse.core.store import CommitRecord, SnapshotRecord, write_commit, write_snapshot
64 from muse.domain import SemVerBump
65
66 branch = "main"
67 ref_file = ref_path(root, branch)
68 raw_parent = ref_file.read_text().strip() if ref_file.exists() else ""
69 parent_id: str | None = raw_parent if raw_parent else None
70 snap_id = compute_snapshot_id({})
71 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest={}))
72 now = datetime.datetime.now(datetime.timezone.utc)
73 parent_ids = [parent_id] if parent_id else []
74 commit_id = compute_commit_id( parent_ids=parent_ids,
75 snapshot_id=snap_id,
76 message=message,
77 committed_at_iso=now.isoformat(),
78 )
79 bump: SemVerBump = sem_ver_bump # type: ignore[assignment]
80 write_commit(root, CommitRecord(
81 commit_id=commit_id, repo_id=repo_id, branch=branch,
82 snapshot_id=snap_id, message=message, committed_at=now,
83 parent_commit_id=parent_id, sem_ver_bump=bump,
84 breaking_changes=[], agent_id=agent_id, model_id=model_id,
85 ))
86 ref_file.write_text(commit_id, encoding="utf-8")
87 return commit_id
88
89
90 def _invoke(root: pathlib.Path, *args: str) -> InvokeResult:
91 from muse.cli.app import main as cli
92 return runner.invoke(cli, ["release", *args],
93 env={"MUSE_REPO_ROOT": str(root)})
94
95
96 def _add(root: pathlib.Path, tag: str = "v0.1.0", **kwargs: str) -> InvokeResult:
97 extra = []
98 for k, v in kwargs.items():
99 extra.extend([f"--{k}", v])
100 return _invoke(root, "add", tag, "--json", *extra)
101
102
103 # ---------------------------------------------------------------------------
104 # Unit — _short_id
105 # ---------------------------------------------------------------------------
106
107
108 class TestShortId:
109 """_short_id normalises sha256:-prefixed commit IDs to sha256:<12-hex>."""
110
111 def test_sha256_prefixed_returns_sha256_short(self) -> None:
112
113 cid = long_id("a" * 64)
114 assert _short_id(cid) == long_id("a" * 12)
115
116 def test_result_is_19_chars(self) -> None:
117
118 assert len(_short_id(long_id("b" * 64))) == 19
119
120 def test_bare_hex_also_handled(self) -> None:
121 # bare hex in → first 12 chars out, no prefix added
122 result = _short_id("c" * 64)
123 assert result == "c" * 12
124
125 def test_matches_short_regex(self) -> None:
126
127 assert _SHA256_SHORT_19.match(_short_id(long_id("d" * 64)))
128
129 def test_idempotent_on_short_input(self) -> None:
130 """Already-short sha256:<12-hex> passes through unchanged."""
131
132 short = long_id("e" * 12)
133 assert _short_id(short) == short
134
135
136 # ---------------------------------------------------------------------------
137 # Integration — duration_ms and exit_code on every JSON path
138 # ---------------------------------------------------------------------------
139
140
141 class TestDurationAndExitCode:
142 """Every subcommand's JSON output must carry duration_ms and exit_code."""
143
144 def test_add_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
145 root, repo_id = _make_repo(tmp_path)
146 _commit(root, repo_id)
147 data = json.loads(_add(root).output)
148 assert "duration_ms" in data
149
150 def test_add_json_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
151 root, repo_id = _make_repo(tmp_path)
152 _commit(root, repo_id)
153 data = json.loads(_add(root).output)
154 assert data["exit_code"] == 0
155
156 def test_add_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
157 root, repo_id = _make_repo(tmp_path)
158 _commit(root, repo_id)
159 assert isinstance(json.loads(_add(root).output)["duration_ms"], float)
160
161 def test_list_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
162 root, repo_id = _make_repo(tmp_path)
163 _commit(root, repo_id)
164 _add(root)
165 data = json.loads(_invoke(root, "list", "--json").output)
166 assert "duration_ms" in data
167
168 def test_list_json_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
169 root, repo_id = _make_repo(tmp_path)
170 data = json.loads(_invoke(root, "list", "--json").output)
171 assert data["exit_code"] == 0
172
173 def test_list_empty_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
174 """Even with zero releases, metadata fields must be present."""
175 root, repo_id = _make_repo(tmp_path)
176 data = json.loads(_invoke(root, "list", "--json").output)
177 assert "duration_ms" in data
178
179 def test_read_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
180 root, repo_id = _make_repo(tmp_path)
181 _commit(root, repo_id)
182 _add(root)
183 data = json.loads(_invoke(root, "read", "v0.1.0", "--json").output)
184 assert "duration_ms" in data
185
186 def test_read_json_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
187 root, repo_id = _make_repo(tmp_path)
188 _commit(root, repo_id)
189 _add(root)
190 data = json.loads(_invoke(root, "read", "v0.1.0", "--json").output)
191 assert data["exit_code"] == 0
192
193 def test_push_dry_run_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
194 root, repo_id = _make_repo(tmp_path)
195 _commit(root, repo_id)
196 _add(root)
197 data = json.loads(
198 _invoke(root, "push", "v0.1.0", "--dry-run", "--json").output
199 )
200 assert "duration_ms" in data
201
202 def test_push_dry_run_json_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
203 root, repo_id = _make_repo(tmp_path)
204 _commit(root, repo_id)
205 _add(root)
206 data = json.loads(
207 _invoke(root, "push", "v0.1.0", "--dry-run", "--json").output
208 )
209 assert data["exit_code"] == 0
210
211 def test_delete_dry_run_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
212 root, repo_id = _make_repo(tmp_path)
213 _commit(root, repo_id)
214 _add(root)
215 data = json.loads(
216 _invoke(root, "delete", "v0.1.0", "--dry-run", "--json").output
217 )
218 assert "duration_ms" in data
219
220 def test_delete_dry_run_json_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
221 root, repo_id = _make_repo(tmp_path)
222 _commit(root, repo_id)
223 _add(root)
224 data = json.loads(
225 _invoke(root, "delete", "v0.1.0", "--dry-run", "--json").output
226 )
227 assert data["exit_code"] == 0
228
229 def test_suggest_json_has_duration_ms(self, tmp_path: pathlib.Path) -> None:
230 root, repo_id = _make_repo(tmp_path)
231 _commit(root, repo_id, sem_ver_bump="minor")
232 data = json.loads(_invoke(root, "suggest", "--json").output)
233 assert "duration_ms" in data
234
235 def test_suggest_json_exit_code_zero(self, tmp_path: pathlib.Path) -> None:
236 root, repo_id = _make_repo(tmp_path)
237 _commit(root, repo_id, sem_ver_bump="minor")
238 data = json.loads(_invoke(root, "suggest", "--json").output)
239 assert data["exit_code"] == 0
240
241 def test_duration_ms_non_negative(self, tmp_path: pathlib.Path) -> None:
242 root, repo_id = _make_repo(tmp_path)
243 _commit(root, repo_id)
244 _add(root)
245 assert json.loads(_add(root, "v0.2.0").output)["duration_ms"] >= 0.0
246
247 def test_duration_ms_3dp_precision(self, tmp_path: pathlib.Path) -> None:
248 root, repo_id = _make_repo(tmp_path)
249 _commit(root, repo_id)
250 ms = json.loads(_add(root).output)["duration_ms"]
251 assert round(ms, 3) == ms
252
253
254 # ---------------------------------------------------------------------------
255 # Data integrity — text format short IDs
256 # ---------------------------------------------------------------------------
257
258
259 class TestTextFormatShortId:
260 """Text output must show sha256:<12-hex> commit IDs, not bare [:8] slices."""
261
262 def _short_tokens(self, text: str) -> list[str]:
263 return [tok for tok in text.split() if _SHA256_SHORT_19.match(tok)]
264
265 def test_read_text_shows_sha256_short_commit(self, tmp_path: pathlib.Path) -> None:
266 root, repo_id = _make_repo(tmp_path)
267 cid = _commit(root, repo_id)
268 _invoke(root, "add", "v0.1.0")
269 result = _invoke(root, "read", "v0.1.0")
270 assert result.exit_code == 0
271 tokens = self._short_tokens(result.output)
272 assert tokens, f"no sha256:<12-hex> token in read text:\n{result.output}"
273
274 def test_list_text_shows_sha256_short_commit(self, tmp_path: pathlib.Path) -> None:
275 root, repo_id = _make_repo(tmp_path)
276 _commit(root, repo_id)
277 _invoke(root, "add", "v0.1.0")
278 result = _invoke(root, "list")
279 assert result.exit_code == 0
280 tokens = self._short_tokens(result.output)
281 assert tokens, f"no sha256:<12-hex> token in list text:\n{result.output}"
282
283 def test_suggest_text_shows_sha256_short_in_driver(self, tmp_path: pathlib.Path) -> None:
284 root, repo_id = _make_repo(tmp_path)
285 _commit(root, repo_id, sem_ver_bump="minor")
286 result = _invoke(root, "suggest")
287 assert result.exit_code == 0
288 tokens = self._short_tokens(result.output)
289 assert tokens, f"no sha256:<12-hex> token in suggest text:\n{result.output}"
290
291 def test_read_text_commit_not_bare_hex_prefix_only(self, tmp_path: pathlib.Path) -> None:
292 """Regression: commit_id[:8] on sha256:-prefixed ID yields 'sha256:a' — wrong."""
293 root, repo_id = _make_repo(tmp_path)
294 _commit(root, repo_id)
295 _invoke(root, "add", "v0.1.0")
296 result = _invoke(root, "read", "v0.1.0")
297 # "sha256:a" with no further hex after colon would indicate the bug
298 assert "sha256:a\n" not in result.output
299 assert "sha256:b\n" not in result.output
300
301 def test_changelog_entries_show_sha256_short(self, tmp_path: pathlib.Path) -> None:
302 root, repo_id = _make_repo(tmp_path)
303 _commit(root, repo_id, message="feat: first", sem_ver_bump="minor")
304 _commit(root, repo_id, message="feat: second", sem_ver_bump="minor")
305 _invoke(root, "add", "v0.1.0")
306 result = _invoke(root, "read", "v0.1.0")
307 tokens = self._short_tokens(result.output)
308 # changelog has 2 entries, each with a short commit ID
309 assert len(tokens) >= 2, f"expected ≥2 sha256:<12-hex> tokens:\n{result.output}"
310
311
312 # ---------------------------------------------------------------------------
313 # Data integrity — list JSON envelope
314 # ---------------------------------------------------------------------------
315
316
317 class TestListJsonEnvelope:
318 """list --json emits {total, releases, duration_ms, exit_code} — not a bare array."""
319
320 def test_list_json_top_level_is_object(self, tmp_path: pathlib.Path) -> None:
321 root, repo_id = _make_repo(tmp_path)
322 data = json.loads(_invoke(root, "list", "--json").output)
323 assert isinstance(data, dict)
324
325 def test_list_json_has_total_key(self, tmp_path: pathlib.Path) -> None:
326 root, repo_id = _make_repo(tmp_path)
327 data = json.loads(_invoke(root, "list", "--json").output)
328 assert "total" in data
329
330 def test_list_json_has_releases_key(self, tmp_path: pathlib.Path) -> None:
331 root, repo_id = _make_repo(tmp_path)
332 data = json.loads(_invoke(root, "list", "--json").output)
333 assert "releases" in data
334
335 def test_list_json_releases_is_array(self, tmp_path: pathlib.Path) -> None:
336 root, repo_id = _make_repo(tmp_path)
337 data = json.loads(_invoke(root, "list", "--json").output)
338 assert isinstance(data["releases"], list)
339
340 def test_list_json_total_matches_releases_length(self, tmp_path: pathlib.Path) -> None:
341 root, repo_id = _make_repo(tmp_path)
342 _commit(root, repo_id)
343 _add(root, "v0.1.0")
344 _commit(root, repo_id)
345 _add(root, "v0.2.0")
346 data = json.loads(_invoke(root, "list", "--json").output)
347 assert data["total"] == len(data["releases"]) == 2
348
349 def test_list_json_empty_total_is_zero(self, tmp_path: pathlib.Path) -> None:
350 root, repo_id = _make_repo(tmp_path)
351 data = json.loads(_invoke(root, "list", "--json").output)
352 assert data["total"] == 0
353 assert data["releases"] == []
354
355 def test_bare_release_json_flag_uses_envelope(self, tmp_path: pathlib.Path) -> None:
356 """`muse release --json` (no subcommand) also uses the envelope."""
357 root, repo_id = _make_repo(tmp_path)
358 data = json.loads(_invoke(root, "--json").output)
359 assert "releases" in data
360 assert "total" in data
361
362
363 # ---------------------------------------------------------------------------
364 # Security
365 # ---------------------------------------------------------------------------
366
367
368 class TestSecuritySupercharge:
369 def test_no_traceback_on_bad_semver(self, tmp_path: pathlib.Path) -> None:
370 root, _ = _make_repo(tmp_path)
371 result = _invoke(root, "add", "not-semver", "--json")
372 assert result.exit_code == ExitCode.USER_ERROR
373 assert "Traceback" not in result.output
374
375 def test_no_traceback_on_missing_read(self, tmp_path: pathlib.Path) -> None:
376 root, _ = _make_repo(tmp_path)
377 result = _invoke(root, "read", "v9.9.9", "--json")
378 assert result.exit_code == ExitCode.NOT_FOUND
379 assert "Traceback" not in result.output
380
381 def test_no_traceback_on_suggest_no_commits(self, tmp_path: pathlib.Path) -> None:
382 root, _ = _make_repo(tmp_path)
383 result = _invoke(root, "suggest", "--json")
384 # Either no commits (exit 1) or 0 unreleased — must not traceback
385 assert "Traceback" not in result.output
386
387 def test_add_json_error_on_duplicate(self, tmp_path: pathlib.Path) -> None:
388 root, repo_id = _make_repo(tmp_path)
389 _commit(root, repo_id)
390 _add(root)
391 result = _invoke(root, "add", "v0.1.0", "--json")
392 assert result.exit_code == ExitCode.USER_ERROR
393 # JSON error is on stdout (first line); ❌ message goes to stderr
394 first_line = result.output.splitlines()[0]
395 data = json.loads(first_line)
396 assert data["error"] == "already_exists"
397
398
399 # ---------------------------------------------------------------------------
400 # Performance
401 # ---------------------------------------------------------------------------
402
403
404 class TestPerformanceSupercharge:
405 def test_list_50_releases_under_500ms(self, tmp_path: pathlib.Path) -> None:
406 root, repo_id = _make_repo(tmp_path)
407 for i in range(50):
408 _commit(root, repo_id, message=f"feat: thing {i}")
409 _invoke(root, "add", f"v0.{i}.0")
410 t0 = time.monotonic()
411 result = _invoke(root, "list", "--json")
412 duration_ms = (time.monotonic() - t0) * 1000
413 assert result.exit_code == 0
414 assert duration_ms < 500
415
416 def test_suggest_50_commits_under_500ms(self, tmp_path: pathlib.Path) -> None:
417 root, repo_id = _make_repo(tmp_path)
418 for i in range(50):
419 _commit(root, repo_id, sem_ver_bump="patch")
420 t0 = time.monotonic()
421 result = _invoke(root, "suggest", "--json")
422 duration_ms = (time.monotonic() - t0) * 1000
423 assert result.exit_code == 0
424 assert duration_ms < 500
425
426 def test_duration_ms_plausible(self, tmp_path: pathlib.Path) -> None:
427 root, repo_id = _make_repo(tmp_path)
428 _commit(root, repo_id)
429 ms = json.loads(_add(root).output)["duration_ms"]
430 assert 0.0 <= ms < 500
431
432
433 # ---------------------------------------------------------------------------
434 # Content-addressed release_id
435 # ---------------------------------------------------------------------------
436
437
438 class TestReleaseIdContentAddressed:
439 """release_id must be sha256: of genesis content, not a UUID."""
440
441 def test_release_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None:
442 root, repo_id = _make_repo(tmp_path)
443 _commit(root, repo_id)
444 data = json.loads(_add(root, "v1.0.0").output)
445 assert data["release_id"].startswith("sha256:"), f"Expected sha256: prefix, got {data['release_id']!r}"
446 assert len(data["release_id"]) == 71
447
448 def test_release_id_is_sha256_not_uuid4(self, tmp_path: pathlib.Path) -> None:
449 import re
450 root, repo_id = _make_repo(tmp_path)
451 _commit(root, repo_id)
452 data = json.loads(_add(root, "v1.0.0").output)
453 uuid4_re = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$")
454 assert not uuid4_re.match(data["release_id"])
455
456 def test_release_id_is_deterministic(self, tmp_path: pathlib.Path) -> None:
457 """Same repo + tag + commit → same release_id."""
458 from muse.core.store import compute_release_id
459 release_id = compute_release_id(repo_id="test-repo", tag="v1.0.0", commit_id=long_id("a" * 64))
460 assert release_id == compute_release_id(repo_id="test-repo", tag="v1.0.0", commit_id=long_id("a" * 64))
461
462 def test_release_id_differs_by_tag(self, tmp_path: pathlib.Path) -> None:
463 from muse.core.store import compute_release_id
464 cid = long_id("a" * 64)
465 assert compute_release_id("repo", "v1.0.0", cid) != compute_release_id("repo", "v2.0.0", cid)
466
467 def test_release_id_differs_by_commit(self, tmp_path: pathlib.Path) -> None:
468 from muse.core.store import compute_release_id
469 assert compute_release_id("repo", "v1.0.0", long_id("a" * 64)) != compute_release_id("repo", "v1.0.0", long_id("b" * 64))
470
471
472 class TestRegisterFlags:
473 def test_default_json_out_is_false(self) -> None:
474 import argparse
475 from muse.cli.commands.release import register
476 p = argparse.ArgumentParser()
477 subs = p.add_subparsers()
478 register(subs)
479 args = p.parse_args(["release"])
480 assert args.json_out is False
481
482 def test_json_flag_sets_json_out(self) -> None:
483 import argparse
484 from muse.cli.commands.release import register
485 p = argparse.ArgumentParser()
486 subs = p.add_subparsers()
487 register(subs)
488 args = p.parse_args(["release", "--json"])
489 assert args.json_out is True
490
491 def test_j_shorthand_sets_json_out(self) -> None:
492 import argparse
493 from muse.cli.commands.release import register
494 p = argparse.ArgumentParser()
495 subs = p.add_subparsers()
496 register(subs)
497 args = p.parse_args(["release", "-j"])
498 assert args.json_out is True
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 124 days ago