"""Comprehensive tests for ``muse log --graph`` DAG rendering. Coverage taxonomy ----------------- Unit / integration Isolated calls to ``_render_graph`` with real on-disk repos built in ``tmp_path``. Each fixture constructs the minimum commit graph needed to exercise one behaviour, then asserts on the captured stdout lines. End-to-end Full CLI invocations via ``CliRunner`` to verify the ``--graph`` flag is wired correctly and produces structurally correct output. Stress Large or wide graphs (100-commit chains, many concurrent branches, octopus merges) to confirm the lane allocator doesn't corrupt state under load. Data integrity Repos with partial object stores, orphaned parent references, or degenerate commits (multiline messages, duplicate parents) to ensure the renderer degrades gracefully. Performance Timing assertions that the renderer completes well under a human-perceptible threshold for graphs of the sizes typically encountered in real repos. Security Commit messages and branch names containing ANSI escape sequences, control characters, and shell metacharacters must be sanitised before reaching the terminal — the renderer must never pass raw untrusted bytes to print(). """ from __future__ import annotations import contextlib import datetime import io import json import pathlib import time from collections.abc import Mapping import pytest from tests.cli_test_helper import CliRunner from muse.core.types import blob_id, fake_id from muse.core.object_store import write_object from muse.core.paths import heads_dir, muse_dir, ref_path from muse.core.commits import CommitRecord, write_commit from muse.core.snapshots import SnapshotRecord, write_snapshot from muse.core.ids import hash_snapshot, hash_commit from muse.cli.commands.log import _render_graph, _topo_sort, _collect_all_commits runner = CliRunner() cli = None # --------------------------------------------------------------------------- # Shared repo-building helpers # --------------------------------------------------------------------------- def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: """Initialise a minimal Muse repository under *tmp_path*. Creates the required ``.muse/`` directory structure, writes ``repo.json`` and ``HEAD``, and returns ``(root, repo_id)``. The default branch is ``main``; tests that need additional branches create them by writing ref files under ``.muse/refs/heads/``. """ dot_muse = muse_dir(tmp_path) dot_muse.mkdir() repo_id = fake_id("repo") (dot_muse / "repo.json").write_text( json.dumps( { "repo_id": repo_id, "domain": "code", "default_branch": "main", "created_at": "2025-01-01T00:00:00+00:00", } ), encoding="utf-8", ) (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "snapshots").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "objects").mkdir() return tmp_path, repo_id _commit_counter = 0 # monotonic counter so timestamps are strictly ordered def _make_commit( root: pathlib.Path, branch: str, files: dict[str, bytes], parent_id: str | None = None, parent2_id: str | None = None, message: str = "commit", ) -> str: """Write a commit record to *root* and advance the branch ref. ``files`` is a mapping of repo-relative path → bytes content. Both blob objects and snapshot records are written to the object store. If ``parent2_id`` is supplied the commit becomes a two-parent merge commit. Returns the new commit ID (``sha256:`` string). """ global _commit_counter _commit_counter += 1 manifest: dict[str, str] = {} for rel, content in files.items(): oid = blob_id(content) write_object(root, oid, content) manifest[rel] = oid dest = root / rel dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(content) snap_id = hash_snapshot(manifest) # Use a fixed base time offset by counter so every commit has a unique, # strictly-ordered timestamp. This makes _topo_sort deterministic. committed_at = datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(seconds=_commit_counter) parent_ids = [p for p in (parent_id, parent2_id) if p] commit_id = hash_commit( parent_ids=parent_ids, snapshot_id=snap_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) write_commit( root, CommitRecord( commit_id=commit_id, branch=branch, snapshot_id=snap_id, message=message, committed_at=committed_at, parent_commit_id=parent_id, parent2_commit_id=parent2_id, ), ) rf = ref_path(root, branch) rf.parent.mkdir(parents=True, exist_ok=True) rf.write_text(commit_id, encoding="utf-8") return commit_id def _env(root: pathlib.Path) -> Mapping[str, str]: """Return the environment dict needed to target a test repo.""" return {"MUSE_REPO_ROOT": str(root)} def _capture_graph( root: pathlib.Path, branch: str = "main", all_branches: bool = False, ) -> str: """Invoke ``_render_graph`` and return all printed lines as a single string.""" buf = io.StringIO() with contextlib.redirect_stdout(buf): _render_graph(root, branch, all_branches=all_branches, tty=False) return buf.getvalue() def _graph_lines(output: str) -> list[str]: """Return non-empty lines from captured graph output.""" return [line for line in output.splitlines() if line.strip()] def _commit_prefix(line: str) -> str: """Extract the ASCII art prefix from a commit line (the part before sha256:). Connector-only lines (``|/``, ``|\\``, etc.) are returned verbatim. """ if "sha256:" in line: return line[: line.index("sha256:")].rstrip() return line.rstrip() def _prefixes(output: str) -> list[str]: """Return the graph-art prefix for every non-empty line in *output*.""" return [_commit_prefix(ln) for ln in _graph_lines(output)] # --------------------------------------------------------------------------- # Unit / integration — linear chain # --------------------------------------------------------------------------- class TestLinearChain: """A single-branch chain with no merges must show only ``*`` commit rows and ``|`` connector rows between them — no backslashes, no slashes.""" def test_single_commit_shows_asterisk(self, tmp_path: pathlib.Path) -> None: """A repo with exactly one commit should print a single ``*`` row.""" root, repo_id = _init_repo(tmp_path) _make_commit(root, "main", {"a.txt": b"hello"}, message="init") output = _capture_graph(root) lines = _graph_lines(output) assert len(lines) == 1 assert lines[0].startswith("*"), f"Expected '* sha256:...', got: {lines[0]!r}" def test_two_commits_connector_is_pipe(self, tmp_path: pathlib.Path) -> None: """Two commits in a linear chain must have a ``|`` connector between them.""" root, repo_id = _init_repo(tmp_path) c1 = _make_commit(root, "main", {"a.txt": b"v1"}, message="first") _make_commit(root, "main", {"a.txt": b"v2"}, parent_id=c1, message="second") prefixes = _prefixes(_capture_graph(root)) assert prefixes[0] == "*", f"Top commit prefix: {prefixes[0]!r}" assert prefixes[1] == "|", f"Connector prefix: {prefixes[1]!r}" assert prefixes[2] == "*", f"Bottom commit prefix: {prefixes[2]!r}" def test_three_commits_no_backslash_no_slash(self, tmp_path: pathlib.Path) -> None: """A three-commit linear chain must contain no backslash or slash connectors.""" root, repo_id = _init_repo(tmp_path) c1 = _make_commit(root, "main", {"a.txt": b"1"}, message="one") c2 = _make_commit(root, "main", {"a.txt": b"2"}, parent_id=c1, message="two") _make_commit(root, "main", {"a.txt": b"3"}, parent_id=c2, message="three") output = _capture_graph(root) assert "\\" not in output, f"Unexpected \\ in linear output:\n{output}" assert "/" not in output, f"Unexpected / in linear output:\n{output}" def test_linear_commit_count_matches(self, tmp_path: pathlib.Path) -> None: """Every commit in the chain must appear exactly once in the graph.""" root, repo_id = _init_repo(tmp_path) prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0") for i in range(1, 6): prev = _make_commit(root, "main", {"a.txt": f"v{i}".encode()}, parent_id=prev, message=f"c{i}") output = _capture_graph(root) commit_lines = [ln for ln in _graph_lines(output) if "sha256:" in ln] assert len(commit_lines) == 6, f"Expected 6 commit lines, got {len(commit_lines)}" def test_no_commits_prints_sentinel(self, tmp_path: pathlib.Path) -> None: """A branch with no commits must print ``(no commits)`` rather than crashing.""" root, _ = _init_repo(tmp_path) output = _capture_graph(root) assert "(no commits)" in output, f"Expected '(no commits)', got: {output!r}" # --------------------------------------------------------------------------- # Unit / integration — merge commit connector # --------------------------------------------------------------------------- class TestMergeConnector: """After a two-parent merge commit the connector row must be ``|\\`` (pipe-backslash) — the fix for Bug 1 which was producing ``|\\|``.""" def test_merge_connector_is_backslash(self, tmp_path: pathlib.Path) -> None: """``|\\`` must appear immediately after a merge commit row.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"merge"}, parent_id=ours, parent2_id=feat, message="merge") output = _capture_graph(root) lines = _graph_lines(output) # The first line is the merge commit; the second must be the connector. assert lines[0].startswith("*"), f"First line should be merge commit: {lines[0]!r}" assert lines[1] == "|\\", ( f"Connector after merge must be '|\\\\', got: {lines[1]!r}\nFull output:\n{output}" ) def test_merge_connector_no_extra_trailing_pipe(self, tmp_path: pathlib.Path) -> None: """The newly-opened extra-parent lane must not produce a trailing ``|`` after the backslash — that was the ``|\\|`` bug.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge") output = _capture_graph(root) lines = _graph_lines(output) assert "|\\|" not in output, ( f"Found |\\\\| (old bug) in output:\n{output}" ) assert lines[1] == "|\\", ( f"Connector must be exactly '|\\\\', got {lines[1]!r}" ) def test_merge_connector_has_no_extra_columns(self, tmp_path: pathlib.Path) -> None: """The connector row for a two-parent merge must be exactly 2 chars wide.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge") lines = _graph_lines(_capture_graph(root)) assert len(lines[1]) == 2, ( f"Connector row must be exactly 2 chars ('|\\\\'), got {lines[1]!r}" ) def test_merge_commit_row_has_asterisk_at_col_zero(self, tmp_path: pathlib.Path) -> None: """The merge commit itself must print ``*`` in column 0.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge") lines = _graph_lines(_capture_graph(root)) prefix = _commit_prefix(lines[0]) assert prefix == "*", f"Merge commit must be at column 0, got: {prefix!r}" # --------------------------------------------------------------------------- # Unit / integration — convergence connector # --------------------------------------------------------------------------- class TestConvergenceConnector: """When two lanes both converge on the same ancestor commit the connector row before that commit must show ``|/`` — the fix for Bug 2.""" def test_convergence_connector_present(self, tmp_path: pathlib.Path) -> None: """``|/`` must appear before the shared ancestor in a diamond graph.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit") output = _capture_graph(root) assert "|/" in output, ( f"Expected '|/' convergence connector in output:\n{output}" ) def test_convergence_connector_before_shared_ancestor( self, tmp_path: pathlib.Path ) -> None: """The ``|/`` connector must appear immediately before the shared-ancestor row.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit") lines = _graph_lines(_capture_graph(root)) # Find the |/ line. slash_idx = next((i for i, ln in enumerate(lines) if "|/" in ln), None) assert slash_idx is not None, "No |/ connector found" # The row immediately following |/ must be the shared ancestor commit. next_line = lines[slash_idx + 1] assert "sha256:" in next_line and "initial commit" in next_line, ( f"Row after |/ should be initial commit, got: {next_line!r}" ) def test_initial_commit_has_no_trailing_pipe(self, tmp_path: pathlib.Path) -> None: """After convergence the initial commit row must show only ``*``, not ``* |`` with a dangling second column — that was the original bug.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit") lines = _graph_lines(_capture_graph(root)) last_commit_line = next( ln for ln in reversed(lines) if "sha256:" in ln ) prefix = _commit_prefix(last_commit_line) assert prefix == "*", ( f"Shared ancestor must render at col 0 with no trailing '|', " f"got prefix {prefix!r}" ) def test_no_dangling_pipe_column_after_convergence( self, tmp_path: pathlib.Path ) -> None: """No ``| |`` connector should appear after ``|/`` has closed a lane.""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat commit") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours commit") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge commit") output = _capture_graph(root) lines = _graph_lines(output) slash_idx = next((i for i, ln in enumerate(lines) if "|/" in ln), None) assert slash_idx is not None for ln in lines[slash_idx + 1 :]: assert "| |" not in ln, ( f"Found '| |' connector after convergence (stale lane):\n{output}" ) # --------------------------------------------------------------------------- # Unit / integration — full diamond pattern # --------------------------------------------------------------------------- class TestDiamondPattern: """A canonical diamond: merge → (left, right) → base. Validates the complete shape of the rendered graph end-to-end.""" def _build_diamond( self, tmp_path: pathlib.Path ) -> tuple[pathlib.Path, str, str, str, str]: """Build a four-commit diamond and return (root, base, left, right, merge).""" root, repo_id = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="initial commit") (heads_dir(root) / "feat").write_text(base) left = _make_commit(root, "main", {"f.txt": b"left"}, parent_id=base, message="left commit") right = _make_commit(root, "feat", {"f.txt": b"right"}, parent_id=base, message="right commit") merge = _make_commit( root, "main", {"f.txt": b"merged"}, parent_id=left, parent2_id=right, message="merge commit" ) return root, base, left, right, merge def test_diamond_line_count(self, tmp_path: pathlib.Path) -> None: """A diamond graph must produce exactly the right number of output lines: 4 commit rows + 1 merge connector + inter-commit connectors + 1 slash.""" root, *_ = self._build_diamond(tmp_path) lines = _graph_lines(_capture_graph(root)) commit_rows = [ln for ln in lines if "sha256:" in ln] assert len(commit_rows) == 4, f"Expected 4 commits, got {len(commit_rows)}: {commit_rows}" def test_diamond_contains_backslash(self, tmp_path: pathlib.Path) -> None: """The merge connector ``|\\`` must appear exactly once.""" root, *_ = self._build_diamond(tmp_path) output = _capture_graph(root) lines = _graph_lines(output) backslash_lines = [ln for ln in lines if ln == "|\\"] assert len(backslash_lines) == 1, ( f"Expected exactly one '|\\\\' connector, got: {backslash_lines}\n{output}" ) def test_diamond_contains_slash(self, tmp_path: pathlib.Path) -> None: """The convergence connector ``|/`` must appear exactly once.""" root, *_ = self._build_diamond(tmp_path) output = _capture_graph(root) lines = _graph_lines(output) slash_lines = [ln for ln in lines if "|/" in ln] assert len(slash_lines) == 1, ( f"Expected exactly one '|/' connector, got: {slash_lines}\n{output}" ) def test_diamond_first_line_is_merge(self, tmp_path: pathlib.Path) -> None: """The topologically-first line must be the merge commit.""" root, *_, merge_id = self._build_diamond(tmp_path) merge_short = merge_id[7:19] # strip 'sha256:' + take 12 hex chars lines = _graph_lines(_capture_graph(root)) assert merge_short in lines[0], ( f"First line must be merge commit (sha {merge_short}), got: {lines[0]!r}" ) def test_diamond_last_line_is_base(self, tmp_path: pathlib.Path) -> None: """The last commit row must be the shared ancestor (base).""" root, base_id, *_ = self._build_diamond(tmp_path) base_short = base_id[7:19] lines = _graph_lines(_capture_graph(root)) commit_lines = [ln for ln in lines if "sha256:" in ln] assert base_short in commit_lines[-1], ( f"Last commit must be base ({base_short}), got: {commit_lines[-1]!r}" ) def test_diamond_no_wide_pipe_after_slash(self, tmp_path: pathlib.Path) -> None: """After the ``|/`` convergence connector there must be no two-column connector rows — the second lane has been closed.""" root, *_ = self._build_diamond(tmp_path) output = _capture_graph(root) lines = _graph_lines(output) slash_idx = next(i for i, ln in enumerate(lines) if "|/" in ln) for ln in lines[slash_idx + 1 :]: assert "| |" not in ln, ( f"Stale two-column connector found after convergence:\n{output}" ) # --------------------------------------------------------------------------- # Unit / integration — decorations # --------------------------------------------------------------------------- class TestDecorations: """HEAD and branch tip labels must appear on the correct commit rows.""" def test_head_decoration_on_tip(self, tmp_path: pathlib.Path) -> None: """The current HEAD commit must carry the ``HEAD -> main`` decoration.""" root, _ = _init_repo(tmp_path) _make_commit(root, "main", {"a.txt": b"x"}, message="init") output = _capture_graph(root) assert "HEAD" in output, f"HEAD label missing from graph:\n{output}" assert "main" in output, f"Branch name missing from graph:\n{output}" def test_feature_branch_tip_decorated(self, tmp_path: pathlib.Path) -> None: """A feature branch tip commit must show its branch name in the decoration.""" root, _ = _init_repo(tmp_path) base = _make_commit(root, "main", {"a.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) _make_commit(root, "feat", {"a.txt": b"feat"}, parent_id=base, message="feat work") _make_commit(root, "main", {"a.txt": b"ours"}, parent_id=base, message="main work") output = _capture_graph(root, all_branches=True) assert "feat" in output, f"Feature branch name not in graph:\n{output}" def test_message_first_line_only(self, tmp_path: pathlib.Path) -> None: """Only the first line of a multiline commit message must appear in the graph.""" root, _ = _init_repo(tmp_path) _make_commit(root, "main", {"a.txt": b"x"}, message="first line\nsecond line\nthird line") output = _capture_graph(root) assert "first line" in output assert "second line" not in output assert "third line" not in output # --------------------------------------------------------------------------- # Unit / integration — lane reuse # --------------------------------------------------------------------------- class TestLaneReuse: """Closed lanes (``None`` slots) must be reused for new branches instead of ever-growing the lane array width.""" def test_lane_slot_reused_after_close(self, tmp_path: pathlib.Path) -> None: """After a branch terminates (root commit, no parents) its column slot must be reused for the next new branch, keeping the graph compact.""" root, _ = _init_repo(tmp_path) # Build two independent branches that share no common ancestor. # Branch A: a1 (root, no parent) → a2 # Branch B: b1 (root, no parent) → b2 (after a1 is done) a1 = _make_commit(root, "main", {"a.txt": b"a1"}, message="a1 root") a2 = _make_commit(root, "main", {"a.txt": b"a2"}, parent_id=a1, message="a2") (heads_dir(root) / "branchB").write_text("") # placeholder b1 = _make_commit(root, "branchB", {"b.txt": b"b1"}, message="b1 root") b2 = _make_commit(root, "branchB", {"b.txt": b"b2"}, parent_id=b1, message="b2") output = _capture_graph(root, all_branches=True) # Ensure both branches rendered without crash. assert "a1 root" in output or "a2" in output assert "b1 root" in output or "b2" in output # --------------------------------------------------------------------------- # End-to-end — CLI # --------------------------------------------------------------------------- class TestCliEndToEnd: """Full CLI invocations through ``CliRunner`` — verifies ``--graph`` is wired correctly and the output meets structural expectations.""" def _simple_linear_repo(self, tmp_path: pathlib.Path) -> pathlib.Path: root, _ = _init_repo(tmp_path) c1 = _make_commit(root, "main", {"a.txt": b"v1"}, message="first") _make_commit(root, "main", {"a.txt": b"v2"}, parent_id=c1, message="second") return root def test_graph_flag_exits_zero(self, tmp_path: pathlib.Path) -> None: """``muse log --graph`` must exit with code 0 on a valid repo.""" root = self._simple_linear_repo(tmp_path) result = runner.invoke(cli, ["log", "--graph"], env=_env(root)) assert result.exit_code == 0, ( f"Expected exit 0, got {result.exit_code}:\n{result.stderr}" ) def test_graph_flag_contains_asterisk(self, tmp_path: pathlib.Path) -> None: """``muse log --graph`` output must contain at least one ``*`` commit row.""" root = self._simple_linear_repo(tmp_path) result = runner.invoke(cli, ["log", "--graph"], env=_env(root)) assert "*" in result.stdout, f"No * in graph output:\n{result.stdout}" def test_graph_flag_contains_sha256(self, tmp_path: pathlib.Path) -> None: """Every commit row must contain its content-addressed ID.""" root = self._simple_linear_repo(tmp_path) result = runner.invoke(cli, ["log", "--graph"], env=_env(root)) assert "sha256:" in result.stdout, f"No sha256 in output:\n{result.stdout}" def test_cli_merge_graph_contains_backslash(self, tmp_path: pathlib.Path) -> None: """``muse log --graph`` on a repo with a merge must contain ``|\\``.""" root, _ = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge") result = runner.invoke(cli, ["log", "--graph"], env=_env(root)) assert "|\\" in result.stdout, ( f"No '|\\\\' connector in CLI graph output:\n{result.stdout}" ) def test_cli_diamond_graph_contains_slash(self, tmp_path: pathlib.Path) -> None: """``muse log --graph`` on a diamond must contain ``|/``.""" root, _ = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge") result = runner.invoke(cli, ["log", "--graph"], env=_env(root)) assert "|/" in result.stdout, ( f"No '|/' convergence in CLI graph output:\n{result.stdout}" ) def test_cli_no_commits_graceful(self, tmp_path: pathlib.Path) -> None: """``muse log --graph`` on an empty repo must exit 0 and say '(no commits)'.""" root, _ = _init_repo(tmp_path) result = runner.invoke(cli, ["log", "--graph"], env=_env(root)) assert result.exit_code == 0 assert "no commits" in result.stdout.lower(), ( f"Expected '(no commits)' for empty repo:\n{result.stdout}" ) def test_cli_oneline_not_graph(self, tmp_path: pathlib.Path) -> None: """``muse log --oneline`` (no ``--graph``) must not include ASCII art.""" root, _ = _init_repo(tmp_path) _make_commit(root, "main", {"a.txt": b"x"}, message="init") result = runner.invoke(cli, ["log", "--oneline"], env=_env(root)) assert "|" not in result.stdout, ( f"--oneline must not show graph art:\n{result.stdout}" ) # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: """Large graphs and wide branching patterns must not crash, corrupt lane state, or produce malformed ASCII art.""" def test_long_linear_chain_100_commits(self, tmp_path: pathlib.Path) -> None: """A 100-commit linear chain must render all 100 commits without error.""" root, _ = _init_repo(tmp_path) prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0") for i in range(1, 100): prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}") output = _capture_graph(root) commit_rows = [ln for ln in _graph_lines(output) if "sha256:" in ln] assert len(commit_rows) == 100, ( f"Expected 100 commit rows, got {len(commit_rows)}" ) def test_long_chain_only_asterisk_and_pipe(self, tmp_path: pathlib.Path) -> None: """In a 50-commit linear chain, every commit prefix must be ``*`` and every connector must be ``|`` — no slash, no backslash.""" root, _ = _init_repo(tmp_path) prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0") for i in range(1, 50): prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}") output = _capture_graph(root) prefixes = _prefixes(output) for p in prefixes: assert p in ("*", "|"), f"Unexpected graph prefix {p!r} in linear chain" def test_sequential_merges_no_crash(self, tmp_path: pathlib.Path) -> None: """Five sequential feature branches, each merged back to main, must render without crashing and without any corrupted lane state.""" root, _ = _init_repo(tmp_path) tip = _make_commit(root, "main", {"a.txt": b"init"}, message="init") for i in range(5): feat_name = f"feat{i}" (heads_dir(root) / feat_name).write_text(tip) feat_tip = _make_commit( root, feat_name, {"a.txt": f"feat{i}".encode()}, parent_id=tip, message=f"feat{i} work" ) tip = _make_commit( root, "main", {"a.txt": f"merge{i}".encode()}, parent_id=tip, parent2_id=feat_tip, message=f"merge feat{i}" ) output = _capture_graph(root) assert output.strip(), "Graph output must not be empty" # No consecutive identical pipes — that would indicate a runaway lane. assert "| | | | | |" not in output, ( f"Suspiciously many concurrent lanes:\n{output}" ) def test_wide_fan_three_branches(self, tmp_path: pathlib.Path) -> None: """Three concurrent branches (A, B, C) merged together must produce a graph with exactly three ``*`` rows plus the merge, and both ``|\\`` and ``|/`` connectors.""" root, _ = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "branchA").write_text(base) (heads_dir(root) / "branchB").write_text(base) a = _make_commit(root, "branchA", {"f.txt": b"A"}, parent_id=base, message="branch A") b = _make_commit(root, "branchB", {"f.txt": b"B"}, parent_id=base, message="branch B") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="main work") # Merge A into main first, then merge B. m1 = _make_commit(root, "main", {"f.txt": b"m1"}, parent_id=ours, parent2_id=a, message="merge A") _make_commit(root, "main", {"f.txt": b"m2"}, parent_id=m1, parent2_id=b, message="merge B") output = _capture_graph(root) assert "\\" in output, f"Expected \\\\ in wide fan output:\n{output}" assert "/" in output, f"Expected / in wide fan output:\n{output}" def test_graph_truncation_prints_warning(self, tmp_path: pathlib.Path) -> None: """When the commit count exceeds the configured cap a truncation warning must appear before the graph rows.""" root, _ = _init_repo(tmp_path) prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0") for i in range(1, 12): prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}") # Write a very low cap so 12 commits triggers truncation. config_path = muse_dir(root) / "config.toml" config_path.write_text("[limits]\nmax_graph_commits = 5\n", encoding="utf-8") output = _capture_graph(root) assert "truncated" in output.lower(), ( f"Expected truncation warning with low cap:\n{output}" ) # --------------------------------------------------------------------------- # Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: """Degenerate or partially corrupt repos must be handled gracefully — the renderer must never crash with an unhandled exception.""" def test_orphaned_parent_ref_skipped(self, tmp_path: pathlib.Path) -> None: """A commit whose parent_commit_id does not exist in the object store must still render without crashing; the orphaned parent is silently excluded from the graph. This test bypasses ``write_commit``'s parent-existence guard to deliberately inject a commit that references a non-existent parent — simulating object-store corruption or a partial fetch.""" import json as _json from muse.core.commits import commit_path as _commit_path root, _ = _init_repo(tmp_path) # Write a well-formed commit first so we have a real tip commit. real_cid = _make_commit(root, "main", {"f.txt": b"real"}, message="real commit") # Now inject a second commit that references a nonexistent parent by # writing the JSON directly — bypassing write_commit's validation. fake_parent = "sha256:" + "a" * 64 orphan_cid = "sha256:" + "b" * 64 orphan_data = { "commit_id": orphan_cid, "branch": "main", "snapshot_id": "sha256:" + "0" * 64, "message": "orphan commit", "committed_at": "2025-01-01T00:00:01+00:00", "parent_commit_id": fake_parent, "parent2_commit_id": None, "author": "gabriel", "agent_id": "", "model_id": "", "signature": None, "signer_public_key": None, "sem_ver_bump": "patch", "breaking_changes": [], "metadata": {}, "structured_delta": None, } p = _commit_path(root, orphan_cid) p.parent.mkdir(parents=True, exist_ok=True) p.write_text(_json.dumps(orphan_data), encoding="utf-8") # The renderer uses BFS from the branch tip (real_cid) — the orphan # commit is not reachable from the tip so it is never visited. # The fake parent is not in the store so BFS stops there gracefully. output = _capture_graph(root) assert "sha256:" in output, f"Commit row missing:\n{output}" def test_root_commit_closes_its_lane(self, tmp_path: pathlib.Path) -> None: """A root commit (no parents) must close its lane column — subsequent renders must not leave a dangling pipe for a terminated branch.""" root, _ = _init_repo(tmp_path) _make_commit(root, "main", {"a.txt": b"root"}, message="root commit") lines = _graph_lines(_capture_graph(root)) # Only one line — the commit itself. No connector should follow. commit_lines = [ln for ln in lines if "sha256:" in ln] connector_lines = [ln for ln in lines if "sha256:" not in ln] assert len(commit_lines) == 1 assert len(connector_lines) == 0, ( f"Root commit must not produce a trailing connector: {connector_lines}" ) def test_empty_branch_ref_falls_back_to_no_commits( self, tmp_path: pathlib.Path ) -> None: """If the branch ref file exists but is empty, the renderer must print '(no commits)' and not crash.""" root, _ = _init_repo(tmp_path) # Write a blank ref — simulates a branch that was created but never committed. ref_path(root, "main").parent.mkdir(parents=True, exist_ok=True) ref_path(root, "main").write_text("", encoding="utf-8") output = _capture_graph(root) assert "no commits" in output.lower(), ( f"Expected '(no commits)' for blank ref, got:\n{output}" ) def test_multiline_commit_message_renders_first_line_only( self, tmp_path: pathlib.Path ) -> None: """A commit with a multiline message must expose only the first line in the graph — downstream parsers rely on one-line graph output.""" root, _ = _init_repo(tmp_path) _make_commit( root, "main", {"a.txt": b"x"}, message="Summary line\n\nDetailed paragraph.\nMore detail." ) output = _capture_graph(root) assert "Summary line" in output assert "Detailed paragraph" not in output def test_commit_with_no_files_renders(self, tmp_path: pathlib.Path) -> None: """An empty-manifest commit (no files) must render without crashing.""" root, _ = _init_repo(tmp_path) _make_commit(root, "main", {}, message="empty manifest commit") output = _capture_graph(root) assert "empty manifest commit" in output def test_graph_output_has_no_blank_prefix_on_commit_rows( self, tmp_path: pathlib.Path ) -> None: """No commit row in the graph may have a blank graph prefix — every commit must be preceded by at least a ``*``.""" root, _ = _init_repo(tmp_path) prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0") for i in range(1, 5): prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}") lines = _graph_lines(_capture_graph(root)) for ln in lines: if "sha256:" in ln: prefix = _commit_prefix(ln) assert prefix, f"Commit row has empty graph prefix: {ln!r}" assert "*" in prefix, f"Commit row prefix missing '*': {ln!r}" # --------------------------------------------------------------------------- # Performance # --------------------------------------------------------------------------- class TestPerformance: """The renderer must complete within a human-imperceptible threshold for graph sizes typical of real development repos.""" def test_hundred_commits_renders_under_two_seconds( self, tmp_path: pathlib.Path ) -> None: """100 linear commits must render in under 2 seconds on any reasonable CI or developer machine.""" root, _ = _init_repo(tmp_path) prev = _make_commit(root, "main", {"a.txt": b"0"}, message="c0") for i in range(1, 100): prev = _make_commit(root, "main", {"a.txt": f"{i}".encode()}, parent_id=prev, message=f"c{i}") start = time.perf_counter() output = _capture_graph(root) elapsed = time.perf_counter() - start assert elapsed < 2.0, ( f"Graph rendering took {elapsed:.2f}s for 100 commits — too slow" ) assert "sha256:" in output # sanity: output was actually produced def test_diamond_renders_under_one_second(self, tmp_path: pathlib.Path) -> None: """A four-commit diamond graph (typical merge scenario) must render in under 1 second — this path is hit constantly during normal dev workflows.""" root, _ = _init_repo(tmp_path) base = _make_commit(root, "main", {"f.txt": b"base"}, message="base") (heads_dir(root) / "feat").write_text(base) feat = _make_commit(root, "feat", {"f.txt": b"feat"}, parent_id=base, message="feat") ours = _make_commit(root, "main", {"f.txt": b"ours"}, parent_id=base, message="ours") _make_commit(root, "main", {"f.txt": b"mg"}, parent_id=ours, parent2_id=feat, message="merge") start = time.perf_counter() _capture_graph(root) elapsed = time.perf_counter() - start assert elapsed < 1.0, ( f"Diamond graph rendering took {elapsed:.2f}s — expected < 1s" ) def test_fifty_sequential_merges_renders_under_five_seconds( self, tmp_path: pathlib.Path ) -> None: """A repo with 50 sequential feature-branch merges (100 total commits) must render in under 5 seconds.""" root, _ = _init_repo(tmp_path) tip = _make_commit(root, "main", {"a.txt": b"init"}, message="init") for i in range(25): feat_name = f"stress{i}" (heads_dir(root) / feat_name).write_text(tip) feat_tip = _make_commit( root, feat_name, {"a.txt": f"f{i}".encode()}, parent_id=tip, message=f"feat {i}" ) tip = _make_commit( root, "main", {"a.txt": f"m{i}".encode()}, parent_id=tip, parent2_id=feat_tip, message=f"merge {i}" ) start = time.perf_counter() output = _capture_graph(root) elapsed = time.perf_counter() - start assert elapsed < 5.0, ( f"50-merge graph took {elapsed:.2f}s — expected < 5s" ) assert "sha256:" in output # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: """The graph renderer must never pass raw untrusted commit data directly to the terminal. ANSI escape sequences, control characters, and shell metacharacters embedded in commit messages or branch names must be neutralised before output. All assertions are against the captured string — ``_render_graph`` routes through ``sanitize_display`` from ``muse.core.validation``, which strips or replaces dangerous characters before they reach the tty.""" def test_ansi_escape_in_commit_message_neutralised( self, tmp_path: pathlib.Path ) -> None: """A commit message containing an ANSI escape sequence must not pass the raw escape bytes through to the printed output. An attacker could embed ``\\x1b[2J`` (clear screen) or a colour-change sequence in a commit message to hijack the terminal display of anyone who runs ``muse log --graph``.""" root, _ = _init_repo(tmp_path) evil_message = "normal prefix \x1b[31mRED INJECTION\x1b[0m suffix" _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message) output = _capture_graph(root) assert "\x1b[31m" not in output, ( "Raw ANSI colour escape must not reach graph output" ) assert "\x1b[0m" not in output, ( "Raw ANSI reset escape must not reach graph output" ) def test_null_byte_in_commit_message_neutralised( self, tmp_path: pathlib.Path ) -> None: """A commit message containing null bytes must not produce a broken or truncated graph line. Null bytes in terminal output can cause visual corruption and in some terminal emulators trigger unsafe behaviour.""" root, _ = _init_repo(tmp_path) evil_message = "before\x00after" _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message) output = _capture_graph(root) assert "\x00" not in output, "Null byte must not appear in graph output" # The graph must still render something for this commit. assert "sha256:" in output def test_carriage_return_in_message_neutralised( self, tmp_path: pathlib.Path ) -> None: """A ``\\r`` (carriage return) in a commit message can overwrite the beginning of the line in a terminal, causing the graph art prefix to be visually erased. It must not pass through.""" root, _ = _init_repo(tmp_path) evil_message = "legit text\r" _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message) output = _capture_graph(root) assert "\r" not in output, "Carriage return must not appear in graph output" def test_terminal_title_escape_in_message_neutralised( self, tmp_path: pathlib.Path ) -> None: """The OSC sequence ``\\x1b]0;title\\x07`` sets the terminal window title. Embedding this in a commit message could silently rename a user's terminal window — it must be stripped.""" root, _ = _init_repo(tmp_path) evil_message = "\x1b]0;malicious title\x07normal text" _make_commit(root, "main", {"a.txt": b"x"}, message=evil_message) output = _capture_graph(root) assert "\x1b]" not in output, "OSC escape sequence must not reach graph output" def test_shell_metacharacters_in_message_pass_through_safe( self, tmp_path: pathlib.Path ) -> None: """Shell metacharacters in commit messages (``$``, backtick, ``!``) are benign in Python ``print()`` output — they must not be over-escaped either, since aggressive escaping corrupts legitimate message text. This test documents the expected behaviour: shell-special chars are preserved as-is because they only have significance when interpreted by a shell, not when printed to stdout.""" root, _ = _init_repo(tmp_path) message = "fix: avoid $PATH collision & handle `nul` correctly; cost=O(1)" _make_commit(root, "main", {"a.txt": b"x"}, message=message) output = _capture_graph(root) assert "$PATH" in output, "Dollar sign in commit message must not be mangled" assert "O(1)" in output, "Parentheses in commit message must not be mangled" def test_very_long_commit_message_does_not_break_line_structure( self, tmp_path: pathlib.Path ) -> None: """A 4 000-character commit message must not cause the graph prefix to wrap onto multiple lines in the captured output (output goes to a non-tty StringIO, so no terminal wrapping occurs).""" root, _ = _init_repo(tmp_path) long_msg = "A" * 4000 _make_commit(root, "main", {"a.txt": b"x"}, message=long_msg) output = _capture_graph(root) commit_lines = [ln for ln in _graph_lines(output) if "sha256:" in ln] assert len(commit_lines) == 1, ( f"Long message must not produce multiple commit rows, got:\n{output[:500]}" ) prefix = _commit_prefix(commit_lines[0]) assert "*" in prefix, f"Long-message commit must still show '*' prefix" def test_unicode_in_commit_message_renders_safely( self, tmp_path: pathlib.Path ) -> None: """Emoji and non-ASCII Unicode in commit messages must render without raising ``UnicodeEncodeError`` or corrupting the graph prefix.""" root, _ = _init_repo(tmp_path) _make_commit( root, "main", {"a.txt": b"x"}, message="feat: 🎵 add MIDI export — café résumé naïve" ) # Must not raise. output = _capture_graph(root) assert "sha256:" in output, f"Unicode commit did not render:\n{output}" # --------------------------------------------------------------------------- # _topo_sort unit tests # --------------------------------------------------------------------------- class TestTopoSort: """Unit tests for the ``_topo_sort`` helper in isolation from I/O. ``_topo_sort`` must return all commits, children before parents, with ties broken by timestamp (most recent first).""" def _make_record( self, cid: str, parent: str | None = None, parent2: str | None = None, ts_offset: int = 0, ) -> CommitRecord: return CommitRecord( commit_id=cid, branch="main", snapshot_id="sha256:" + "0" * 64, message="msg", committed_at=datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc) + datetime.timedelta(seconds=ts_offset), parent_commit_id=parent, parent2_commit_id=parent2, ) def test_single_commit_returned(self) -> None: """A single commit must be returned as a one-element list.""" c = self._make_record("sha256:" + "a" * 64) result = _topo_sort({"sha256:" + "a" * 64: c}) assert len(result) == 1 assert result[0].commit_id == "sha256:" + "a" * 64 def test_parent_comes_after_child(self) -> None: """In a two-commit chain child must appear before parent.""" parent_id = "sha256:" + "p" * 64 child_id = "sha256:" + "c" * 64 parent = self._make_record(parent_id, ts_offset=0) child = self._make_record(child_id, parent=parent_id, ts_offset=1) result = _topo_sort({parent_id: parent, child_id: child}) assert result[0].commit_id == child_id assert result[1].commit_id == parent_id def test_merge_commit_before_both_parents(self) -> None: """A merge commit must appear before both of its parents.""" base_id = "sha256:" + "b" * 64 left_id = "sha256:" + "l" * 64 right_id = "sha256:" + "r" * 64 merge_id = "sha256:" + "m" * 64 base = self._make_record(base_id, ts_offset=0) left = self._make_record(left_id, parent=base_id, ts_offset=1) right = self._make_record(right_id, parent=base_id, ts_offset=2) merge = self._make_record(merge_id, parent=left_id, parent2=right_id, ts_offset=3) commits = {base_id: base, left_id: left, right_id: right, merge_id: merge} result = _topo_sort(commits) ids = [r.commit_id for r in result] assert ids.index(merge_id) < ids.index(left_id) assert ids.index(merge_id) < ids.index(right_id) assert ids.index(left_id) < ids.index(base_id) assert ids.index(right_id) < ids.index(base_id) def test_all_commits_present_in_result(self) -> None: """``_topo_sort`` must return every commit exactly once.""" ids = ["sha256:" + c * 64 for c in "abcde"] records = {} for i, cid in enumerate(ids): parent = ids[i - 1] if i > 0 else None records[cid] = self._make_record(cid, parent=parent, ts_offset=i) result = _topo_sort(records) assert len(result) == 5 assert {r.commit_id for r in result} == set(ids)