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