gabriel / muse public
test_cmd_describe_hardening.py python
844 lines 30.4 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Hardening test suite for ``muse describe``.
2
3 Coverage:
4 - Unit: describe_commit core — no tags, exact, distance, long, match_pattern,
5 first_parent, abbrev, exact_match, _MAX_WALK budget, multi-tag tie-break
6 - Security: ANSI injection in tag names sanitized in text output,
7 raw in JSON; --ref ANSI passthrough in error message
8 - Error routing: all user errors routed to stderr
9 - JSON schema: _DescribeJson shape, all fields present, repo_id + branch
10 - New flags: --match, --exact-match, --first-parent, --abbrev, --json
11 - Integration: tag walk across merge commits, --first-parent vs full walk
12 - E2E: help output, combined flags
13 - Stress: 5 000-commit chain, 200-tag repo, concurrent reads
14 """
15
16 from __future__ import annotations
17
18 import datetime
19 import json
20 import pathlib
21 import threading
22 from typing import TypedDict
23 from unittest.mock import patch
24
25 import pytest
26 from tests.cli_test_helper import CliRunner, InvokeResult
27
28 from muse.core.describe import describe_commit, _MAX_WALK
29 from muse.core.object_store import write_object
30 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
31 from muse.core.store import CommitRecord, SnapshotRecord, TagRecord, write_commit, write_snapshot, write_tag
32 from muse.core.types import Manifest, blob_id, fake_id, content_hash
33 from muse.core.paths import muse_dir, ref_path
34
35 runner = CliRunner()
36 _REPO_ID = content_hash({"name": "describe-hard-test"})
37
38
39 # ---------------------------------------------------------------------------
40 # Helpers
41 # ---------------------------------------------------------------------------
42
43
44
45
46 def _init_repo(path: pathlib.Path, *, domain: str = "midi") -> pathlib.Path:
47 dot_muse = muse_dir(path)
48 for sub in ("commits", "snapshots", "objects", "refs/heads", "tags"):
49 (dot_muse / sub).mkdir(parents=True, exist_ok=True)
50 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
51 (dot_muse / "repo.json").write_text(
52 json.dumps({"repo_id": _REPO_ID, "domain": domain}),
53 encoding="utf-8",
54 )
55 return path
56
57
58 def _make_commit(
59 root: pathlib.Path,
60 parent_id: str | None = None,
61 parent2_id: str | None = None,
62 content: bytes = b"data",
63 branch: str = "main",
64 ) -> str:
65 obj_id = blob_id(content)
66 write_object(root, obj_id, content)
67 manifest = {f"f_{obj_id[len('sha256:'):len('sha256:') + 8]}.txt": obj_id}
68 snap_id = compute_snapshot_id(manifest)
69 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
70 committed_at = datetime.datetime.now(datetime.timezone.utc)
71 parent_ids = [pid for pid in (parent_id, parent2_id) if pid is not None]
72 commit_id = compute_commit_id(
73 parent_ids=parent_ids,
74 snapshot_id=snap_id,
75 message="msg",
76 committed_at_iso=committed_at.isoformat(),
77 )
78 rec = CommitRecord(
79 commit_id=commit_id,
80 repo_id=_REPO_ID,
81 branch=branch,
82 snapshot_id=snap_id,
83 message="msg",
84 committed_at=committed_at,
85 parent_commit_id=parent_id,
86 parent2_commit_id=parent2_id,
87 )
88 write_commit(root, rec)
89 (ref_path(root, branch)).write_text(
90 commit_id, encoding="utf-8"
91 )
92 return commit_id
93
94
95 def _make_tag(root: pathlib.Path, tag: str, commit_id: str) -> None:
96 write_tag(
97 root,
98 TagRecord(
99 tag_id=content_hash({"tag": tag, "commit_id": commit_id}),
100 repo_id=_REPO_ID,
101 tag=tag,
102 commit_id=commit_id,
103 created_at=datetime.datetime.now(datetime.timezone.utc),
104 ),
105 )
106
107
108 def _env(repo: pathlib.Path) -> Manifest:
109 return {"MUSE_REPO_ROOT": str(repo)}
110
111
112 def _invoke(args: list[str], env: Manifest) -> InvokeResult:
113 return runner.invoke(None, args, env=env)
114
115
116 class _DescribeOut(TypedDict):
117 commit_id: str
118 tag: str | None
119 distance: int
120 short_sha: str
121 name: str
122 exact: bool
123 repo_id: str
124 branch: str
125 duration_ms: float
126 exit_code: int
127
128
129 def _parse_json(result: InvokeResult) -> _DescribeOut:
130 raw = json.loads(result.output.strip())
131 return _DescribeOut(
132 commit_id=raw["commit_id"],
133 tag=raw["tag"],
134 distance=raw["distance"],
135 short_sha=raw["short_sha"],
136 name=raw["name"],
137 exact=raw["exact"],
138 repo_id=raw["repo_id"],
139 branch=raw["branch"],
140 duration_ms=raw["duration_ms"],
141 exit_code=raw["exit_code"],
142 )
143
144
145 # ---------------------------------------------------------------------------
146 # Unit: describe_commit core
147 # ---------------------------------------------------------------------------
148
149
150 def test_core_no_tags_returns_shortblob_id(tmp_path: pathlib.Path) -> None:
151 _init_repo(tmp_path)
152 cid = _make_commit(tmp_path, content=b"a")
153 r = describe_commit(tmp_path, _REPO_ID, cid)
154 assert r["tag"] is None
155 assert r["short_sha"] == cid[:len("sha256:") + 12]
156 assert r["name"] == r["short_sha"]
157 assert r["exact"] is False
158
159
160 def test_core_exact_tag(tmp_path: pathlib.Path) -> None:
161 _init_repo(tmp_path)
162 cid = _make_commit(tmp_path, content=b"b")
163 _make_tag(tmp_path, "v1.0.0", cid)
164 r = describe_commit(tmp_path, _REPO_ID, cid)
165 assert r["tag"] == "v1.0.0"
166 assert r["distance"] == 0
167 assert r["exact"] is True
168 assert r["name"] == "v1.0.0"
169
170
171 def test_core_distance_one(tmp_path: pathlib.Path) -> None:
172 _init_repo(tmp_path)
173 c1 = _make_commit(tmp_path, content=b"c1")
174 _make_tag(tmp_path, "v0.9", c1)
175 c2 = _make_commit(tmp_path, parent_id=c1, content=b"c2")
176 r = describe_commit(tmp_path, _REPO_ID, c2)
177 assert r["tag"] == "v0.9"
178 assert r["distance"] == 1
179 assert r["exact"] is False
180 assert r["name"] == "v0.9~1"
181
182
183 def test_core_long_format_on_tag(tmp_path: pathlib.Path) -> None:
184 _init_repo(tmp_path)
185 cid = _make_commit(tmp_path, content=b"long")
186 _make_tag(tmp_path, "v2.0.0", cid)
187 r = describe_commit(tmp_path, _REPO_ID, cid, long_format=True)
188 assert r["name"].startswith("v2.0.0-0-sha256:")
189
190
191 def test_core_long_format_with_distance(tmp_path: pathlib.Path) -> None:
192 _init_repo(tmp_path)
193 c1 = _make_commit(tmp_path, content=b"root")
194 _make_tag(tmp_path, "v1.0", c1)
195 c2 = _make_commit(tmp_path, parent_id=c1, content=b"next")
196 r = describe_commit(tmp_path, _REPO_ID, c2, long_format=True)
197 assert "-1-sha256:" in r["name"]
198
199
200 def test_core_abbrev_controls_sha_length(tmp_path: pathlib.Path) -> None:
201 _init_repo(tmp_path)
202 cid = _make_commit(tmp_path, content=b"abbrev")
203 r = describe_commit(tmp_path, _REPO_ID, cid, abbrev=8)
204 assert r["short_sha"] == cid[:len("sha256:") + 8]
205 assert r["short_sha"].startswith("sha256:")
206 assert len(r["short_sha"]) == len("sha256:") + 8
207
208
209 def test_core_match_pattern_filters_tags(tmp_path: pathlib.Path) -> None:
210 _init_repo(tmp_path)
211 cid = _make_commit(tmp_path, content=b"match")
212 _make_tag(tmp_path, "release-1", cid)
213 _make_tag(tmp_path, "v1.0.0", cid)
214 # Only semver tags — "release-1" excluded.
215 r = describe_commit(tmp_path, _REPO_ID, cid, match_pattern="v*")
216 assert r["tag"] == "v1.0.0"
217
218
219 def test_core_match_pattern_no_match_returnsblob_id(tmp_path: pathlib.Path) -> None:
220 _init_repo(tmp_path)
221 cid = _make_commit(tmp_path, content=b"nomatch")
222 _make_tag(tmp_path, "nightly-123", cid)
223 r = describe_commit(tmp_path, _REPO_ID, cid, match_pattern="v*")
224 assert r["tag"] is None
225 assert r["name"] == cid[:len("sha256:") + 12]
226
227
228 def test_core_exact_match_on_tag(tmp_path: pathlib.Path) -> None:
229 _init_repo(tmp_path)
230 cid = _make_commit(tmp_path, content=b"exact")
231 _make_tag(tmp_path, "v3.0", cid)
232 r = describe_commit(tmp_path, _REPO_ID, cid, exact_match=True)
233 assert r["tag"] == "v3.0"
234 assert r["exact"] is True
235
236
237 def test_core_exact_match_off_tag_returnsblob_id(tmp_path: pathlib.Path) -> None:
238 _init_repo(tmp_path)
239 c1 = _make_commit(tmp_path, content=b"root")
240 _make_tag(tmp_path, "v3.0", c1)
241 c2 = _make_commit(tmp_path, parent_id=c1, content=b"after")
242 r = describe_commit(tmp_path, _REPO_ID, c2, exact_match=True)
243 assert r["tag"] is None
244 assert r["name"] == c2[:len("sha256:") + 12]
245
246
247 def test_core_first_parent_skips_merge_branch(tmp_path: pathlib.Path) -> None:
248 """With --first-parent the tag on a merged branch is invisible."""
249 _init_repo(tmp_path)
250 # main: c1 → c3 (merge of feat)
251 # feat: c1 → c2 (tag here)
252 c1 = _make_commit(tmp_path, content=b"root")
253 c2 = _make_commit(tmp_path, parent_id=c1, content=b"feat", branch="feat")
254 _make_tag(tmp_path, "feat-tag", c2)
255 c3 = _make_commit(
256 tmp_path, parent_id=c1, parent2_id=c2, content=b"merge", branch="main"
257 )
258 # Without first_parent: feat-tag is reachable via second parent.
259 r_full = describe_commit(tmp_path, _REPO_ID, c3)
260 # With first_parent: only first parent chain; feat-tag not reachable.
261 r_fp = describe_commit(tmp_path, _REPO_ID, c3, first_parent=True)
262 assert r_fp["tag"] is None
263 # Full walk should find the tag (via c2).
264 assert r_full["tag"] == "feat-tag"
265
266
267 def test_core_multi_tag_same_commit_lex_greatest(tmp_path: pathlib.Path) -> None:
268 """When multiple tags point at the same commit, greatest lex name wins."""
269 _init_repo(tmp_path)
270 cid = _make_commit(tmp_path, content=b"multi")
271 _make_tag(tmp_path, "v1.0.0", cid)
272 _make_tag(tmp_path, "v2.0.0", cid)
273 _make_tag(tmp_path, "v1.5.0", cid)
274 r = describe_commit(tmp_path, _REPO_ID, cid)
275 assert r["tag"] == "v2.0.0"
276
277
278 def test_core_max_walk_budget(tmp_path: pathlib.Path) -> None:
279 """Walk stops at _MAX_WALK without crashing; returns short-SHA fallback."""
280 _init_repo(tmp_path)
281 # Inject a fake read_commit that always returns a parent so BFS never
282 # finds a commit-store miss — we just want to trigger the budget guard.
283 cid = _make_commit(tmp_path, content=b"budget")
284 _make_tag(tmp_path, "very-far", cid)
285
286 call_count = 0
287
288 import muse.core.store as _store
289 from muse.core.store import CommitRecord as _CR
290 import datetime as _dt
291
292 orig = _store.read_commit
293
294 def _fake_read(root: pathlib.Path, cid: str) -> _CR | None:
295 nonlocal call_count
296 call_count += 1
297 if call_count > _MAX_WALK + 5:
298 return None
299 fake_parent = fake_id(cid)
300 return _CR(
301 commit_id=cid,
302 branch="main",
303 snapshot_id="snap",
304 message="x",
305 committed_at=_dt.datetime.now(_dt.timezone.utc),
306 parent_commit_id=fake_parent,
307 )
308
309 with patch.object(_store, "read_commit", side_effect=_fake_read):
310 # Start from a commit far from any tag — BFS will exhaust budget.
311 far_commit = blob_id(b"far")
312 r = describe_commit(tmp_path, _REPO_ID, far_commit)
313
314 # Budget exhausted → tag not found → name is short SHA.
315 assert r["tag"] is None
316
317
318 # ---------------------------------------------------------------------------
319 # Security: ANSI injection in tag names
320 # ---------------------------------------------------------------------------
321
322
323 def test_ansi_in_tag_name_stripped_in_text_output(tmp_path: pathlib.Path) -> None:
324 _init_repo(tmp_path)
325 cid = _make_commit(tmp_path, content=b"ansi")
326 malicious_tag = "v1.0\x1b[31mRED\x1b[0m"
327 _make_tag(tmp_path, malicious_tag, cid)
328 result = _invoke(["describe"], _env(tmp_path))
329 assert result.exit_code == 0
330 assert "\x1b[31m" not in result.output
331
332
333 def test_ansi_in_tag_name_preserved_in_json(tmp_path: pathlib.Path) -> None:
334 """JSON output must not sanitize so callers see the raw value."""
335 _init_repo(tmp_path)
336 cid = _make_commit(tmp_path, content=b"ansi-json")
337 malicious_tag = "v1.0\x1b[31mRED\x1b[0m"
338 _make_tag(tmp_path, malicious_tag, cid)
339 result = _invoke(["describe", "--json"], _env(tmp_path))
340 assert result.exit_code == 0
341 data = _parse_json(result)
342 assert data["tag"] == malicious_tag
343
344
345 # ---------------------------------------------------------------------------
346 # Error routing: all user errors go to stderr
347 # ---------------------------------------------------------------------------
348
349
350 def test_no_commits_error_on_stderr(tmp_path: pathlib.Path) -> None:
351 _init_repo(tmp_path)
352 result = _invoke(["describe"], _env(tmp_path))
353 assert result.exit_code != 0
354 assert result.stderr != "" or "commits" in result.output.lower()
355
356
357 def test_ref_not_found_error_on_stderr(tmp_path: pathlib.Path) -> None:
358 _init_repo(tmp_path)
359 _make_commit(tmp_path, content=b"x")
360 result = _invoke(["describe", "--ref", "nonexistent"], _env(tmp_path))
361 assert result.exit_code != 0
362
363
364 def test_require_tag_no_tags_error(tmp_path: pathlib.Path) -> None:
365 _init_repo(tmp_path)
366 _make_commit(tmp_path, content=b"no-tag")
367 result = _invoke(["describe", "--require-tag"], _env(tmp_path))
368 assert result.exit_code != 0
369
370
371 def test_exact_match_not_on_tag_error(tmp_path: pathlib.Path) -> None:
372 _init_repo(tmp_path)
373 c1 = _make_commit(tmp_path, content=b"c1")
374 _make_tag(tmp_path, "v1", c1)
375 _make_commit(tmp_path, parent_id=c1, content=b"c2")
376 result = _invoke(["describe", "--exact-match"], _env(tmp_path))
377 assert result.exit_code != 0
378
379
380 def test_abbrev_too_small_error(tmp_path: pathlib.Path) -> None:
381 _init_repo(tmp_path)
382 _make_commit(tmp_path, content=b"ab")
383 result = _invoke(["describe", "--abbrev", "2"], _env(tmp_path))
384 assert result.exit_code != 0
385
386
387 def test_abbrev_too_large_error(tmp_path: pathlib.Path) -> None:
388 _init_repo(tmp_path)
389 _make_commit(tmp_path, content=b"ab")
390 result = _invoke(["describe", "--abbrev", "65"], _env(tmp_path))
391 assert result.exit_code != 0
392
393
394 # ---------------------------------------------------------------------------
395 # JSON schema: _DescribeJson
396 # ---------------------------------------------------------------------------
397
398
399 def test_json_schema_all_fields(tmp_path: pathlib.Path) -> None:
400 _init_repo(tmp_path)
401 cid = _make_commit(tmp_path, content=b"schema")
402 _make_tag(tmp_path, "v1.0.0", cid)
403 result = _invoke(["describe", "--json"], _env(tmp_path))
404 assert result.exit_code == 0
405 data = _parse_json(result)
406 assert data["tag"] == "v1.0.0"
407 assert data["distance"] == 0
408 assert data["exact"] is True
409 assert data["repo_id"] == _REPO_ID
410 assert data["branch"] == "main"
411 assert data["commit_id"] == cid
412 assert data["short_sha"] == cid[:len("sha256:") + 12]
413
414
415 def test_json_schema_no_tag(tmp_path: pathlib.Path) -> None:
416 _init_repo(tmp_path)
417 cid = _make_commit(tmp_path, content=b"no-tag-json")
418 result = _invoke(["describe", "--json"], _env(tmp_path))
419 assert result.exit_code == 0
420 data = _parse_json(result)
421 assert data["tag"] is None
422 assert data["name"] == cid[:len("sha256:") + 12]
423 assert data["exact"] is False
424
425
426 def test_json_schema_with_distance(tmp_path: pathlib.Path) -> None:
427 _init_repo(tmp_path)
428 c1 = _make_commit(tmp_path, content=b"root")
429 _make_tag(tmp_path, "v0.1", c1)
430 _make_commit(tmp_path, parent_id=c1, content=b"next")
431 result = _invoke(["describe", "--json"], _env(tmp_path))
432 assert result.exit_code == 0
433 data = _parse_json(result)
434 assert data["tag"] == "v0.1"
435 assert data["distance"] == 1
436 assert data["exact"] is False
437
438
439 def test_json_abbrev_reflected(tmp_path: pathlib.Path) -> None:
440 _init_repo(tmp_path)
441 cid = _make_commit(tmp_path, content=b"abbrev-json")
442 result = _invoke(["describe", "--abbrev", "8", "--json"], _env(tmp_path))
443 assert result.exit_code == 0
444 data = _parse_json(result)
445 assert data["short_sha"].startswith("sha256:")
446 assert len(data["short_sha"]) == len("sha256:") + 8
447
448
449 # ---------------------------------------------------------------------------
450 # New flags: --match, --exact-match, --first-parent, --abbrev
451 # ---------------------------------------------------------------------------
452
453
454 def test_flag_match_filters_tags(tmp_path: pathlib.Path) -> None:
455 _init_repo(tmp_path)
456 cid = _make_commit(tmp_path, content=b"match-flag")
457 _make_tag(tmp_path, "nightly-1", cid)
458 _make_tag(tmp_path, "v1.0.0", cid)
459 result = _invoke(["describe", "--match", "v*", "--json"], _env(tmp_path))
460 assert result.exit_code == 0
461 data = _parse_json(result)
462 assert data["tag"] == "v1.0.0"
463
464
465 def test_flag_match_no_matching_tag(tmp_path: pathlib.Path) -> None:
466 _init_repo(tmp_path)
467 cid = _make_commit(tmp_path, content=b"match-none")
468 _make_tag(tmp_path, "nightly-1", cid)
469 result = _invoke(["describe", "--match", "v*", "--json"], _env(tmp_path))
470 assert result.exit_code == 0
471 data = _parse_json(result)
472 assert data["tag"] is None
473
474
475 def test_flag_exact_match_on_tag(tmp_path: pathlib.Path) -> None:
476 _init_repo(tmp_path)
477 cid = _make_commit(tmp_path, content=b"exact-flag")
478 _make_tag(tmp_path, "v1.0", cid)
479 result = _invoke(["describe", "--exact-match", "--json"], _env(tmp_path))
480 assert result.exit_code == 0
481 data = _parse_json(result)
482 assert data["exact"] is True
483
484
485 def test_flag_exact_match_off_tag_fails(tmp_path: pathlib.Path) -> None:
486 _init_repo(tmp_path)
487 c1 = _make_commit(tmp_path, content=b"em-root")
488 _make_tag(tmp_path, "v1.0", c1)
489 _make_commit(tmp_path, parent_id=c1, content=b"em-next")
490 result = _invoke(["describe", "--exact-match"], _env(tmp_path))
491 assert result.exit_code != 0
492
493
494 def test_flag_first_parent(tmp_path: pathlib.Path) -> None:
495 _init_repo(tmp_path)
496 c1 = _make_commit(tmp_path, content=b"fp-root")
497 c2 = _make_commit(tmp_path, parent_id=c1, content=b"feat-side", branch="feat")
498 _make_tag(tmp_path, "side-tag", c2)
499 c3 = _make_commit(
500 tmp_path, parent_id=c1, parent2_id=c2, content=b"fp-merge", branch="main"
501 )
502 # --first-parent should not see side-tag.
503 result = _invoke(["describe", "--first-parent", "--json"], _env(tmp_path))
504 assert result.exit_code == 0
505 data = _parse_json(result)
506 assert data["tag"] is None # side-tag not reachable via first-parent
507
508
509 def test_flag_abbrev(tmp_path: pathlib.Path) -> None:
510 _init_repo(tmp_path)
511 _make_commit(tmp_path, content=b"abbrev-flag")
512 result = _invoke(["describe", "--abbrev", "16", "--json"], _env(tmp_path))
513 assert result.exit_code == 0
514 data = _parse_json(result)
515 assert len(data["short_sha"]) == len("sha256:") + 16
516
517
518 # ---------------------------------------------------------------------------
519 # Integration
520 # ---------------------------------------------------------------------------
521
522
523 def test_integration_ref_to_branch_tip(tmp_path: pathlib.Path) -> None:
524 _init_repo(tmp_path)
525 c1 = _make_commit(tmp_path, content=b"ref-root")
526 _make_tag(tmp_path, "v10.0", c1)
527 _make_commit(tmp_path, parent_id=c1, content=b"ref-next")
528 # Describe the HEAD (which is 1 hop past the tag).
529 result = _invoke(["describe", "--json"], _env(tmp_path))
530 assert result.exit_code == 0
531 data = _parse_json(result)
532 assert data["distance"] == 1
533 assert data["tag"] == "v10.0"
534
535
536 def test_integration_long_and_match_combined(tmp_path: pathlib.Path) -> None:
537 _init_repo(tmp_path)
538 cid = _make_commit(tmp_path, content=b"combo")
539 _make_tag(tmp_path, "v5.0.0", cid)
540 result = _invoke(
541 ["describe", "--long", "--match", "v*", "--json"], _env(tmp_path)
542 )
543 assert result.exit_code == 0
544 data = _parse_json(result)
545 assert data["name"].startswith("v5.0.0-0-sha256:")
546
547
548 def test_integration_require_tag_passes_when_tag_exists(
549 tmp_path: pathlib.Path,
550 ) -> None:
551 _init_repo(tmp_path)
552 cid = _make_commit(tmp_path, content=b"req-tag")
553 _make_tag(tmp_path, "v7.0", cid)
554 result = _invoke(["describe", "--require-tag", "--json"], _env(tmp_path))
555 assert result.exit_code == 0
556
557
558 def test_integration_text_output_sanitized(tmp_path: pathlib.Path) -> None:
559 _init_repo(tmp_path)
560 cid = _make_commit(tmp_path, content=b"text-sanitize")
561 _make_tag(tmp_path, "v1.0\x1b[1mBOLD\x1b[0m", cid)
562 result = _invoke(["describe"], _env(tmp_path))
563 assert result.exit_code == 0
564 assert "\x1b[1m" not in result.output
565
566
567 # ---------------------------------------------------------------------------
568 # E2E: help output
569 # ---------------------------------------------------------------------------
570
571
572 def test_help_contains_new_flags() -> None:
573 result = _invoke(["describe", "--help"], {})
574 assert result.exit_code == 0
575 for flag in ("--match", "--exact-match", "--first-parent", "--abbrev", "--json"):
576 assert flag in result.output, f"Missing flag in help: {flag}"
577
578
579 def test_help_mentions_json_schema() -> None:
580 result = _invoke(["describe", "--help"], {})
581 assert "json" in result.output.lower()
582
583
584 # ---------------------------------------------------------------------------
585 # Stress: deep ancestry + many tags + concurrent reads
586 # ---------------------------------------------------------------------------
587
588
589 def test_stress_5000_commit_chain(tmp_path: pathlib.Path) -> None:
590 _init_repo(tmp_path)
591 prev: str | None = None
592 root_cid = ""
593 for i in range(5_000):
594 cid = _make_commit(tmp_path, parent_id=prev, content=f"s{i}".encode())
595 if i == 0:
596 root_cid = cid
597 prev = cid
598
599 _make_tag(tmp_path, "v-deep", root_cid)
600 assert prev is not None
601 r = describe_commit(tmp_path, _REPO_ID, prev)
602 assert r["tag"] == "v-deep"
603 assert r["distance"] == 4_999
604
605
606 def test_stress_200_tags_repo(tmp_path: pathlib.Path) -> None:
607 """Many tags — describe still picks the nearest one efficiently."""
608 _init_repo(tmp_path)
609 commits: list[str] = []
610 prev: str | None = None
611 for i in range(200):
612 cid = _make_commit(tmp_path, parent_id=prev, content=f"t{i}".encode())
613 commits.append(cid)
614 # Tag every 10th commit.
615 if i % 10 == 0:
616 _make_tag(tmp_path, f"v{i}.0", cid)
617 prev = cid
618
619 # HEAD is commits[-1], nearest tag is v190.0 (at commits[190]).
620 r = describe_commit(tmp_path, _REPO_ID, commits[-1])
621 assert r["tag"] == "v190.0"
622 assert r["distance"] == 9
623
624
625 def test_stress_concurrent_describe(tmp_path: pathlib.Path) -> None:
626 """Concurrent --json calls must all return consistent, valid JSON."""
627 _init_repo(tmp_path)
628 c1 = _make_commit(tmp_path, content=b"conc-root")
629 _make_tag(tmp_path, "v-conc", c1)
630 _make_commit(tmp_path, parent_id=c1, content=b"conc-next")
631
632 invoke_lock = threading.Lock()
633 errors: list[str] = []
634
635 def _worker() -> None:
636 with invoke_lock:
637 r = _invoke(["describe", "--json"], _env(tmp_path))
638 try:
639 assert r.exit_code == 0
640 data = _parse_json(r)
641 assert data["tag"] == "v-conc"
642 assert data["distance"] == 1
643 except Exception as exc:
644 errors.append(str(exc))
645
646 threads = [threading.Thread(target=_worker) for _ in range(8)]
647 for t in threads:
648 t.start()
649 for t in threads:
650 t.join()
651
652 assert errors == [], f"Concurrent failures: {errors}"
653
654
655 # ---------------------------------------------------------------------------
656 # JSON schema: duration_ms + exit_code always present
657 # ---------------------------------------------------------------------------
658
659
660 class TestJsonSchemaComplete:
661 """Every --json path includes duration_ms and exit_code."""
662
663 def test_elapsed_present_on_tag(self, tmp_path: pathlib.Path) -> None:
664 _init_repo(tmp_path)
665 cid = _make_commit(tmp_path, content=b"sc-tag")
666 _make_tag(tmp_path, "v1.0.0", cid)
667 r = _invoke(["describe", "--json"], _env(tmp_path))
668 assert r.exit_code == 0
669 raw = json.loads(r.output)
670 assert "duration_ms" in raw
671 assert "exit_code" in raw
672
673 def test_elapsed_present_no_tag(self, tmp_path: pathlib.Path) -> None:
674 _init_repo(tmp_path)
675 _make_commit(tmp_path, content=b"sc-notag")
676 r = _invoke(["describe", "--json"], _env(tmp_path))
677 assert r.exit_code == 0
678 raw = json.loads(r.output)
679 assert "duration_ms" in raw
680 assert "exit_code" in raw
681
682 def test_elapsed_present_with_distance(self, tmp_path: pathlib.Path) -> None:
683 _init_repo(tmp_path)
684 c1 = _make_commit(tmp_path, content=b"sc-dist-root")
685 _make_tag(tmp_path, "v0.1", c1)
686 _make_commit(tmp_path, parent_id=c1, content=b"sc-dist-next")
687 r = _invoke(["describe", "--json"], _env(tmp_path))
688 assert r.exit_code == 0
689 raw = json.loads(r.output)
690 assert "duration_ms" in raw
691 assert raw["exit_code"] == 0
692
693 def test_elapsed_present_long_format(self, tmp_path: pathlib.Path) -> None:
694 _init_repo(tmp_path)
695 cid = _make_commit(tmp_path, content=b"sc-long")
696 _make_tag(tmp_path, "v2.0.0", cid)
697 r = _invoke(["describe", "--long", "--json"], _env(tmp_path))
698 assert r.exit_code == 0
699 raw = json.loads(r.output)
700 assert "duration_ms" in raw
701 assert raw["exit_code"] == 0
702
703 def test_elapsed_present_with_match(self, tmp_path: pathlib.Path) -> None:
704 _init_repo(tmp_path)
705 cid = _make_commit(tmp_path, content=b"sc-match")
706 _make_tag(tmp_path, "v3.0.0", cid)
707 r = _invoke(["describe", "--match", "v*", "--json"], _env(tmp_path))
708 assert r.exit_code == 0
709 raw = json.loads(r.output)
710 assert "duration_ms" in raw
711
712 def test_elapsed_present_with_abbrev(self, tmp_path: pathlib.Path) -> None:
713 _init_repo(tmp_path)
714 _make_commit(tmp_path, content=b"sc-abbrev")
715 r = _invoke(["describe", "--abbrev", "8", "--json"], _env(tmp_path))
716 assert r.exit_code == 0
717 raw = json.loads(r.output)
718 assert "duration_ms" in raw
719 assert raw["exit_code"] == 0
720
721 def test_all_eight_base_fields_present(self, tmp_path: pathlib.Path) -> None:
722 _init_repo(tmp_path)
723 cid = _make_commit(tmp_path, content=b"sc-all")
724 _make_tag(tmp_path, "v9.0.0", cid)
725 r = _invoke(["describe", "--json"], _env(tmp_path))
726 assert r.exit_code == 0
727 raw = json.loads(r.output)
728 for field in ("commit_id", "tag", "distance", "short_sha", "name",
729 "exact", "repo_id", "branch", "duration_ms", "exit_code"):
730 assert field in raw, f"Missing field: {field}"
731
732 def test_exit_code_field_is_zero(self, tmp_path: pathlib.Path) -> None:
733 _init_repo(tmp_path)
734 cid = _make_commit(tmp_path, content=b"sc-exit")
735 _make_tag(tmp_path, "v10.0.0", cid)
736 r = _invoke(["describe", "--json"], _env(tmp_path))
737 assert r.exit_code == 0
738 raw = json.loads(r.output)
739 assert raw["exit_code"] == 0
740
741
742 # ---------------------------------------------------------------------------
743 # duration_ms: type and magnitude checks
744 # ---------------------------------------------------------------------------
745
746
747 class TestElapsedSeconds:
748 """duration_ms is a non-negative float in a reasonable range."""
749
750 def test_elapsed_is_float(self, tmp_path: pathlib.Path) -> None:
751 _init_repo(tmp_path)
752 _make_commit(tmp_path, content=b"el-float")
753 r = _invoke(["describe", "--json"], _env(tmp_path))
754 raw = json.loads(r.output)
755 assert isinstance(raw["duration_ms"], float)
756
757 def test_elapsed_non_negative(self, tmp_path: pathlib.Path) -> None:
758 _init_repo(tmp_path)
759 _make_commit(tmp_path, content=b"el-nonneg")
760 r = _invoke(["describe", "--json"], _env(tmp_path))
761 raw = json.loads(r.output)
762 assert raw["duration_ms"] >= 0.0
763
764 def test_elapsed_under_ten_seconds(self, tmp_path: pathlib.Path) -> None:
765 _init_repo(tmp_path)
766 _make_commit(tmp_path, content=b"el-under")
767 r = _invoke(["describe", "--json"], _env(tmp_path))
768 raw = json.loads(r.output)
769 assert raw["duration_ms"] < 10.0
770
771 def test_elapsed_with_tag(self, tmp_path: pathlib.Path) -> None:
772 _init_repo(tmp_path)
773 cid = _make_commit(tmp_path, content=b"el-tag")
774 _make_tag(tmp_path, "v1.2.3", cid)
775 r = _invoke(["describe", "--json"], _env(tmp_path))
776 raw = json.loads(r.output)
777 assert raw["duration_ms"] >= 0.0
778
779 def test_elapsed_with_require_tag(self, tmp_path: pathlib.Path) -> None:
780 _init_repo(tmp_path)
781 cid = _make_commit(tmp_path, content=b"el-req")
782 _make_tag(tmp_path, "v1.0", cid)
783 r = _invoke(["describe", "--require-tag", "--json"], _env(tmp_path))
784 assert r.exit_code == 0
785 raw = json.loads(r.output)
786 assert "duration_ms" in raw
787
788 def test_elapsed_six_decimal_places(self, tmp_path: pathlib.Path) -> None:
789 _init_repo(tmp_path)
790 _make_commit(tmp_path, content=b"el-prec")
791 r = _invoke(["describe", "--json"], _env(tmp_path))
792 raw = json.loads(r.output)
793 # round(..., 6) produces at most 6 decimal places — str check
794 s = str(raw["duration_ms"])
795 dec = s.split(".")[-1] if "." in s else ""
796 assert len(dec) <= 6
797
798
799 # ---------------------------------------------------------------------------
800 # exit_code field
801 # ---------------------------------------------------------------------------
802
803
804 class TestExitCode:
805 """exit_code field mirrors process exit code; always 0 on success."""
806
807 def test_exit_code_zero_no_tag(self, tmp_path: pathlib.Path) -> None:
808 _init_repo(tmp_path)
809 _make_commit(tmp_path, content=b"ec-notag")
810 r = _invoke(["describe", "--json"], _env(tmp_path))
811 assert r.exit_code == 0
812 assert json.loads(r.output)["exit_code"] == 0
813
814 def test_exit_code_zero_on_tag(self, tmp_path: pathlib.Path) -> None:
815 _init_repo(tmp_path)
816 cid = _make_commit(tmp_path, content=b"ec-tag")
817 _make_tag(tmp_path, "v1.0.0", cid)
818 r = _invoke(["describe", "--json"], _env(tmp_path))
819 assert r.exit_code == 0
820 assert json.loads(r.output)["exit_code"] == 0
821
822 def test_exit_code_zero_with_distance(self, tmp_path: pathlib.Path) -> None:
823 _init_repo(tmp_path)
824 c1 = _make_commit(tmp_path, content=b"ec-dist-root")
825 _make_tag(tmp_path, "v0.5", c1)
826 _make_commit(tmp_path, parent_id=c1, content=b"ec-dist-next")
827 r = _invoke(["describe", "--json"], _env(tmp_path))
828 assert r.exit_code == 0
829 assert json.loads(r.output)["exit_code"] == 0
830
831 def test_exit_code_zero_long_format(self, tmp_path: pathlib.Path) -> None:
832 _init_repo(tmp_path)
833 cid = _make_commit(tmp_path, content=b"ec-long")
834 _make_tag(tmp_path, "v2.0.0", cid)
835 r = _invoke(["describe", "--long", "--json"], _env(tmp_path))
836 assert r.exit_code == 0
837 assert json.loads(r.output)["exit_code"] == 0
838
839 def test_exit_code_zero_with_abbrev(self, tmp_path: pathlib.Path) -> None:
840 _init_repo(tmp_path)
841 _make_commit(tmp_path, content=b"ec-abbrev")
842 r = _invoke(["describe", "--abbrev", "10", "--json"], _env(tmp_path))
843 assert r.exit_code == 0
844 assert json.loads(r.output)["exit_code"] == 0
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago