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