gabriel / muse public
test_log_graph.py python
1,205 lines 51.6 KB
Raw
sha256:7781e508756c81b7ddb0b08b408fd2b99bad87798cefa596773373efc360952c chore: typing audit — zero violations, zero untyped defs Sonnet 4.6 patch 23 days ago
1 """Comprehensive tests for ``muse log --graph`` DAG rendering.
2
3 Coverage taxonomy
4 -----------------
5 Unit / integration
6 Isolated calls to ``_render_graph`` with real on-disk repos built in
7 ``tmp_path``. Each fixture constructs the minimum commit graph needed to
8 exercise one behaviour, then asserts on the captured stdout lines.
9
10 End-to-end
11 Full CLI invocations via ``CliRunner`` to verify the ``--graph`` flag is
12 wired correctly and produces structurally correct output.
13
14 Stress
15 Large or wide graphs (100-commit chains, many concurrent branches, octopus
16 merges) to confirm the lane allocator doesn't corrupt state under load.
17
18 Data integrity
19 Repos with partial object stores, orphaned parent references, or degenerate
20 commits (multiline messages, duplicate parents) to ensure the renderer
21 degrades gracefully.
22
23 Performance
24 Timing assertions that the renderer completes well under a human-perceptible
25 threshold for graphs of the sizes typically encountered in real repos.
26
27 Security
28 Commit messages and branch names containing ANSI escape sequences, control
29 characters, and shell metacharacters must be sanitised before reaching the
30 terminal — the renderer must never pass raw untrusted bytes to print().
31 """
32
33 from __future__ import annotations
34
35 import contextlib
36 import datetime
37 import io
38 import json
39 import pathlib
40 import time
41 from collections.abc import Mapping
42
43 import pytest
44
45 from tests.cli_test_helper import CliRunner
46 from muse.core.types import blob_id, fake_id
47 from muse.core.object_store import write_object
48 from muse.core.paths import heads_dir, muse_dir, ref_path
49 from muse.core.commits import CommitRecord, write_commit
50 from muse.core.snapshots import SnapshotRecord, write_snapshot
51 from muse.core.ids import hash_snapshot, hash_commit
52 from muse.cli.commands.log import _render_graph, _topo_sort, _collect_all_commits
53
54 runner = CliRunner()
55 cli = None
56
57 # ---------------------------------------------------------------------------
58 # Shared repo-building helpers
59 # ---------------------------------------------------------------------------
60
61
62 def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]:
63 """Initialise a minimal Muse repository under *tmp_path*.
64
65 Creates the required ``.muse/`` directory structure, writes ``repo.json``
66 and ``HEAD``, and returns ``(root, repo_id)``. The default branch is
67 ``main``; tests that need additional branches create them by writing ref
68 files under ``.muse/refs/heads/``.
69 """
70 dot_muse = muse_dir(tmp_path)
71 dot_muse.mkdir()
72 repo_id = fake_id("repo")
73 (dot_muse / "repo.json").write_text(
74 json.dumps(
75 {
76 "repo_id": repo_id,
77 "domain": "code",
78 "default_branch": "main",
79 "created_at": "2025-01-01T00:00:00+00:00",
80 }
81 ),
82 encoding="utf-8",
83 )
84 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
85 (dot_muse / "refs" / "heads").mkdir(parents=True)
86 (dot_muse / "snapshots").mkdir()
87 (dot_muse / "commits").mkdir()
88 (dot_muse / "objects").mkdir()
89 return tmp_path, repo_id
90
91
92 _commit_counter = 0 # monotonic counter so timestamps are strictly ordered
93
94
95 def _make_commit(
96 root: pathlib.Path,
97 branch: str,
98 files: dict[str, bytes],
99 parent_id: str | None = None,
100 parent2_id: str | None = None,
101 message: str = "commit",
102 ) -> str:
103 """Write a commit record to *root* and advance the branch ref.
104
105 ``files`` is a mapping of repo-relative path → bytes content. Both blob
106 objects and snapshot records are written to the object store. If
107 ``parent2_id`` is supplied the commit becomes a two-parent merge commit.
108
109 Returns the new commit ID (``sha256:<hex>`` string).
110 """
111 global _commit_counter
112 _commit_counter += 1
113
114 manifest: dict[str, str] = {}
115 for rel, content in files.items():
116 oid = blob_id(content)
117 write_object(root, oid, content)
118 manifest[rel] = oid
119 dest = root / rel
120 dest.parent.mkdir(parents=True, exist_ok=True)
121 dest.write_bytes(content)
122
123 snap_id = hash_snapshot(manifest)
124 # Use a fixed base time offset by counter so every commit has a unique,
125 # strictly-ordered timestamp. This makes _topo_sort deterministic.
126 committed_at = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(seconds=_commit_counter)
127 parent_ids = [p for p in (parent_id, parent2_id) if p]
128 commit_id = hash_commit(
129 parent_ids=parent_ids,
130 snapshot_id=snap_id,
131 message=message,
132 committed_at_iso=committed_at.isoformat(),
133 )
134 write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest))
135 write_commit(
136 root,
137 CommitRecord(
138 commit_id=commit_id,
139 branch=branch,
140 snapshot_id=snap_id,
141 message=message,
142 committed_at=committed_at,
143 parent_commit_id=parent_id,
144 parent2_commit_id=parent2_id,
145 ),
146 )
147 rf = ref_path(root, branch)
148 rf.parent.mkdir(parents=True, exist_ok=True)
149 rf.write_text(commit_id, encoding="utf-8")
150 return commit_id
151
152
153 def _env(root: pathlib.Path) -> Mapping[str, str]:
154 """Return the environment dict needed to target a test repo."""
155 return {"MUSE_REPO_ROOT": str(root)}
156
157
158 def _capture_graph(
159 root: pathlib.Path,
160 branch: str = "main",
161 all_branches: bool = False,
162 ) -> str:
163 """Invoke ``_render_graph`` and return all printed lines as a single string."""
164 buf = io.StringIO()
165 with contextlib.redirect_stdout(buf):
166 _render_graph(root, branch, all_branches=all_branches, tty=False)
167 return buf.getvalue()
168
169
170 def _graph_lines(output: str) -> list[str]:
171 """Return non-empty lines from captured graph output."""
172 return [line for line in output.splitlines() if line.strip()]
173
174
175 def _commit_prefix(line: str) -> str:
176 """Extract the ASCII art prefix from a commit line (the part before sha256:).
177
178 Connector-only lines (``|/``, ``|\\``, etc.) are returned verbatim.
179 """
180 if "sha256:" in line:
181 return line[: line.index("sha256:")].rstrip()
182 return line.rstrip()
183
184
185 def _prefixes(output: str) -> list[str]:
186 """Return the graph-art prefix for every non-empty line in *output*."""
187 return [_commit_prefix(ln) for ln in _graph_lines(output)]
188
189
190 # ---------------------------------------------------------------------------
191 # Unit / integration — linear chain
192 # ---------------------------------------------------------------------------
193
194
195 class TestLinearChain:
196 """A single-branch chain with no merges must show only ``*`` commit rows
197 and ``|`` connector rows between them — no backslashes, no slashes."""
198
199 def test_single_commit_shows_asterisk(self, tmp_path: pathlib.Path) -> None:
200 """A repo with exactly one commit should print a single ``*`` row."""
201 root, repo_id = _init_repo(tmp_path)
202 _make_commit(root, "main", {"a.txt": b"hello"}, message="init")
203
204 output = _capture_graph(root)
205 lines = _graph_lines(output)
206
207 assert len(lines) == 1
208 assert lines[0].startswith("*"), f"Expected '* sha256:...', got: {lines[0]!r}"
209
210 def test_two_commits_connector_is_pipe(self, tmp_path: pathlib.Path) -> None:
211 """Two commits in a linear chain must have a ``|`` connector between them."""
212 root, repo_id = _init_repo(tmp_path)
213 c1 = _make_commit(root, "main", {"a.txt": b"v1"}, message="first")
214 _make_commit(root, "main", {"a.txt": b"v2"}, parent_id=c1, message="second")
215
216 prefixes = _prefixes(_capture_graph(root))
217
218 assert prefixes[0] == "*", f"Top commit prefix: {prefixes[0]!r}"
219 assert prefixes[1] == "|", f"Connector prefix: {prefixes[1]!r}"
220 assert prefixes[2] == "*", f"Bottom commit prefix: {prefixes[2]!r}"
221
222 def test_three_commits_no_backslash_no_slash(self, tmp_path: pathlib.Path) -> None:
223 """A three-commit linear chain must contain no backslash or slash connectors."""
224 root, repo_id = _init_repo(tmp_path)
225 c1 = _make_commit(root, "main", {"a.txt": b"1"}, message="one")
226 c2 = _make_commit(root, "main", {"a.txt": b"2"}, parent_id=c1, message="two")
227 _make_commit(root, "main", {"a.txt": b"3"}, parent_id=c2, message="three")
228
229 output = _capture_graph(root)
230
231 assert "\\" not in output, f"Unexpected \\ in linear output:\n{output}"
232 assert "/" not in output, f"Unexpected / in linear output:\n{output}"
233
234 def test_linear_commit_count_matches(self, tmp_path: pathlib.Path) -> None:
235 """Every commit in the chain must appear exactly once in the graph."""
236 root, repo_id = _init_repo(tmp_path)
237 prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0")
238 for i in range(1, 6):
239 prev = _make_commit(root, "main", {"a.txt": f"v{i}".encode()}, parent_id=prev, message=f"c{i}")
240
241 output = _capture_graph(root)
242 commit_lines = [ln for ln in _graph_lines(output) if "sha256:" in ln]
243
244 assert len(commit_lines) == 6, f"Expected 6 commit lines, got {len(commit_lines)}"
245
246 def test_no_commits_prints_sentinel(self, tmp_path: pathlib.Path) -> None:
247 """A branch with no commits must print ``(no commits)`` rather than crashing."""
248 root, _ = _init_repo(tmp_path)
249
250 output = _capture_graph(root)
251
252 assert "(no commits)" in output, f"Expected '(no commits)', got: {output!r}"
253
254
255 # ---------------------------------------------------------------------------
256 # Unit / integration — merge commit connector
257 # ---------------------------------------------------------------------------
258
259
260 class TestMergeConnector:
261 """After a two-parent merge commit the connector row must be ``|\\``
262 (pipe-backslash) — the fix for Bug 1 which was producing ``|\\|``."""
263
264 def test_merge_connector_is_backslash(self, tmp_path: pathlib.Path) -> None:
265 """``|\\`` must appear immediately after a merge commit row."""
266 root, repo_id = _init_repo(tmp_path)
267 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
268 (heads_dir(root) / "feat").write_text(base)
269 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
270 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
271 _make_commit(root, "main", {"f.txt": b"merge"}, parent_id=ours, parent2_id=feat, message="merge")
272
273 output = _capture_graph(root)
274 lines = _graph_lines(output)
275
276 # The first line is the merge commit; the second must be the connector.
277 assert lines[0].startswith("*"), f"First line should be merge commit: {lines[0]!r}"
278 assert lines[1] == "|\\", (
279 f"Connector after merge must be '|\\\\', got: {lines[1]!r}\nFull output:\n{output}"
280 )
281
282 def test_merge_connector_no_extra_trailing_pipe(self, tmp_path: pathlib.Path) -> None:
283 """The newly-opened extra-parent lane must not produce a trailing ``|``
284 after the backslash — that was the ``|\\|`` bug."""
285 root, repo_id = _init_repo(tmp_path)
286 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
287 (heads_dir(root) / "feat").write_text(base)
288 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
289 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
290 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge")
291
292 output = _capture_graph(root)
293 lines = _graph_lines(output)
294
295 assert "|\\|" not in output, (
296 f"Found |\\\\| (old bug) in output:\n{output}"
297 )
298 assert lines[1] == "|\\", (
299 f"Connector must be exactly '|\\\\', got {lines[1]!r}"
300 )
301
302 def test_merge_connector_has_no_extra_columns(self, tmp_path: pathlib.Path) -> None:
303 """The connector row for a two-parent merge must be exactly 2 chars wide."""
304 root, repo_id = _init_repo(tmp_path)
305 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
306 (heads_dir(root) / "feat").write_text(base)
307 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
308 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
309 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge")
310
311 lines = _graph_lines(_capture_graph(root))
312
313 assert len(lines[1]) == 2, (
314 f"Connector row must be exactly 2 chars ('|\\\\'), got {lines[1]!r}"
315 )
316
317 def test_merge_commit_row_has_asterisk_at_col_zero(self, tmp_path: pathlib.Path) -> None:
318 """The merge commit itself must print ``*`` in column 0."""
319 root, repo_id = _init_repo(tmp_path)
320 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
321 (heads_dir(root) / "feat").write_text(base)
322 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
323 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
324 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge")
325
326 lines = _graph_lines(_capture_graph(root))
327 prefix = _commit_prefix(lines[0])
328
329 assert prefix == "*", f"Merge commit must be at column 0, got: {prefix!r}"
330
331
332 # ---------------------------------------------------------------------------
333 # Unit / integration — convergence connector
334 # ---------------------------------------------------------------------------
335
336
337 class TestConvergenceConnector:
338 """When two lanes both converge on the same ancestor commit the connector
339 row before that commit must show ``|/`` — the fix for Bug 2."""
340
341 def test_convergence_connector_present(self, tmp_path: pathlib.Path) -> None:
342 """``|/`` must appear before the shared ancestor in a diamond graph."""
343 root, repo_id = _init_repo(tmp_path)
344 base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit")
345 (heads_dir(root) / "feat").write_text(base)
346 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit")
347 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit")
348 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit")
349
350 output = _capture_graph(root)
351
352 assert "|/" in output, (
353 f"Expected '|/' convergence connector in output:\n{output}"
354 )
355
356 def test_convergence_connector_before_shared_ancestor(
357 self, tmp_path: pathlib.Path
358 ) -> None:
359 """The ``|/`` connector must appear immediately before the shared-ancestor row."""
360 root, repo_id = _init_repo(tmp_path)
361 base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit")
362 (heads_dir(root) / "feat").write_text(base)
363 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit")
364 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit")
365 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit")
366
367 lines = _graph_lines(_capture_graph(root))
368
369 # Find the |/ line.
370 slash_idx = next((i for i, ln in enumerate(lines) if "|/" in ln), None)
371 assert slash_idx is not None, "No |/ connector found"
372
373 # The row immediately following |/ must be the shared ancestor commit.
374 next_line = lines[slash_idx + 1]
375 assert "sha256:" in next_line and "initial commit" in next_line, (
376 f"Row after |/ should be initial commit, got: {next_line!r}"
377 )
378
379 def test_initial_commit_has_no_trailing_pipe(self, tmp_path: pathlib.Path) -> None:
380 """After convergence the initial commit row must show only ``*``,
381 not ``* |`` with a dangling second column — that was the original bug."""
382 root, repo_id = _init_repo(tmp_path)
383 base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit")
384 (heads_dir(root) / "feat").write_text(base)
385 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit")
386 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit")
387 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit")
388
389 lines = _graph_lines(_capture_graph(root))
390 last_commit_line = next(
391 ln for ln in reversed(lines) if "sha256:" in ln
392 )
393 prefix = _commit_prefix(last_commit_line)
394
395 assert prefix == "*", (
396 f"Shared ancestor must render at col 0 with no trailing '|', "
397 f"got prefix {prefix!r}"
398 )
399
400 def test_no_dangling_pipe_column_after_convergence(
401 self, tmp_path: pathlib.Path
402 ) -> None:
403 """No ``| |`` connector should appear after ``|/`` has closed a lane."""
404 root, repo_id = _init_repo(tmp_path)
405 base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit")
406 (heads_dir(root) / "feat").write_text(base)
407 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit")
408 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit")
409 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit")
410
411 output = _capture_graph(root)
412 lines = _graph_lines(output)
413
414 slash_idx = next((i for i, ln in enumerate(lines) if "|/" in ln), None)
415 assert slash_idx is not None
416
417 for ln in lines[slash_idx + 1 :]:
418 assert "| |" not in ln, (
419 f"Found '| |' connector after convergence (stale lane):\n{output}"
420 )
421
422
423 # ---------------------------------------------------------------------------
424 # Unit / integration — full diamond pattern
425 # ---------------------------------------------------------------------------
426
427
428 class TestDiamondPattern:
429 """A canonical diamond: merge → (left, right) → base. Validates the
430 complete shape of the rendered graph end-to-end."""
431
432 def _build_diamond(
433 self, tmp_path: pathlib.Path
434 ) -> tuple[pathlib.Path, str, str, str, str]:
435 """Build a four-commit diamond and return (root, base, left, right, merge)."""
436 root, repo_id = _init_repo(tmp_path)
437 base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit")
438 (heads_dir(root) / "feat").write_text(base)
439 left = _make_commit(root, "main", {"f.txt": b"left"}, parent_id=base, message="left commit")
440 right = _make_commit(root, "feat", {"f.txt": b"right"}, parent_id=base, message="right commit")
441 merge = _make_commit(
442 root, "main", {"f.txt": b"merged"}, parent_id=left, parent2_id=right, message="merge commit"
443 )
444 return root, base, left, right, merge
445
446 def test_diamond_line_count(self, tmp_path: pathlib.Path) -> None:
447 """A diamond graph must produce exactly the right number of output lines:
448 4 commit rows + 1 merge connector + inter-commit connectors + 1 slash."""
449 root, *_ = self._build_diamond(tmp_path)
450 lines = _graph_lines(_capture_graph(root))
451
452 commit_rows = [ln for ln in lines if "sha256:" in ln]
453 assert len(commit_rows) == 4, f"Expected 4 commits, got {len(commit_rows)}: {commit_rows}"
454
455 def test_diamond_contains_backslash(self, tmp_path: pathlib.Path) -> None:
456 """The merge connector ``|\\`` must appear exactly once."""
457 root, *_ = self._build_diamond(tmp_path)
458 output = _capture_graph(root)
459 lines = _graph_lines(output)
460
461 backslash_lines = [ln for ln in lines if ln == "|\\"]
462 assert len(backslash_lines) == 1, (
463 f"Expected exactly one '|\\\\' connector, got: {backslash_lines}\n{output}"
464 )
465
466 def test_diamond_contains_slash(self, tmp_path: pathlib.Path) -> None:
467 """The convergence connector ``|/`` must appear exactly once."""
468 root, *_ = self._build_diamond(tmp_path)
469 output = _capture_graph(root)
470 lines = _graph_lines(output)
471
472 slash_lines = [ln for ln in lines if "|/" in ln]
473 assert len(slash_lines) == 1, (
474 f"Expected exactly one '|/' connector, got: {slash_lines}\n{output}"
475 )
476
477 def test_diamond_first_line_is_merge(self, tmp_path: pathlib.Path) -> None:
478 """The topologically-first line must be the merge commit."""
479 root, *_, merge_id = self._build_diamond(tmp_path)
480 merge_short = merge_id[7:19] # strip 'sha256:' + take 12 hex chars
481
482 lines = _graph_lines(_capture_graph(root))
483 assert merge_short in lines[0], (
484 f"First line must be merge commit (sha {merge_short}), got: {lines[0]!r}"
485 )
486
487 def test_diamond_last_line_is_base(self, tmp_path: pathlib.Path) -> None:
488 """The last commit row must be the shared ancestor (base)."""
489 root, base_id, *_ = self._build_diamond(tmp_path)
490 base_short = base_id[7:19]
491
492 lines = _graph_lines(_capture_graph(root))
493 commit_lines = [ln for ln in lines if "sha256:" in ln]
494 assert base_short in commit_lines[-1], (
495 f"Last commit must be base ({base_short}), got: {commit_lines[-1]!r}"
496 )
497
498 def test_diamond_no_wide_pipe_after_slash(self, tmp_path: pathlib.Path) -> None:
499 """After the ``|/`` convergence connector there must be no two-column
500 connector rows — the second lane has been closed."""
501 root, *_ = self._build_diamond(tmp_path)
502 output = _capture_graph(root)
503 lines = _graph_lines(output)
504
505 slash_idx = next(i for i, ln in enumerate(lines) if "|/" in ln)
506 for ln in lines[slash_idx + 1 :]:
507 assert "| |" not in ln, (
508 f"Stale two-column connector found after convergence:\n{output}"
509 )
510
511
512 # ---------------------------------------------------------------------------
513 # Unit / integration — decorations
514 # ---------------------------------------------------------------------------
515
516
517 class TestDecorations:
518 """HEAD and branch tip labels must appear on the correct commit rows."""
519
520 def test_head_decoration_on_tip(self, tmp_path: pathlib.Path) -> None:
521 """The current HEAD commit must carry the ``HEAD -> main`` decoration."""
522 root, _ = _init_repo(tmp_path)
523 _make_commit(root, "main", {"a.txt": b"x"}, message="init")
524
525 output = _capture_graph(root)
526
527 assert "HEAD" in output, f"HEAD label missing from graph:\n{output}"
528 assert "main" in output, f"Branch name missing from graph:\n{output}"
529
530 def test_feature_branch_tip_decorated(self, tmp_path: pathlib.Path) -> None:
531 """A feature branch tip commit must show its branch name in the decoration."""
532 root, _ = _init_repo(tmp_path)
533 base = _make_commit(root, "main", {"a.txt": b"base"}, message="base")
534 (heads_dir(root) / "feat").write_text(base)
535 _make_commit(root, "feat", {"a.txt": b"feat"}, parent_id=base, message="feat work")
536 _make_commit(root, "main", {"a.txt": b"ours"}, parent_id=base, message="main work")
537
538 output = _capture_graph(root, all_branches=True)
539
540 assert "feat" in output, f"Feature branch name not in graph:\n{output}"
541
542 def test_message_first_line_only(self, tmp_path: pathlib.Path) -> None:
543 """Only the first line of a multiline commit message must appear in the graph."""
544 root, _ = _init_repo(tmp_path)
545 _make_commit(root, "main", {"a.txt": b"x"}, message="first line\nsecond line\nthird line")
546
547 output = _capture_graph(root)
548
549 assert "first line" in output
550 assert "second line" not in output
551 assert "third line" not in output
552
553
554 # ---------------------------------------------------------------------------
555 # Unit / integration — lane reuse
556 # ---------------------------------------------------------------------------
557
558
559 class TestLaneReuse:
560 """Closed lanes (``None`` slots) must be reused for new branches instead of
561 ever-growing the lane array width."""
562
563 def test_lane_slot_reused_after_close(self, tmp_path: pathlib.Path) -> None:
564 """After a branch terminates (root commit, no parents) its column slot must
565 be reused for the next new branch, keeping the graph compact."""
566 root, _ = _init_repo(tmp_path)
567
568 # Build two independent branches that share no common ancestor.
569 # Branch A: a1 (root, no parent) → a2
570 # Branch B: b1 (root, no parent) → b2 (after a1 is done)
571 a1 = _make_commit(root, "main", {"a.txt": b"a1"}, message="a1 root")
572 a2 = _make_commit(root, "main", {"a.txt": b"a2"}, parent_id=a1, message="a2")
573
574 (heads_dir(root) / "branchB").write_text("") # placeholder
575 b1 = _make_commit(root, "branchB", {"b.txt": b"b1"}, message="b1 root")
576 b2 = _make_commit(root, "branchB", {"b.txt": b"b2"}, parent_id=b1, message="b2")
577
578 output = _capture_graph(root, all_branches=True)
579
580 # Ensure both branches rendered without crash.
581 assert "a1 root" in output or "a2" in output
582 assert "b1 root" in output or "b2" in output
583
584
585 # ---------------------------------------------------------------------------
586 # End-to-end — CLI
587 # ---------------------------------------------------------------------------
588
589
590 class TestCliEndToEnd:
591 """Full CLI invocations through ``CliRunner`` — verifies ``--graph`` is
592 wired correctly and the output meets structural expectations."""
593
594 def _simple_linear_repo(self, tmp_path: pathlib.Path) -> pathlib.Path:
595 root, _ = _init_repo(tmp_path)
596 c1 = _make_commit(root, "main", {"a.txt": b"v1"}, message="first")
597 _make_commit(root, "main", {"a.txt": b"v2"}, parent_id=c1, message="second")
598 return root
599
600 def test_graph_flag_exits_zero(self, tmp_path: pathlib.Path) -> None:
601 """``muse log --graph`` must exit with code 0 on a valid repo."""
602 root = self._simple_linear_repo(tmp_path)
603 result = runner.invoke(cli, ["log", "--graph"], env=_env(root))
604
605 assert result.exit_code == 0, (
606 f"Expected exit 0, got {result.exit_code}:\n{result.stderr}"
607 )
608
609 def test_graph_flag_contains_asterisk(self, tmp_path: pathlib.Path) -> None:
610 """``muse log --graph`` output must contain at least one ``*`` commit row."""
611 root = self._simple_linear_repo(tmp_path)
612 result = runner.invoke(cli, ["log", "--graph"], env=_env(root))
613
614 assert "*" in result.stdout, f"No * in graph output:\n{result.stdout}"
615
616 def test_graph_flag_contains_sha256(self, tmp_path: pathlib.Path) -> None:
617 """Every commit row must contain its content-addressed ID."""
618 root = self._simple_linear_repo(tmp_path)
619 result = runner.invoke(cli, ["log", "--graph"], env=_env(root))
620
621 assert "sha256:" in result.stdout, f"No sha256 in output:\n{result.stdout}"
622
623 def test_cli_merge_graph_contains_backslash(self, tmp_path: pathlib.Path) -> None:
624 """``muse log --graph`` on a repo with a merge must contain ``|\\``."""
625 root, _ = _init_repo(tmp_path)
626 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
627 (heads_dir(root) / "feat").write_text(base)
628 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
629 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
630 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge")
631
632 result = runner.invoke(cli, ["log", "--graph"], env=_env(root))
633
634 assert "|\\" in result.stdout, (
635 f"No '|\\\\' connector in CLI graph output:\n{result.stdout}"
636 )
637
638 def test_cli_diamond_graph_contains_slash(self, tmp_path: pathlib.Path) -> None:
639 """``muse log --graph`` on a diamond must contain ``|/``."""
640 root, _ = _init_repo(tmp_path)
641 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
642 (heads_dir(root) / "feat").write_text(base)
643 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
644 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
645 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge")
646
647 result = runner.invoke(cli, ["log", "--graph"], env=_env(root))
648
649 assert "|/" in result.stdout, (
650 f"No '|/' convergence in CLI graph output:\n{result.stdout}"
651 )
652
653 def test_cli_no_commits_graceful(self, tmp_path: pathlib.Path) -> None:
654 """``muse log --graph`` on an empty repo must exit 0 and say '(no commits)'."""
655 root, _ = _init_repo(tmp_path)
656 result = runner.invoke(cli, ["log", "--graph"], env=_env(root))
657
658 assert result.exit_code == 0
659 assert "no commits" in result.stdout.lower(), (
660 f"Expected '(no commits)' for empty repo:\n{result.stdout}"
661 )
662
663 def test_cli_oneline_not_graph(self, tmp_path: pathlib.Path) -> None:
664 """``muse log --oneline`` (no ``--graph``) must not include ASCII art."""
665 root, _ = _init_repo(tmp_path)
666 _make_commit(root, "main", {"a.txt": b"x"}, message="init")
667
668 result = runner.invoke(cli, ["log", "--oneline"], env=_env(root))
669
670 assert "|" not in result.stdout, (
671 f"--oneline must not show graph art:\n{result.stdout}"
672 )
673
674
675 # ---------------------------------------------------------------------------
676 # Stress
677 # ---------------------------------------------------------------------------
678
679
680 class TestStress:
681 """Large graphs and wide branching patterns must not crash, corrupt lane
682 state, or produce malformed ASCII art."""
683
684 def test_long_linear_chain_100_commits(self, tmp_path: pathlib.Path) -> None:
685 """A 100-commit linear chain must render all 100 commits without error."""
686 root, _ = _init_repo(tmp_path)
687 prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0")
688 for i in range(1, 100):
689 prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}")
690
691 output = _capture_graph(root)
692 commit_rows = [ln for ln in _graph_lines(output) if "sha256:" in ln]
693
694 assert len(commit_rows) == 100, (
695 f"Expected 100 commit rows, got {len(commit_rows)}"
696 )
697
698 def test_long_chain_only_asterisk_and_pipe(self, tmp_path: pathlib.Path) -> None:
699 """In a 50-commit linear chain, every commit prefix must be ``*`` and
700 every connector must be ``|`` — no slash, no backslash."""
701 root, _ = _init_repo(tmp_path)
702 prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0")
703 for i in range(1, 50):
704 prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}")
705
706 output = _capture_graph(root)
707 prefixes = _prefixes(output)
708
709 for p in prefixes:
710 assert p in ("*", "|"), f"Unexpected graph prefix {p!r} in linear chain"
711
712 def test_sequential_merges_no_crash(self, tmp_path: pathlib.Path) -> None:
713 """Five sequential feature branches, each merged back to main, must
714 render without crashing and without any corrupted lane state."""
715 root, _ = _init_repo(tmp_path)
716 tip = _make_commit(root, "main", {"a.txt": b"init"}, message="init")
717
718 for i in range(5):
719 feat_name = f"feat{i}"
720 (heads_dir(root) / feat_name).write_text(tip)
721 feat_tip = _make_commit(
722 root, feat_name, {"a.txt": f"feat{i}".encode()},
723 parent_id=tip, message=f"feat{i} work"
724 )
725 tip = _make_commit(
726 root, "main", {"a.txt": f"merge{i}".encode()},
727 parent_id=tip, parent2_id=feat_tip, message=f"merge feat{i}"
728 )
729
730 output = _capture_graph(root)
731
732 assert output.strip(), "Graph output must not be empty"
733 # No consecutive identical pipes — that would indicate a runaway lane.
734 assert "| | | | | |" not in output, (
735 f"Suspiciously many concurrent lanes:\n{output}"
736 )
737
738 def test_wide_fan_three_branches(self, tmp_path: pathlib.Path) -> None:
739 """Three concurrent branches (A, B, C) merged together must produce a
740 graph with exactly three ``*`` rows plus the merge, and both ``|\\`` and
741 ``|/`` connectors."""
742 root, _ = _init_repo(tmp_path)
743 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
744
745 (heads_dir(root) / "branchA").write_text(base)
746 (heads_dir(root) / "branchB").write_text(base)
747 a = _make_commit(root, "branchA", {"f.txt": b"A"}, parent_id=base, message="branch A")
748 b = _make_commit(root, "branchB", {"f.txt": b"B"}, parent_id=base, message="branch B")
749 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="main work")
750 # Merge A into main first, then merge B.
751 m1 = _make_commit(root, "main", {"f.txt": b"m1"}, parent_id=ours, parent2_id=a, message="merge A")
752 _make_commit(root, "main", {"f.txt": b"m2"}, parent_id=m1, parent2_id=b, message="merge B")
753
754 output = _capture_graph(root)
755
756 assert "\\" in output, f"Expected \\\\ in wide fan output:\n{output}"
757 assert "/" in output, f"Expected / in wide fan output:\n{output}"
758
759 def test_graph_truncation_prints_warning(self, tmp_path: pathlib.Path) -> None:
760 """When the commit count exceeds the configured cap a truncation warning
761 must appear before the graph rows."""
762 root, _ = _init_repo(tmp_path)
763 prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0")
764 for i in range(1, 12):
765 prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}")
766
767 # Write a very low cap so 12 commits triggers truncation.
768 config_path = muse_dir(root) / "config.toml"
769 config_path.write_text("[limits]\nmax_graph_commits = 5\n", encoding="utf-8")
770
771 output = _capture_graph(root)
772
773 assert "truncated" in output.lower(), (
774 f"Expected truncation warning with low cap:\n{output}"
775 )
776
777
778 # ---------------------------------------------------------------------------
779 # Data integrity
780 # ---------------------------------------------------------------------------
781
782
783 class TestDataIntegrity:
784 """Degenerate or partially corrupt repos must be handled gracefully — the
785 renderer must never crash with an unhandled exception."""
786
787 def test_orphaned_parent_ref_skipped(self, tmp_path: pathlib.Path) -> None:
788 """A commit whose parent_commit_id does not exist in the object store
789 must still render without crashing; the orphaned parent is silently
790 excluded from the graph.
791
792 This test bypasses ``write_commit``'s parent-existence guard to
793 deliberately inject a commit that references a non-existent parent —
794 simulating object-store corruption or a partial fetch."""
795 import json as _json
796 from muse.core.commits import commit_path as _commit_path
797
798 root, _ = _init_repo(tmp_path)
799
800 # Write a well-formed commit first so we have a real tip commit.
801 real_cid = _make_commit(root, "main", {"f.txt": b"real"}, message="real commit")
802
803 # Now inject a second commit that references a nonexistent parent by
804 # writing the JSON directly — bypassing write_commit's validation.
805 fake_parent = "sha256:" + "a" * 64
806 orphan_cid = "sha256:" + "b" * 64
807 orphan_data = {
808 "commit_id": orphan_cid,
809 "branch": "main",
810 "snapshot_id": "sha256:" + "0" * 64,
811 "message": "orphan commit",
812 "committed_at": "2025-01-01T00:00:01+00:00",
813 "parent_commit_id": fake_parent,
814 "parent2_commit_id": None,
815 "author": "gabriel",
816 "agent_id": "",
817 "model_id": "",
818 "signature": None,
819 "signer_public_key": None,
820 "sem_ver_bump": "patch",
821 "breaking_changes": [],
822 "metadata": {},
823 "structured_delta": None,
824 }
825 p = _commit_path(root, orphan_cid)
826 p.parent.mkdir(parents=True, exist_ok=True)
827 p.write_text(_json.dumps(orphan_data), encoding="utf-8")
828
829 # The renderer uses BFS from the branch tip (real_cid) — the orphan
830 # commit is not reachable from the tip so it is never visited.
831 # The fake parent is not in the store so BFS stops there gracefully.
832 output = _capture_graph(root)
833
834 assert "sha256:" in output, f"Commit row missing:\n{output}"
835
836 def test_root_commit_closes_its_lane(self, tmp_path: pathlib.Path) -> None:
837 """A root commit (no parents) must close its lane column — subsequent
838 renders must not leave a dangling pipe for a terminated branch."""
839 root, _ = _init_repo(tmp_path)
840 _make_commit(root, "main", {"a.txt": b"root"}, message="root commit")
841
842 lines = _graph_lines(_capture_graph(root))
843
844 # Only one line — the commit itself. No connector should follow.
845 commit_lines = [ln for ln in lines if "sha256:" in ln]
846 connector_lines = [ln for ln in lines if "sha256:" not in ln]
847
848 assert len(commit_lines) == 1
849 assert len(connector_lines) == 0, (
850 f"Root commit must not produce a trailing connector: {connector_lines}"
851 )
852
853 def test_empty_branch_ref_falls_back_to_no_commits(
854 self, tmp_path: pathlib.Path
855 ) -> None:
856 """If the branch ref file exists but is empty, the renderer must print
857 '(no commits)' and not crash."""
858 root, _ = _init_repo(tmp_path)
859 # Write a blank ref — simulates a branch that was created but never committed.
860 ref_path(root, "main").parent.mkdir(parents=True, exist_ok=True)
861 ref_path(root, "main").write_text("", encoding="utf-8")
862
863 output = _capture_graph(root)
864
865 assert "no commits" in output.lower(), (
866 f"Expected '(no commits)' for blank ref, got:\n{output}"
867 )
868
869 def test_multiline_commit_message_renders_first_line_only(
870 self, tmp_path: pathlib.Path
871 ) -> None:
872 """A commit with a multiline message must expose only the first line
873 in the graph — downstream parsers rely on one-line graph output."""
874 root, _ = _init_repo(tmp_path)
875 _make_commit(
876 root, "main", {"a.txt": b"x"},
877 message="Summary line\n\nDetailed paragraph.\nMore detail."
878 )
879
880 output = _capture_graph(root)
881
882 assert "Summary line" in output
883 assert "Detailed paragraph" not in output
884
885 def test_commit_with_no_files_renders(self, tmp_path: pathlib.Path) -> None:
886 """An empty-manifest commit (no files) must render without crashing."""
887 root, _ = _init_repo(tmp_path)
888 _make_commit(root, "main", {}, message="empty manifest commit")
889
890 output = _capture_graph(root)
891
892 assert "empty manifest commit" in output
893
894 def test_graph_output_has_no_blank_prefix_on_commit_rows(
895 self, tmp_path: pathlib.Path
896 ) -> None:
897 """No commit row in the graph may have a blank graph prefix — every
898 commit must be preceded by at least a ``*``."""
899 root, _ = _init_repo(tmp_path)
900 prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0")
901 for i in range(1, 5):
902 prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}")
903
904 lines = _graph_lines(_capture_graph(root))
905 for ln in lines:
906 if "sha256:" in ln:
907 prefix = _commit_prefix(ln)
908 assert prefix, f"Commit row has empty graph prefix: {ln!r}"
909 assert "*" in prefix, f"Commit row prefix missing '*': {ln!r}"
910
911
912 # ---------------------------------------------------------------------------
913 # Performance
914 # ---------------------------------------------------------------------------
915
916
917 class TestPerformance:
918 """The renderer must complete within a human-imperceptible threshold for
919 graph sizes typical of real development repos."""
920
921 def test_hundred_commits_renders_under_two_seconds(
922 self, tmp_path: pathlib.Path
923 ) -> None:
924 """100 linear commits must render in under 2 seconds on any reasonable
925 CI or developer machine."""
926 root, _ = _init_repo(tmp_path)
927 prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0")
928 for i in range(1, 100):
929 prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}")
930
931 start = time.perf_counter()
932 output = _capture_graph(root)
933 elapsed = time.perf_counter() - start
934
935 assert elapsed < 2.0, (
936 f"Graph rendering took {elapsed:.2f}s for 100 commits — too slow"
937 )
938 assert "sha256:" in output # sanity: output was actually produced
939
940 def test_diamond_renders_under_one_second(self, tmp_path: pathlib.Path) -> None:
941 """A four-commit diamond graph (typical merge scenario) must render in
942 under 1 second — this path is hit constantly during normal dev workflows."""
943 root, _ = _init_repo(tmp_path)
944 base = _make_commit(root, "main", {"f.txt": b"base"}, message="base")
945 (heads_dir(root) / "feat").write_text(base)
946 feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat")
947 ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours")
948 _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge")
949
950 start = time.perf_counter()
951 _capture_graph(root)
952 elapsed = time.perf_counter() - start
953
954 assert elapsed < 1.0, (
955 f"Diamond graph rendering took {elapsed:.2f}s — expected < 1s"
956 )
957
958 def test_fifty_sequential_merges_renders_under_five_seconds(
959 self, tmp_path: pathlib.Path
960 ) -> None:
961 """A repo with 50 sequential feature-branch merges (100 total commits)
962 must render in under 5 seconds."""
963 root, _ = _init_repo(tmp_path)
964 tip = _make_commit(root, "main", {"a.txt": b"init"}, message="init")
965
966 for i in range(25):
967 feat_name = f"stress{i}"
968 (heads_dir(root) / feat_name).write_text(tip)
969 feat_tip = _make_commit(
970 root, feat_name, {"a.txt": f"f{i}".encode()},
971 parent_id=tip, message=f"feat {i}"
972 )
973 tip = _make_commit(
974 root, "main", {"a.txt": f"m{i}".encode()},
975 parent_id=tip, parent2_id=feat_tip, message=f"merge {i}"
976 )
977
978 start = time.perf_counter()
979 output = _capture_graph(root)
980 elapsed = time.perf_counter() - start
981
982 assert elapsed < 5.0, (
983 f"50-merge graph took {elapsed:.2f}s — expected < 5s"
984 )
985 assert "sha256:" in output
986
987
988 # ---------------------------------------------------------------------------
989 # Security
990 # ---------------------------------------------------------------------------
991
992
993 class TestSecurity:
994 """The graph renderer must never pass raw untrusted commit data directly to
995 the terminal. ANSI escape sequences, control characters, and shell
996 metacharacters embedded in commit messages or branch names must be
997 neutralised before output.
998
999 All assertions are against the captured string — ``_render_graph`` routes
1000 through ``sanitize_display`` from ``muse.core.validation``, which strips
1001 or replaces dangerous characters before they reach the tty."""
1002
1003 def test_ansi_escape_in_commit_message_neutralised(
1004 self, tmp_path: pathlib.Path
1005 ) -> None:
1006 """A commit message containing an ANSI escape sequence must not pass
1007 the raw escape bytes through to the printed output.
1008
1009 An attacker could embed ``\\x1b[2J`` (clear screen) or a colour-change
1010 sequence in a commit message to hijack the terminal display of anyone
1011 who runs ``muse log --graph``."""
1012 root, _ = _init_repo(tmp_path)
1013 evil_message = "normal prefix \x1b[31mRED INJECTION\x1b[0m suffix"
1014 _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message)
1015
1016 output = _capture_graph(root)
1017
1018 assert "\x1b[31m" not in output, (
1019 "Raw ANSI colour escape must not reach graph output"
1020 )
1021 assert "\x1b[0m" not in output, (
1022 "Raw ANSI reset escape must not reach graph output"
1023 )
1024
1025 def test_null_byte_in_commit_message_neutralised(
1026 self, tmp_path: pathlib.Path
1027 ) -> None:
1028 """A commit message containing null bytes must not produce a broken or
1029 truncated graph line.
1030
1031 Null bytes in terminal output can cause visual corruption and in some
1032 terminal emulators trigger unsafe behaviour."""
1033 root, _ = _init_repo(tmp_path)
1034 evil_message = "before\x00after"
1035 _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message)
1036
1037 output = _capture_graph(root)
1038
1039 assert "\x00" not in output, "Null byte must not appear in graph output"
1040 # The graph must still render something for this commit.
1041 assert "sha256:" in output
1042
1043 def test_carriage_return_in_message_neutralised(
1044 self, tmp_path: pathlib.Path
1045 ) -> None:
1046 """A ``\\r`` (carriage return) in a commit message can overwrite the
1047 beginning of the line in a terminal, causing the graph art prefix to
1048 be visually erased. It must not pass through."""
1049 root, _ = _init_repo(tmp_path)
1050 evil_message = "legit text\r<injected overwrite>"
1051 _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message)
1052
1053 output = _capture_graph(root)
1054
1055 assert "\r" not in output, "Carriage return must not appear in graph output"
1056
1057 def test_terminal_title_escape_in_message_neutralised(
1058 self, tmp_path: pathlib.Path
1059 ) -> None:
1060 """The OSC sequence ``\\x1b]0;title\\x07`` sets the terminal window title.
1061 Embedding this in a commit message could silently rename a user's terminal
1062 window — it must be stripped."""
1063 root, _ = _init_repo(tmp_path)
1064 evil_message = "\x1b]0;malicious title\x07normal text"
1065 _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message)
1066
1067 output = _capture_graph(root)
1068
1069 assert "\x1b]" not in output, "OSC escape sequence must not reach graph output"
1070
1071 def test_shell_metacharacters_in_message_pass_through_safe(
1072 self, tmp_path: pathlib.Path
1073 ) -> None:
1074 """Shell metacharacters in commit messages (``$``, backtick, ``!``)
1075 are benign in Python ``print()`` output — they must not be over-escaped
1076 either, since aggressive escaping corrupts legitimate message text.
1077
1078 This test documents the expected behaviour: shell-special chars are
1079 preserved as-is because they only have significance when interpreted
1080 by a shell, not when printed to stdout."""
1081 root, _ = _init_repo(tmp_path)
1082 message = "fix: avoid $PATH collision & handle `nul` correctly; cost=O(1)"
1083 _make_commit(root, "main", {"a.txt": b"x"}, message=message)
1084
1085 output = _capture_graph(root)
1086
1087 assert "$PATH" in output, "Dollar sign in commit message must not be mangled"
1088 assert "O(1)" in output, "Parentheses in commit message must not be mangled"
1089
1090 def test_very_long_commit_message_does_not_break_line_structure(
1091 self, tmp_path: pathlib.Path
1092 ) -> None:
1093 """A 4 000-character commit message must not cause the graph prefix to
1094 wrap onto multiple lines in the captured output (output goes to a
1095 non-tty StringIO, so no terminal wrapping occurs)."""
1096 root, _ = _init_repo(tmp_path)
1097 long_msg = "A" * 4000
1098 _make_commit(root, "main", {"a.txt": b"x"}, message=long_msg)
1099
1100 output = _capture_graph(root)
1101 commit_lines = [ln for ln in _graph_lines(output) if "sha256:" in ln]
1102
1103 assert len(commit_lines) == 1, (
1104 f"Long message must not produce multiple commit rows, got:\n{output[:500]}"
1105 )
1106 prefix = _commit_prefix(commit_lines[0])
1107 assert "*" in prefix, f"Long-message commit must still show '*' prefix"
1108
1109 def test_unicode_in_commit_message_renders_safely(
1110 self, tmp_path: pathlib.Path
1111 ) -> None:
1112 """Emoji and non-ASCII Unicode in commit messages must render without
1113 raising ``UnicodeEncodeError`` or corrupting the graph prefix."""
1114 root, _ = _init_repo(tmp_path)
1115 _make_commit(
1116 root, "main", {"a.txt": b"x"},
1117 message="feat: 🎵 add MIDI export — café résumé naïve"
1118 )
1119
1120 # Must not raise.
1121 output = _capture_graph(root)
1122
1123 assert "sha256:" in output, f"Unicode commit did not render:\n{output}"
1124
1125
1126 # ---------------------------------------------------------------------------
1127 # _topo_sort unit tests
1128 # ---------------------------------------------------------------------------
1129
1130
1131 class TestTopoSort:
1132 """Unit tests for the ``_topo_sort`` helper in isolation from I/O.
1133
1134 ``_topo_sort`` must return all commits, children before parents, with ties
1135 broken by timestamp (most recent first)."""
1136
1137 def _make_record(
1138 self,
1139 cid: str,
1140 parent: str | None = None,
1141 parent2: str | None = None,
1142 ts_offset: int = 0,
1143 ) -> CommitRecord:
1144 return CommitRecord(
1145 commit_id=cid,
1146 branch="main",
1147 snapshot_id="sha256:" + "0" * 64,
1148 message="msg",
1149 committed_at=datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc)
1150 + datetime.timedelta(seconds=ts_offset),
1151 parent_commit_id=parent,
1152 parent2_commit_id=parent2,
1153 )
1154
1155 def test_single_commit_returned(self) -> None:
1156 """A single commit must be returned as a one-element list."""
1157 c = self._make_record("sha256:" + "a" * 64)
1158 result = _topo_sort({"sha256:" + "a" * 64: c})
1159 assert len(result) == 1
1160 assert result[0].commit_id == "sha256:" + "a" * 64
1161
1162 def test_parent_comes_after_child(self) -> None:
1163 """In a two-commit chain child must appear before parent."""
1164 parent_id = "sha256:" + "p" * 64
1165 child_id = "sha256:" + "c" * 64
1166 parent = self._make_record(parent_id, ts_offset=0)
1167 child = self._make_record(child_id, parent=parent_id, ts_offset=1)
1168 result = _topo_sort({parent_id: parent, child_id: child})
1169
1170 assert result[0].commit_id == child_id
1171 assert result[1].commit_id == parent_id
1172
1173 def test_merge_commit_before_both_parents(self) -> None:
1174 """A merge commit must appear before both of its parents."""
1175 base_id = "sha256:" + "b" * 64
1176 left_id = "sha256:" + "l" * 64
1177 right_id = "sha256:" + "r" * 64
1178 merge_id = "sha256:" + "m" * 64
1179
1180 base = self._make_record(base_id, ts_offset=0)
1181 left = self._make_record(left_id, parent=base_id, ts_offset=1)
1182 right = self._make_record(right_id, parent=base_id, ts_offset=2)
1183 merge = self._make_record(merge_id, parent=left_id, parent2=right_id, ts_offset=3)
1184
1185 commits = {base_id: base, left_id: left, right_id: right, merge_id: merge}
1186 result = _topo_sort(commits)
1187 ids = [r.commit_id for r in result]
1188
1189 assert ids.index(merge_id) < ids.index(left_id)
1190 assert ids.index(merge_id) < ids.index(right_id)
1191 assert ids.index(left_id) < ids.index(base_id)
1192 assert ids.index(right_id) < ids.index(base_id)
1193
1194 def test_all_commits_present_in_result(self) -> None:
1195 """``_topo_sort`` must return every commit exactly once."""
1196 ids = ["sha256:" + c * 64 for c in "abcde"]
1197 records = {}
1198 for i, cid in enumerate(ids):
1199 parent = ids[i - 1] if i > 0 else None
1200 records[cid] = self._make_record(cid, parent=parent, ts_offset=i)
1201
1202 result = _topo_sort(records)
1203
1204 assert len(result) == 5
1205 assert {r.commit_id for r in result} == set(ids)
File History 2 commits
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:09656d1b0772ea4c96f8911d7bf8042b33eb0596992c6546dfab3d21e9dee330 fix: align muse read --json schema and test contracts Sonnet 4.6 minor ⚠ 23 days ago