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