gabriel / muse public
test_integrity_I7_history_walk.py python
733 lines 27.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 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 # Two branches each 110 commits deep from a common ancestor — the BFS
345 # cannot find the merge base within the 100-ancestor cap.
346 tip_a, tip_b, _ = self._make_diverging_branches(root, 110, 110)
347
348 with mock.patch("muse.cli.config.get_limit", return_value=100):
349 with pytest.raises(MuseCLIError, match="Ancestor graph exceeds"):
350 find_merge_base(root, tip_a, tip_b)
351
352 def test_b_side_cap_also_raises_muse_cli_error(
353 self, tmp_path: pathlib.Path
354 ) -> None:
355 """B-side exceeding cap must also raise MuseCLIError — consistent behavior."""
356 from muse.core.errors import MuseCLIError
357 root = _repo(tmp_path)
358
359 # Two branches from a common ancestor deep in the history.
360 # A-side is short (won't hit cap), B-side is long.
361 common_record = _make_commit(root, "deep_common", "common")
362 common_id = common_record.commit_id
363
364 # B-side: 110 commits
365 prev = common_id
366 tip_b = common_id
367 for i in range(110):
368 record = _make_commit(root, f"b_side_{i}", parent=prev, branch="b")
369 prev = record.commit_id
370 tip_b = record.commit_id
371
372 # A-side: 5 commits (tiny — won't hit cap)
373 prev = common_id
374 tip_a = common_id
375 for i in range(5):
376 record = _make_commit(root, f"a_side_{i}", parent=prev, branch="a")
377 prev = record.commit_id
378 tip_a = record.commit_id
379
380 # With cap=100, B-side has 110 commits → raises
381 with mock.patch("muse.cli.config.get_limit", return_value=100):
382 with pytest.raises(MuseCLIError, match="Ancestor graph"):
383 find_merge_base(root, tip_a, tip_b)
384
385 def test_error_message_mentions_config_key(
386 self, tmp_path: pathlib.Path
387 ) -> None:
388 """Error message must tell users how to raise the cap."""
389 from muse.core.errors import MuseCLIError
390 root = _repo(tmp_path)
391 tip_a, tip_b, _ = self._make_diverging_branches(root, 110, 110)
392
393 with mock.patch("muse.cli.config.get_limit", return_value=100):
394 with pytest.raises(MuseCLIError) as exc_info:
395 find_merge_base(root, tip_a, tip_b)
396 assert "max_ancestors" in str(exc_info.value)
397 assert "config.toml" in str(exc_info.value)
398
399 @pytest.mark.slow
400 def test_60k_deep_branches_raise_not_wrong_answer(
401 self, tmp_path: pathlib.Path
402 ) -> None:
403 """Two 60k-deep branches: find_merge_base raises, never silently truncates."""
404 from muse.core.errors import MuseCLIError
405 root = _repo(tmp_path)
406 # Use cap=50k (default); build 52k branches — enough to exceed cap, no excess
407 common_record = _make_commit(root, "root60k", "root")
408 common_id = common_record.commit_id
409
410 n = 52_000
411 prev_a = common_id
412 for i in range(n):
413 record = _make_commit(root, f"a60k_{i}", parent=prev_a, branch="a")
414 prev_a = record.commit_id
415 tip_a = prev_a
416
417 # B-side: only 10 commits — A-side will hit the cap first
418 prev_b = common_id
419 for i in range(10):
420 record = _make_commit(root, f"b10_{i}", parent=prev_b, branch="b")
421 prev_b = record.commit_id
422 tip_b = prev_b
423
424 # find_merge_base must raise, not silently return None/wrong answer
425 with pytest.raises(MuseCLIError, match="Ancestor graph exceeds"):
426 find_merge_base(root, tip_a, tip_b)
427
428
429 # ---------------------------------------------------------------------------
430 # 5. _collect_all_commits — O(n) deque (not O(n²) list.pop(0))
431 # ---------------------------------------------------------------------------
432
433
434 class TestCollectAllCommitsPerformance:
435 def test_uses_deque_not_list_for_bfs(self) -> None:
436 """Confirm _collect_all_commits uses deque internally via source inspection."""
437 from muse.cli.commands import log as log_mod
438 import inspect
439 import ast
440 source = inspect.getsource(log_mod._collect_all_commits)
441 # Parse AST to check code (not docstring) for list.pop(0) pattern
442 tree = ast.parse(source)
443 pop0_calls: list[ast.Call] = []
444 for node in ast.walk(tree):
445 if (
446 isinstance(node, ast.Call)
447 and isinstance(node.func, ast.Attribute)
448 and node.func.attr == "pop"
449 and node.args
450 and isinstance(node.args[0], ast.Constant)
451 and node.args[0].value == 0
452 ):
453 pop0_calls.append(node)
454 assert not pop0_calls, (
455 "Found list.pop(0) in _collect_all_commits — this is the O(n²) bug. "
456 "Replace with deque.popleft()."
457 )
458 assert "deque" in source, (
459 "_collect_all_commits must use deque for O(1) popleft."
460 )
461
462 def test_collect_returns_tuple_with_truncated_flag(
463 self, tmp_path: pathlib.Path
464 ) -> None:
465 root = _repo(tmp_path)
466 ids = _build_linear_chain(root, 20)
467
468 from muse.cli.commands.log import _collect_all_commits
469 commits, truncated = _collect_all_commits(root, [ids[0]], max_commits=100)
470 assert isinstance(commits, dict)
471 assert isinstance(truncated, bool)
472 assert truncated is False
473 assert len(commits) == 20
474
475 def test_collect_truncates_at_cap(self, tmp_path: pathlib.Path) -> None:
476 root = _repo(tmp_path)
477 ids = _build_linear_chain(root, 50)
478
479 from muse.cli.commands.log import _collect_all_commits
480 commits, truncated = _collect_all_commits(root, [ids[0]], max_commits=10)
481 assert truncated is True
482 assert len(commits) == 10
483
484 @pytest.mark.slow
485 def test_10k_commits_completes_in_reasonable_time(
486 self, tmp_path: pathlib.Path
487 ) -> None:
488 """10k BFS must complete in < 5s — proves O(n) not O(n²)."""
489 root = _repo(tmp_path)
490 ids = _build_linear_chain(root, 10_000)
491
492 from muse.cli.commands.log import _collect_all_commits
493 t0 = time.perf_counter()
494 commits, _ = _collect_all_commits(root, [ids[0]], max_commits=100_000)
495 elapsed = time.perf_counter() - t0
496
497 assert len(commits) == 10_000
498 assert elapsed < 5.0, (
499 f"_collect_all_commits took {elapsed:.2f}s for 10k commits. "
500 "Expected < 5s. O(n²) list.pop(0) would take ~50s."
501 )
502
503
504 # ---------------------------------------------------------------------------
505 # 6. muse log --json streaming output with "truncated" field
506 # ---------------------------------------------------------------------------
507
508
509 class TestLogJsonOutput:
510 def _run_log(self, root: pathlib.Path, *extra_args: str) -> _LogOutput:
511 """Run muse log --json via CliRunner and parse the output."""
512 runner = CliRunner()
513 result = runner.invoke(None, ["log", "--json"], env={"MUSE_REPO_ROOT": str(root)})
514 out = result.stdout.strip()
515 if not out:
516 return _LogOutput()
517 parsed: _LogOutput = json.loads(out)
518 return parsed
519
520 def test_log_json_has_truncated_field(self, tmp_path: pathlib.Path) -> None:
521 """muse log --json must include a 'truncated' key in output."""
522 root = _repo(tmp_path)
523 ids = _build_linear_chain(root, 5)
524 (root / ".muse" / "refs" / "heads" / "main").write_text(ids[0])
525
526 output = self._run_log(root)
527 assert "truncated" in output, (
528 "muse log --json must include 'truncated' key. "
529 "Agents rely on this to know whether to page."
530 )
531
532 def test_log_json_not_truncated_for_small_history(
533 self, tmp_path: pathlib.Path
534 ) -> None:
535 root = _repo(tmp_path)
536 ids = _build_linear_chain(root, 5)
537 (root / ".muse" / "refs" / "heads" / "main").write_text(ids[0])
538
539 output = self._run_log(root)
540 assert output["truncated"] is False
541
542 def test_log_json_has_commits_array(self, tmp_path: pathlib.Path) -> None:
543 root = _repo(tmp_path)
544 ids = _build_linear_chain(root, 3)
545 (root / ".muse" / "refs" / "heads" / "main").write_text(ids[0])
546
547 output = self._run_log(root)
548 commits = output.get("commits")
549 assert isinstance(commits, list)
550 assert len(commits) == 3
551
552 def test_log_json_commit_fields(self, tmp_path: pathlib.Path) -> None:
553 """Each commit in JSON output has the required agent-facing fields."""
554 root = _repo(tmp_path)
555 ids = _build_linear_chain(root, 2)
556 (root / ".muse" / "refs" / "heads" / "main").write_text(ids[0])
557
558 output = self._run_log(root)
559 commits_raw = output.get("commits", [])
560 assert isinstance(commits_raw, list)
561 assert len(commits_raw) >= 1
562 # Verify required fields are present in the raw dict
563 first_raw = commits_raw[0]
564 assert isinstance(first_raw, dict)
565 required_fields = {
566 "commit_id", "branch", "message", "author",
567 "committed_at", "parent_commit_id", "snapshot_id",
568 "metadata", "sem_ver_bump",
569 }
570 assert required_fields.issubset(set(first_raw.keys())), (
571 f"Missing fields: {required_fields - set(first_raw.keys())}"
572 )
573
574 def test_log_json_empty_history(self, tmp_path: pathlib.Path) -> None:
575 """Empty history emits a valid JSON response (not an exception)."""
576 root = _repo(tmp_path)
577 # No commits, HEAD is empty — must not crash
578 output = self._run_log(root)
579 # Valid outcomes: empty dict (branch not found) or {"truncated":false,"commits":[]}
580 if output:
581 assert "commits" in output or output == {}
582
583
584 # ---------------------------------------------------------------------------
585 # 7. commit_graph plumbing — truncated in all output formats
586 # ---------------------------------------------------------------------------
587
588
589 class TestCommitGraphTruncation:
590 def _run_commit_graph(
591 self,
592 root: pathlib.Path,
593 tip: str,
594 fmt: str = "json",
595 max_commits: int = 10_000,
596 count_only: bool = False,
597 ) -> str:
598 args = ["commit-graph", "--tip", tip,
599 "--max", str(max_commits), "--format", fmt]
600 if count_only:
601 args.append("--count")
602 runner = CliRunner()
603 result = runner.invoke(None, args, env={"MUSE_REPO_ROOT": str(root)})
604 return result.stdout
605
606 def test_json_has_truncated_false_when_under_cap(
607 self, tmp_path: pathlib.Path
608 ) -> None:
609 root = _repo(tmp_path)
610 ids = _build_linear_chain(root, 10)
611 out = json.loads(self._run_commit_graph(root, ids[0], max_commits=100))
612 assert out["truncated"] is False
613 assert out["count"] == 10
614
615 def test_json_has_truncated_true_when_over_cap(
616 self, tmp_path: pathlib.Path
617 ) -> None:
618 root = _repo(tmp_path)
619 ids = _build_linear_chain(root, 50)
620 out = json.loads(self._run_commit_graph(root, ids[0], max_commits=20))
621 assert out["truncated"] is True
622 assert out["count"] == 20
623
624 def test_text_format_has_truncated_comment_when_over_cap(
625 self, tmp_path: pathlib.Path
626 ) -> None:
627 root = _repo(tmp_path)
628 ids = _build_linear_chain(root, 50)
629 out = self._run_commit_graph(root, ids[0], fmt="text", max_commits=10)
630 assert "TRUNCATED" in out, (
631 "text format must emit '# TRUNCATED' when cap is hit"
632 )
633
634 def test_text_format_no_truncated_when_under_cap(
635 self, tmp_path: pathlib.Path
636 ) -> None:
637 root = _repo(tmp_path)
638 ids = _build_linear_chain(root, 5)
639 out = self._run_commit_graph(root, ids[0], fmt="text", max_commits=100)
640 assert "TRUNCATED" not in out
641
642 def test_count_only_has_truncated_field(
643 self, tmp_path: pathlib.Path
644 ) -> None:
645 root = _repo(tmp_path)
646 ids = _build_linear_chain(root, 50)
647 out = json.loads(self._run_commit_graph(
648 root, ids[0], max_commits=20, count_only=True
649 ))
650 assert "truncated" in out, "count-only must include 'truncated'"
651 assert out["truncated"] is True
652 assert out["count"] == 20
653
654 @pytest.mark.slow
655 def test_15k_chain_commit_graph_completes_in_30s(
656 self, tmp_path: pathlib.Path
657 ) -> None:
658 """commit-graph on 15k commits must complete in < 30s."""
659 root = _repo(tmp_path)
660 ids = _build_linear_chain(root, 15_000)
661
662 t0 = time.perf_counter()
663 out = json.loads(self._run_commit_graph(root, ids[0], max_commits=10_000))
664 elapsed = time.perf_counter() - t0
665
666 assert out["truncated"] is True
667 assert elapsed < 30.0, (
668 f"commit-graph on 15k commits took {elapsed:.1f}s — must be < 30s"
669 )
670
671
672 # ---------------------------------------------------------------------------
673 # 8. Regression: B-side was returning None silently (old bug)
674 # ---------------------------------------------------------------------------
675
676
677 class TestMergeBaseConsistency:
678 def test_a_and_b_cap_raise_same_type(self, tmp_path: pathlib.Path) -> None:
679 """Both A-side and B-side cap must raise the same exception type."""
680 from muse.core.errors import MuseCLIError
681 root = _repo(tmp_path)
682
683 # Build a 30-commit chain
684 ids = _build_linear_chain(root, 30)
685 tip_a = ids[0] # newest
686 tip_b = ids[15] # halfway
687
688 # With cap=20, A-side will be exhausted (30 > 20)
689 with mock.patch("muse.cli.config.get_limit", return_value=20):
690 with pytest.raises(MuseCLIError):
691 find_merge_base(root, tip_a, tip_b)
692
693 def test_symmetric_result_for_small_graph(
694 self, tmp_path: pathlib.Path
695 ) -> None:
696 """find_merge_base(a, b) == find_merge_base(b, a) for a small graph."""
697 root = _repo(tmp_path)
698 common_record = _make_commit(root, "sym_root")
699 common_id = common_record.commit_id
700
701 tip_a_record = _make_commit(root, "sym_a_1", parent=common_id)
702 tip_a = tip_a_record.commit_id
703 tip_b_record = _make_commit(root, "sym_b_1", parent=common_id)
704 tip_b = tip_b_record.commit_id
705
706 ab = find_merge_base(root, tip_a, tip_b)
707 ba = find_merge_base(root, tip_b, tip_a)
708 assert ab == ba == common_id
709
710
711 # ---------------------------------------------------------------------------
712 # 9. walk_commits_between_result — from_commit_id exclusion
713 # ---------------------------------------------------------------------------
714
715
716 class TestWalkFromCommitExclusion:
717 def test_from_commit_excluded(self, tmp_path: pathlib.Path) -> None:
718 """from_commit_id is exclusive — it must not appear in the result."""
719 root = _repo(tmp_path)
720 ids = _build_linear_chain(root, 10) # newest first
721 stop = ids[5] # stop before this one
722 result = walk_commits_between_result(root, ids[0], from_commit_id=stop)
723 result_ids = {c.commit_id for c in result["commits"]}
724 assert stop not in result_ids
725 assert result["truncated"] is False
726 assert result["count"] == 5 # ids[0]..ids[4]
727
728 def test_no_from_commit_walks_all(self, tmp_path: pathlib.Path) -> None:
729 root = _repo(tmp_path)
730 ids = _build_linear_chain(root, 20)
731 result = walk_commits_between_result(root, ids[0], max_commits=100)
732 assert result["count"] == 20
733 assert result["truncated"] is False
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago