gabriel / muse public
test_integrity_I7_history_walk.py python
738 lines 27.5 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Phase 1.7 — Linux-kernel scale: commit history walk.
2
3 Tests cover:
4 - 15k commit chain: walk_commits_between_result truncation flag
5 - 15k commit chain: muse log --json emits "truncated" in JSON
6 - 60k-deep branches: find_merge_base raises a clear error (not wrong answer)
7 - commit_graph on 15k: emits "truncated": true (JSON + text + count-only)
8 - Configurable caps via [limits] in config.toml
9 - O(n²) regression: _collect_all_commits uses deque (benchmarked)
10 - Streaming JSON: muse log --json doesn't hold all commits in memory
11 - walk_commits_between_result returns WalkResult TypedDict
12 - Truncation NOT flagged when walk naturally completes under cap
13 - find_merge_base raises on BOTH A-side and B-side cap hit
14 - get_commits_for_branch respects configurable cap via walk_limit
15 """
16
17 from __future__ import annotations
18 from collections.abc import Mapping
19
20 type _ConfigMap = dict[str, int]
21
22 import collections
23 import datetime
24 import json
25 import pathlib
26 import sys
27 import time
28 import tomllib
29 import unittest.mock as mock
30 from typing import TypedDict
31
32 import pytest
33
34 from tests.cli_test_helper import CliRunner
35
36
37 class _LogOutput(TypedDict, total=False):
38 """Shape of muse log --json output."""
39
40 truncated: bool
41 commits: list[Mapping[str, str]]
42
43
44 class _CommitJson(TypedDict, total=False):
45 """Shape of a single commit entry in muse log --json."""
46
47 commit_id: str
48 branch: str
49 message: str
50 author: str
51 committed_at: str
52 parent_commit_id: str
53 snapshot_id: str
54 metadata: Manifest
55 sem_ver_bump: str
56
57 from muse.core.merge_engine import find_merge_base
58 from muse.core.snapshot import compute_commit_id
59
60 from muse.core.types import Manifest, fake_id
61 from muse.core.store import (
62 CommitRecord,
63 WalkResult,
64 get_commits_for_branch,
65 walk_commits_between,
66 walk_commits_between_result,
67 write_commit,
68 )
69 from muse.core.paths import config_toml_path, heads_dir, muse_dir
70
71 # ---------------------------------------------------------------------------
72 # Helpers
73 # ---------------------------------------------------------------------------
74
75
76 _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
77
78
79 def _repo(tmp_path: pathlib.Path) -> pathlib.Path:
80 """Create a minimal .muse/ directory structure."""
81 dot_muse = muse_dir(tmp_path)
82 (dot_muse / "commits").mkdir(parents=True)
83 (dot_muse / "snapshots").mkdir(parents=True)
84 (dot_muse / "refs" / "heads").mkdir(parents=True)
85 (dot_muse / "repo.json").write_text(json.dumps({"repo_id": "test-repo"}))
86 (dot_muse / "HEAD").write_text("ref: refs/heads/main\n")
87 (dot_muse / "refs" / "heads" / "main").write_text("")
88 return tmp_path
89
90
91 def _make_commit(
92 root: pathlib.Path,
93 label: str = "",
94 message: str = "msg",
95 parent: str | None = None,
96 parent2: str | None = None,
97 branch: str = "main",
98 ) -> CommitRecord:
99 """Write a commit with a real content-addressed ID derived from its inputs.
100
101 *label* is used to derive a unique snapshot_id; it need not be a real
102 snapshot in the object store. The commit_id is computed via
103 ``compute_commit_id`` so that ``read_commit`` can verify it on read.
104 """
105 snapshot_id = fake_id(label) if label else fake_id("default")
106 parent_ids = [p for p in [parent, parent2] if p is not None]
107 commit_id = compute_commit_id(
108 parent_ids=parent_ids,
109 snapshot_id=snapshot_id,
110 message=message,
111 committed_at_iso=_DT.isoformat(),
112 )
113 c = CommitRecord(
114 repo_id="test-repo",
115 commit_id=commit_id,
116 branch=branch,
117 snapshot_id=snapshot_id,
118 message=message,
119 committed_at=_DT,
120 parent_commit_id=parent,
121 parent2_commit_id=parent2,
122 )
123 write_commit(root, c)
124 return c
125
126
127 def _build_linear_chain(root: pathlib.Path, n: int) -> list[str]:
128 """Write a linear chain of *n* commits; return list of IDs newest-first."""
129 real_ids: list[str] = []
130 prev: str | None = None
131 for i in range(n):
132 record = _make_commit(root, f"commit_{i:08d}", parent=prev)
133 prev = record.commit_id
134 real_ids.append(record.commit_id)
135 # HEAD points to the last commit (newest)
136 tip = real_ids[-1]
137 (heads_dir(root) / "main").write_text(tip)
138 return list(reversed(real_ids)) # newest-first
139
140
141 def _write_config(root: pathlib.Path, limits: _ConfigMap) -> None:
142 """Write a [limits] section to .muse/config.toml."""
143 lines = ["[limits]\n"]
144 for k, v in limits.items():
145 lines.append(f"{k} = {v}\n")
146 (config_toml_path(root)).write_text("".join(lines))
147
148
149 # ---------------------------------------------------------------------------
150 # 1. WalkResult TypedDict
151 # ---------------------------------------------------------------------------
152
153
154 class TestWalkResultType:
155 def test_walk_result_is_typed_dict(self, tmp_path: pathlib.Path) -> None:
156 root = _repo(tmp_path)
157 ids = _build_linear_chain(root, 5)
158 result = walk_commits_between_result(root, ids[0], max_commits=100)
159 assert isinstance(result, dict)
160 assert "commits" in result
161 assert "truncated" in result
162 assert "count" in result
163
164 def test_walk_result_not_truncated_when_chain_fits(
165 self, tmp_path: pathlib.Path
166 ) -> None:
167 root = _repo(tmp_path)
168 ids = _build_linear_chain(root, 10)
169 result = walk_commits_between_result(root, ids[0], max_commits=100)
170 assert result["truncated"] is False
171 assert result["count"] == 10
172 assert len(result["commits"]) == 10
173
174 def test_walk_result_truncated_at_cap(self, tmp_path: pathlib.Path) -> None:
175 root = _repo(tmp_path)
176 ids = _build_linear_chain(root, 50)
177 result = walk_commits_between_result(root, ids[0], max_commits=20)
178 assert result["truncated"] is True
179 assert result["count"] == 20
180 assert len(result["commits"]) == 20
181
182 def test_walk_commits_between_backward_compat(
183 self, tmp_path: pathlib.Path
184 ) -> None:
185 """walk_commits_between still returns list[CommitRecord]."""
186 root = _repo(tmp_path)
187 ids = _build_linear_chain(root, 5)
188 result = walk_commits_between(root, ids[0], max_commits=100)
189 assert isinstance(result, list)
190 assert len(result) == 5
191
192 def test_count_matches_len_commits(self, tmp_path: pathlib.Path) -> None:
193 root = _repo(tmp_path)
194 ids = _build_linear_chain(root, 30)
195 for cap in (5, 10, 30, 100):
196 r = walk_commits_between_result(root, ids[0], max_commits=cap)
197 assert r["count"] == len(r["commits"])
198
199
200 # ---------------------------------------------------------------------------
201 # 2. 15k commit chain — truncation and correctness
202 # ---------------------------------------------------------------------------
203
204
205 @pytest.mark.slow
206 class TestLinearChainScale:
207 def test_15k_chain_walk_truncates_at_default_cap(
208 self, tmp_path: pathlib.Path
209 ) -> None:
210 """15k chain with default cap (10k) → truncated=True, 10k commits."""
211 root = _repo(tmp_path)
212 ids = _build_linear_chain(root, 15_000)
213 result = walk_commits_between_result(root, ids[0]) # default cap = 10k
214 assert result["truncated"] is True
215 assert result["count"] == 10_000
216
217 def test_15k_chain_walk_completes_with_raised_cap(
218 self, tmp_path: pathlib.Path
219 ) -> None:
220 """15k chain with cap=20k → truncated=False, 15k commits."""
221 root = _repo(tmp_path)
222 ids = _build_linear_chain(root, 15_000)
223 result = walk_commits_between_result(root, ids[0], max_commits=20_000)
224 assert result["truncated"] is False
225 assert result["count"] == 15_000
226
227 def test_15k_chain_order_newest_first(self, tmp_path: pathlib.Path) -> None:
228 """First commit returned must be the tip (newest)."""
229 root = _repo(tmp_path)
230 ids = _build_linear_chain(root, 100)
231 result = walk_commits_between_result(root, ids[0], max_commits=200)
232 assert result["commits"][0].commit_id == ids[0] # newest first
233
234
235 # ---------------------------------------------------------------------------
236 # 3. Configurable caps via [limits] in config.toml
237 # ---------------------------------------------------------------------------
238
239
240 class TestConfigurableCaps:
241 def test_walk_cap_from_config(self, tmp_path: pathlib.Path) -> None:
242 root = _repo(tmp_path)
243 _build_linear_chain(root, 200)
244 _write_config(root, {"max_walk_commits": 50})
245
246 from muse.cli.config import get_limit
247 cap = get_limit("max_walk_commits", root)
248 assert cap == 50
249
250 def test_graph_cap_from_config(self, tmp_path: pathlib.Path) -> None:
251 root = _repo(tmp_path)
252 _write_config(root, {"max_graph_commits": 1000})
253
254 from muse.cli.config import get_limit
255 assert get_limit("max_graph_commits", root) == 1000
256
257 def test_ancestors_cap_from_config(self, tmp_path: pathlib.Path) -> None:
258 root = _repo(tmp_path)
259 _write_config(root, {"max_ancestors": 999})
260
261 from muse.cli.config import get_limit
262 assert get_limit("max_ancestors", root) == 999
263
264 def test_default_cap_when_config_absent(self, tmp_path: pathlib.Path) -> None:
265 root = _repo(tmp_path)
266 from muse.cli.config import (
267 _DEFAULT_MAX_ANCESTORS,
268 _DEFAULT_MAX_WALK_COMMITS,
269 get_limit,
270 )
271 assert get_limit("max_walk_commits", root) == _DEFAULT_MAX_WALK_COMMITS
272 assert get_limit("max_ancestors", root) == _DEFAULT_MAX_ANCESTORS
273
274 def test_invalid_cap_ignored_uses_default(self, tmp_path: pathlib.Path) -> None:
275 """Negative or zero limits in config must be ignored — use default."""
276 root = _repo(tmp_path)
277 _write_config(root, {"max_walk_commits": -1}) # invalid
278
279 from muse.cli.config import _DEFAULT_MAX_WALK_COMMITS, get_limit
280 assert get_limit("max_walk_commits", root) == _DEFAULT_MAX_WALK_COMMITS
281
282 def test_get_config_value_reads_limits(self, tmp_path: pathlib.Path) -> None:
283 root = _repo(tmp_path)
284 _write_config(root, {"max_walk_commits": 777})
285
286 from muse.cli.config import get_config_value
287 assert get_config_value("limits.max_walk_commits", root) == "777"
288
289 def test_limits_config_parsed_correctly(self, tmp_path: pathlib.Path) -> None:
290 """All three limit keys are correctly parsed from config.toml."""
291 root = _repo(tmp_path)
292 _write_config(root, {
293 "max_walk_commits": 111,
294 "max_ancestors": 222,
295 "max_graph_commits": 333,
296 })
297
298 from muse.cli.config import get_limit
299 assert get_limit("max_walk_commits", root) == 111
300 assert get_limit("max_ancestors", root) == 222
301 assert get_limit("max_graph_commits", root) == 333
302
303
304 # ---------------------------------------------------------------------------
305 # 4. find_merge_base — consistency at cap (both sides raise)
306 # ---------------------------------------------------------------------------
307
308
309 class TestFindMergeBaseAtCap:
310 def _make_diverging_branches(
311 self, root: pathlib.Path, depth_a: int, depth_b: int
312 ) -> tuple[str, str, str]:
313 """Create a common root then two branches of given depths.
314 Returns (tip_a, tip_b, common_id)."""
315 common_record = _make_commit(root, "common", "common root")
316 common_id = common_record.commit_id
317
318 prev_a = common_id
319 for i in range(depth_a):
320 record = _make_commit(root, f"branch_a_{i}", parent=prev_a, branch="branch-a")
321 prev_a = record.commit_id
322 tip_a = prev_a
323
324 prev_b = common_id
325 for i in range(depth_b):
326 record = _make_commit(root, f"branch_b_{i}", parent=prev_b, branch="branch-b")
327 prev_b = record.commit_id
328 tip_b = prev_b
329
330 return tip_a, tip_b, common_id
331
332 def test_small_graph_finds_base_correctly(
333 self, tmp_path: pathlib.Path
334 ) -> None:
335 root = _repo(tmp_path)
336 tip_a, tip_b, common_id = self._make_diverging_branches(root, 5, 5)
337 base = find_merge_base(root, tip_a, tip_b)
338 assert base == common_id
339
340 def test_a_side_cap_raises_muse_cli_error(
341 self, tmp_path: pathlib.Path
342 ) -> None:
343 """A-side exceeding cap must raise MuseCLIError (not silently truncate)."""
344 from muse.core.errors import MuseCLIError
345 root = _repo(tmp_path)
346 # Two branches each 110 commits deep from a common ancestor — the BFS
347 # cannot find the merge base within the 100-ancestor cap.
348 tip_a, tip_b, _ = self._make_diverging_branches(root, 110, 110)
349
350 with mock.patch("muse.cli.config.get_limit", return_value=100):
351 with pytest.raises(MuseCLIError, match="Ancestor graph exceeds"):
352 find_merge_base(root, tip_a, tip_b)
353
354 def test_b_side_cap_also_raises_muse_cli_error(
355 self, tmp_path: pathlib.Path
356 ) -> None:
357 """B-side exceeding cap must also raise MuseCLIError — consistent behavior."""
358 from muse.core.errors import MuseCLIError
359 root = _repo(tmp_path)
360
361 # Two branches from a common ancestor deep in the history.
362 # A-side is short (won't hit cap), B-side is long.
363 common_record = _make_commit(root, "deep_common", "common")
364 common_id = common_record.commit_id
365
366 # B-side: 110 commits
367 prev = common_id
368 tip_b = common_id
369 for i in range(110):
370 record = _make_commit(root, f"b_side_{i}", parent=prev, branch="b")
371 prev = record.commit_id
372 tip_b = record.commit_id
373
374 # A-side: 5 commits (tiny — won't hit cap)
375 prev = common_id
376 tip_a = common_id
377 for i in range(5):
378 record = _make_commit(root, f"a_side_{i}", parent=prev, branch="a")
379 prev = record.commit_id
380 tip_a = record.commit_id
381
382 # With cap=100, B-side has 110 commits → raises
383 with mock.patch("muse.cli.config.get_limit", return_value=100):
384 with pytest.raises(MuseCLIError, match="Ancestor graph"):
385 find_merge_base(root, tip_a, tip_b)
386
387 def test_error_message_mentions_config_key(
388 self, tmp_path: pathlib.Path
389 ) -> None:
390 """Error message must tell users how to raise the cap."""
391 from muse.core.errors import MuseCLIError
392 root = _repo(tmp_path)
393 tip_a, tip_b, _ = self._make_diverging_branches(root, 110, 110)
394
395 with mock.patch("muse.cli.config.get_limit", return_value=100):
396 with pytest.raises(MuseCLIError) as exc_info:
397 find_merge_base(root, tip_a, tip_b)
398 assert "max_ancestors" in str(exc_info.value)
399 assert "config.toml" in str(exc_info.value)
400
401 @pytest.mark.slow
402 def test_60k_deep_branches_raise_not_wrong_answer(
403 self, tmp_path: pathlib.Path
404 ) -> None:
405 """Two 60k-deep branches: find_merge_base raises, never silently truncates."""
406 from muse.core.errors import MuseCLIError
407 root = _repo(tmp_path)
408 # Use cap=50k (default); build 52k branches — enough to exceed cap, no excess
409 common_record = _make_commit(root, "root60k", "root")
410 common_id = common_record.commit_id
411
412 n = 52_000
413 prev_a = common_id
414 for i in range(n):
415 record = _make_commit(root, f"a60k_{i}", parent=prev_a, branch="a")
416 prev_a = record.commit_id
417 tip_a = prev_a
418
419 # B-side: only 10 commits — A-side will hit the cap first
420 prev_b = common_id
421 for i in range(10):
422 record = _make_commit(root, f"b10_{i}", parent=prev_b, branch="b")
423 prev_b = record.commit_id
424 tip_b = prev_b
425
426 # find_merge_base must raise, not silently return None/wrong answer
427 with pytest.raises(MuseCLIError, match="Ancestor graph exceeds"):
428 find_merge_base(root, tip_a, tip_b)
429
430
431 # ---------------------------------------------------------------------------
432 # 5. _collect_all_commits — delegates to iter_ancestors (O(1) popleft
433 # guaranteed by graph.py; no O(n²) list.pop(0) pattern here)
434 # ---------------------------------------------------------------------------
435
436
437 class TestCollectAllCommitsPerformance:
438 def test_uses_deque_not_list_for_bfs(self) -> None:
439 """_collect_all_commits must delegate to iter_ancestors; no inline BFS."""
440 from muse.cli.commands import log as log_mod
441 import inspect
442 import ast
443 source = inspect.getsource(log_mod._collect_all_commits)
444 # Parse AST to check code (not docstring) for list.pop(0) pattern
445 tree = ast.parse(source)
446 pop0_calls: list[ast.Call] = []
447 for node in ast.walk(tree):
448 if (
449 isinstance(node, ast.Call)
450 and isinstance(node.func, ast.Attribute)
451 and node.func.attr == "pop"
452 and node.args
453 and isinstance(node.args[0], ast.Constant)
454 and node.args[0].value == 0
455 ):
456 pop0_calls.append(node)
457 assert not pop0_calls, (
458 "Found list.pop(0) in _collect_all_commits — this is the O(n²) bug. "
459 "Replace with deque.popleft()."
460 )
461 assert "iter_ancestors" in source, (
462 "_collect_all_commits must delegate to iter_ancestors. "
463 "O(1) popleft is guaranteed by graph.py's walk_dag."
464 )
465
466 def test_collect_returns_tuple_with_truncated_flag(
467 self, tmp_path: pathlib.Path
468 ) -> None:
469 root = _repo(tmp_path)
470 ids = _build_linear_chain(root, 20)
471
472 from muse.cli.commands.log import _collect_all_commits
473 commits, truncated = _collect_all_commits(root, [ids[0]], max_commits=100)
474 assert isinstance(commits, dict)
475 assert isinstance(truncated, bool)
476 assert truncated is False
477 assert len(commits) == 20
478
479 def test_collect_truncates_at_cap(self, tmp_path: pathlib.Path) -> None:
480 root = _repo(tmp_path)
481 ids = _build_linear_chain(root, 50)
482
483 from muse.cli.commands.log import _collect_all_commits
484 commits, truncated = _collect_all_commits(root, [ids[0]], max_commits=10)
485 assert truncated is True
486 assert len(commits) == 10
487
488 @pytest.mark.slow
489 def test_10k_commits_completes_in_reasonable_time(
490 self, tmp_path: pathlib.Path
491 ) -> None:
492 """10k BFS must complete in < 5s — proves O(n) not O(n²)."""
493 root = _repo(tmp_path)
494 ids = _build_linear_chain(root, 10_000)
495
496 from muse.cli.commands.log import _collect_all_commits
497 t0 = time.perf_counter()
498 commits, _ = _collect_all_commits(root, [ids[0]], max_commits=100_000)
499 elapsed = time.perf_counter() - t0
500
501 assert len(commits) == 10_000
502 assert elapsed < 5.0, (
503 f"_collect_all_commits took {elapsed:.2f}s for 10k commits. "
504 "Expected < 5s. O(n²) list.pop(0) would take ~50s."
505 )
506
507
508 # ---------------------------------------------------------------------------
509 # 6. muse log --json streaming output with "truncated" field
510 # ---------------------------------------------------------------------------
511
512
513 class TestLogJsonOutput:
514 def _run_log(self, root: pathlib.Path, *extra_args: str) -> _LogOutput:
515 """Run muse log --json via CliRunner and parse the output."""
516 runner = CliRunner()
517 result = runner.invoke(None, ["log", "--json"], env={"MUSE_REPO_ROOT": str(root)})
518 out = result.stdout.strip()
519 if not out:
520 return _LogOutput()
521 parsed: _LogOutput = json.loads(out)
522 return parsed
523
524 def test_log_json_has_truncated_field(self, tmp_path: pathlib.Path) -> None:
525 """muse log --json must include a 'truncated' key in output."""
526 root = _repo(tmp_path)
527 ids = _build_linear_chain(root, 5)
528 (heads_dir(root) / "main").write_text(ids[0])
529
530 output = self._run_log(root)
531 assert "truncated" in output, (
532 "muse log --json must include 'truncated' key. "
533 "Agents rely on this to know whether to page."
534 )
535
536 def test_log_json_not_truncated_for_small_history(
537 self, tmp_path: pathlib.Path
538 ) -> None:
539 root = _repo(tmp_path)
540 ids = _build_linear_chain(root, 5)
541 (heads_dir(root) / "main").write_text(ids[0])
542
543 output = self._run_log(root)
544 assert output["truncated"] is False
545
546 def test_log_json_has_commits_array(self, tmp_path: pathlib.Path) -> None:
547 root = _repo(tmp_path)
548 ids = _build_linear_chain(root, 3)
549 (heads_dir(root) / "main").write_text(ids[0])
550
551 output = self._run_log(root)
552 commits = output.get("commits")
553 assert isinstance(commits, list)
554 assert len(commits) == 3
555
556 def test_log_json_commit_fields(self, tmp_path: pathlib.Path) -> None:
557 """Each commit in JSON output has the required agent-facing fields."""
558 root = _repo(tmp_path)
559 ids = _build_linear_chain(root, 2)
560 (heads_dir(root) / "main").write_text(ids[0])
561
562 output = self._run_log(root)
563 commits_raw = output.get("commits", [])
564 assert isinstance(commits_raw, list)
565 assert len(commits_raw) >= 1
566 # Verify required fields are present in the raw dict
567 first_raw = commits_raw[0]
568 assert isinstance(first_raw, dict)
569 required_fields = {
570 "commit_id", "branch", "message", "author",
571 "committed_at", "parent_commit_id", "snapshot_id",
572 "metadata", "sem_ver_bump",
573 }
574 assert required_fields.issubset(set(first_raw.keys())), (
575 f"Missing fields: {required_fields - set(first_raw.keys())}"
576 )
577
578 def test_log_json_empty_history(self, tmp_path: pathlib.Path) -> None:
579 """Empty history emits a valid JSON response (not an exception)."""
580 root = _repo(tmp_path)
581 # No commits, HEAD is empty — must not crash
582 output = self._run_log(root)
583 # Valid outcomes: empty dict (branch not found) or {"truncated":false,"commits":[]}
584 if output:
585 assert "commits" in output or output == {}
586
587
588 # ---------------------------------------------------------------------------
589 # 7. commit_graph — truncated in all output formats
590 # ---------------------------------------------------------------------------
591
592
593 class TestCommitGraphTruncation:
594 def _run_commit_graph(
595 self,
596 root: pathlib.Path,
597 tip: str,
598 fmt: str = "json",
599 max_commits: int = 10_000,
600 count_only: bool = False,
601 ) -> str:
602 args = ["commit-graph", "--tip", tip, "--max", str(max_commits)]
603 if fmt == "json":
604 args.append("--json")
605 if count_only:
606 args.append("--count")
607 runner = CliRunner()
608 result = runner.invoke(None, args, env={"MUSE_REPO_ROOT": str(root)})
609 return result.stdout
610
611 def test_json_has_truncated_false_when_under_cap(
612 self, tmp_path: pathlib.Path
613 ) -> None:
614 root = _repo(tmp_path)
615 ids = _build_linear_chain(root, 10)
616 out = json.loads(self._run_commit_graph(root, ids[0], max_commits=100))
617 assert out["truncated"] is False
618 assert out["count"] == 10
619
620 def test_json_has_truncated_true_when_over_cap(
621 self, tmp_path: pathlib.Path
622 ) -> None:
623 root = _repo(tmp_path)
624 ids = _build_linear_chain(root, 50)
625 out = json.loads(self._run_commit_graph(root, ids[0], max_commits=20))
626 assert out["truncated"] is True
627 assert out["count"] == 20
628
629 def test_text_format_has_truncated_comment_when_over_cap(
630 self, tmp_path: pathlib.Path
631 ) -> None:
632 root = _repo(tmp_path)
633 ids = _build_linear_chain(root, 50)
634 out = self._run_commit_graph(root, ids[0], fmt="text", max_commits=10)
635 assert "TRUNCATED" in out, (
636 "text format must emit '# TRUNCATED' when cap is hit"
637 )
638
639 def test_text_format_no_truncated_when_under_cap(
640 self, tmp_path: pathlib.Path
641 ) -> None:
642 root = _repo(tmp_path)
643 ids = _build_linear_chain(root, 5)
644 out = self._run_commit_graph(root, ids[0], fmt="text", max_commits=100)
645 assert "TRUNCATED" not in out
646
647 def test_count_only_has_truncated_field(
648 self, tmp_path: pathlib.Path
649 ) -> None:
650 root = _repo(tmp_path)
651 ids = _build_linear_chain(root, 50)
652 out = json.loads(self._run_commit_graph(
653 root, ids[0], max_commits=20, count_only=True
654 ))
655 assert "truncated" in out, "count-only must include 'truncated'"
656 assert out["truncated"] is True
657 assert out["count"] == 20
658
659 @pytest.mark.slow
660 def test_15k_chain_commit_graph_completes_in_30s(
661 self, tmp_path: pathlib.Path
662 ) -> None:
663 """commit-graph on 15k commits must complete in < 30s."""
664 root = _repo(tmp_path)
665 ids = _build_linear_chain(root, 15_000)
666
667 t0 = time.perf_counter()
668 out = json.loads(self._run_commit_graph(root, ids[0], max_commits=10_000))
669 elapsed = time.perf_counter() - t0
670
671 assert out["truncated"] is True
672 assert elapsed < 30.0, (
673 f"commit-graph on 15k commits took {elapsed:.1f}s — must be < 30s"
674 )
675
676
677 # ---------------------------------------------------------------------------
678 # 8. Regression: B-side was returning None silently (old bug)
679 # ---------------------------------------------------------------------------
680
681
682 class TestMergeBaseConsistency:
683 def test_a_and_b_cap_raise_same_type(self, tmp_path: pathlib.Path) -> None:
684 """Both A-side and B-side cap must raise the same exception type."""
685 from muse.core.errors import MuseCLIError
686 root = _repo(tmp_path)
687
688 # Build a 30-commit chain
689 ids = _build_linear_chain(root, 30)
690 tip_a = ids[0] # newest
691 tip_b = ids[15] # halfway
692
693 # With cap=20, A-side will be exhausted (30 > 20)
694 with mock.patch("muse.cli.config.get_limit", return_value=20):
695 with pytest.raises(MuseCLIError):
696 find_merge_base(root, tip_a, tip_b)
697
698 def test_symmetric_result_for_small_graph(
699 self, tmp_path: pathlib.Path
700 ) -> None:
701 """find_merge_base(a, b) == find_merge_base(b, a) for a small graph."""
702 root = _repo(tmp_path)
703 common_record = _make_commit(root, "sym_root")
704 common_id = common_record.commit_id
705
706 tip_a_record = _make_commit(root, "sym_a_1", parent=common_id)
707 tip_a = tip_a_record.commit_id
708 tip_b_record = _make_commit(root, "sym_b_1", parent=common_id)
709 tip_b = tip_b_record.commit_id
710
711 ab = find_merge_base(root, tip_a, tip_b)
712 ba = find_merge_base(root, tip_b, tip_a)
713 assert ab == ba == common_id
714
715
716 # ---------------------------------------------------------------------------
717 # 9. walk_commits_between_result — from_commit_id exclusion
718 # ---------------------------------------------------------------------------
719
720
721 class TestWalkFromCommitExclusion:
722 def test_from_commit_excluded(self, tmp_path: pathlib.Path) -> None:
723 """from_commit_id is exclusive — it must not appear in the result."""
724 root = _repo(tmp_path)
725 ids = _build_linear_chain(root, 10) # newest first
726 stop = ids[5] # stop before this one
727 result = walk_commits_between_result(root, ids[0], from_commit_id=stop)
728 result_ids = {c.commit_id for c in result["commits"]}
729 assert stop not in result_ids
730 assert result["truncated"] is False
731 assert result["count"] == 5 # ids[0]..ids[4]
732
733 def test_no_from_commit_walks_all(self, tmp_path: pathlib.Path) -> None:
734 root = _repo(tmp_path)
735 ids = _build_linear_chain(root, 20)
736 result = walk_commits_between_result(root, ids[0], max_commits=100)
737 assert result["count"] == 20
738 assert result["truncated"] is False
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago