"""Supercharge tests for ``muse merge-tree``. Coverage tiers -------------- - JSON envelope: exit_code and duration_ms present on clean and conflict outcomes - Error payload: errors go to stdout as JSON in --json mode, no dual stderr prose - Data integrity: sha256: OID prefix preserved in merged_manifest; --base with sha256:-prefixed commit ID accepted - TypedDicts: _MergeTreeJson and _MergeTreeErrorJson with required annotations - Docstring: module docstring covers exit_code and duration_ms - No-prose pollution: JSON stdout is valid on all non-error paths - Stress: 100-file manifest, 40% conflict rate — correct counts, correct exit code """ from __future__ import annotations from collections.abc import Mapping import argparse import datetime import json import pathlib from typing import get_type_hints from muse.core.object_store import write_object from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.types import blob_id from muse.core.paths import ref_path, muse_dir runner = CliRunner() _REPO_ID = "mt-supercharge" _DT = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init_repo(tmp_path: pathlib.Path) -> pathlib.Path: dot_muse = muse_dir(tmp_path) for sub in ("objects", "commits", "snapshots", "refs/heads"): (dot_muse / sub).mkdir(parents=True) (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8") (dot_muse / "repo.json").write_text( json.dumps({"repo_id": _REPO_ID, "domain": "code"}), encoding="utf-8" ) return tmp_path def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _write_obj(root: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(root, oid, content) return oid def _make_commit( root: pathlib.Path, manifest: Mapping[str, str], branch: str, parent: str | None = None, msg: str = "test", ) -> str: snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest, created_at=_DT)) parent_ids = [parent] if parent else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=snap_id, message=msg, committed_at_iso=_DT.isoformat(), ) write_commit(root, CommitRecord( commit_id=cid, branch=branch, snapshot_id=snap_id, message=msg, committed_at=_DT, parent_commit_id=parent, )) ref = ref_path(root, branch) ref.parent.mkdir(parents=True, exist_ok=True) ref.write_text(cid, encoding="utf-8") return cid def _mt(root: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke(cli, ["merge-tree", *args], env=_env(root)) def _diverged(root: pathlib.Path) -> tuple[str, str, str]: """base → branch-a (adds a.py) and base → branch-b (adds b.py). No conflict.""" base_oid = _write_obj(root, b"base") base_cid = _make_commit(root, {"base.py": base_oid}, "main") a_oid = _write_obj(root, b"a content") a_cid = _make_commit(root, {"base.py": base_oid, "a.py": a_oid}, "branch-a", parent=base_cid) b_oid = _write_obj(root, b"b content") b_cid = _make_commit(root, {"base.py": base_oid, "b.py": b_oid}, "branch-b", parent=base_cid) return base_cid, a_cid, b_cid def _conflicted(root: pathlib.Path) -> tuple[str, str, str]: """base → branch-a and branch-b both modify shared.py differently.""" v1 = _write_obj(root, b"v1") base_cid = _make_commit(root, {"shared.py": v1}, "main") va = _write_obj(root, b"version-a") a_cid = _make_commit(root, {"shared.py": va}, "branch-a", parent=base_cid) vb = _write_obj(root, b"version-b") b_cid = _make_commit(root, {"shared.py": vb}, "branch-b", parent=base_cid) return base_cid, a_cid, b_cid # --------------------------------------------------------------------------- # JSON envelope — exit_code # --------------------------------------------------------------------------- class TestJsonEnvelopeExitCode: def test_clean_merge_has_exit_code_zero(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--json") assert r.exit_code == 0 d = json.loads(r.output) assert "exit_code" in d, "exit_code missing from clean merge envelope" assert d["exit_code"] == 0 def test_conflict_merge_has_exit_code_nonzero(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _conflicted(root) r = _mt(root, "branch-a", "branch-b", "--json") assert r.exit_code != 0 d = json.loads(r.output) assert "exit_code" in d, "exit_code missing from conflict envelope" assert d["exit_code"] != 0 # --------------------------------------------------------------------------- # JSON envelope — duration_ms # --------------------------------------------------------------------------- class TestJsonEnvelopeDurationMs: def test_clean_merge_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--json") d = json.loads(r.output) assert "duration_ms" in d, "duration_ms missing from clean merge envelope" assert isinstance(d["duration_ms"], float) assert d["duration_ms"] >= 0.0 def test_conflict_merge_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _conflicted(root) r = _mt(root, "branch-a", "branch-b", "--json") d = json.loads(r.output) assert "duration_ms" in d, "duration_ms missing from conflict envelope" assert isinstance(d["duration_ms"], float) def test_write_objects_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--write-objects", "--json") d = json.loads(r.output) assert "duration_ms" in d # --------------------------------------------------------------------------- # Error payload — errors route to stdout as JSON in --json mode # --------------------------------------------------------------------------- class TestErrorPayload: def test_bad_branch1_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) r = _mt(root, "no-such", "also-no", "--json") assert r.exit_code != 0 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}" d = json.loads(r.output) assert d["status"] == "error" def test_bad_branch2_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) base_oid = _write_obj(root, b"x") _make_commit(root, {"x.py": base_oid}, "main") r = _mt(root, "main", "nonexistent", "--json") assert r.exit_code != 0 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}" d = json.loads(r.output) assert d["status"] == "error" def test_no_common_ancestor_error_goes_to_stdout(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) oid = _write_obj(root, b"x") _make_commit(root, {"x.py": oid}, "orphan-a") oid2 = _write_obj(root, b"y") _make_commit(root, {"y.py": oid2}, "orphan-b") r = _mt(root, "orphan-a", "orphan-b", "--json") assert r.exit_code != 0 assert not r.stderr.strip(), f"unexpected stderr: {r.stderr!r}" d = json.loads(r.output) assert d["status"] == "error" def test_error_payload_has_status_error(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) r = _mt(root, "ghost", "phantom", "--json") d = json.loads(r.output) assert d["status"] == "error" def test_error_payload_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) r = _mt(root, "ghost", "phantom", "--json") d = json.loads(r.output) assert "exit_code" in d assert d["exit_code"] != 0 def test_error_payload_has_error_field(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) r = _mt(root, "ghost", "phantom", "--json") d = json.loads(r.output) assert "error" in d assert d["error"] def test_no_duplicate_stderr_prose(self, tmp_path: pathlib.Path) -> None: """In --json mode, errors must not also print ❌ prose to stderr.""" root = _init_repo(tmp_path) r = _mt(root, "ghost", "phantom", "--json") assert "❌" not in r.stderr # --------------------------------------------------------------------------- # Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: def test_merged_manifest_oids_are_sha256_prefixed(self, tmp_path: pathlib.Path) -> None: """All non-null object IDs in merged_manifest must carry sha256: prefix.""" root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--json") d = json.loads(r.output) for path, oid in d["merged_manifest"].items(): if oid is not None: assert oid.startswith("sha256:"), ( f"OID for '{path}' missing sha256: prefix: {oid!r}" ) def test_base_with_sha256_prefixed_commit_id(self, tmp_path: pathlib.Path) -> None: """--base accepts sha256:-prefixed commit IDs (not just branch names).""" root = _init_repo(tmp_path) base_cid, a_cid, b_cid = _diverged(root) r = _mt(root, "branch-a", "branch-b", "--base", base_cid, "--json") assert r.exit_code == 0 d = json.loads(r.output) assert d["base"] == base_cid def test_branch_ids_echoed_in_response(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) base_cid, a_cid, b_cid = _diverged(root) r = _mt(root, "branch-a", "branch-b", "--json") d = json.loads(r.output) assert d["branch1"] == a_cid assert d["branch2"] == b_cid def test_conflict_paths_have_null_oid(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _conflicted(root) r = _mt(root, "branch-a", "branch-b", "--json") d = json.loads(r.output) for path in d["conflicts"]: assert d["merged_manifest"][path] is None def test_snapshot_id_is_sha256_prefixed(self, tmp_path: pathlib.Path) -> None: """--write-objects snapshot_id must carry sha256: prefix.""" root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--write-objects", "--json") d = json.loads(r.output) assert "snapshot_id" in d assert d["snapshot_id"].startswith("sha256:"), ( f"snapshot_id missing sha256: prefix: {d['snapshot_id']!r}" ) # --------------------------------------------------------------------------- # No-prose pollution # --------------------------------------------------------------------------- class TestNoProsePollution: def test_clean_merge_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--json") json.loads(r.output) # must not raise def test_conflict_merge_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _conflicted(root) json.loads(_mt(root, "branch-a", "branch-b", "--json").output) def test_no_emoji_in_clean_json(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) _diverged(root) r = _mt(root, "branch-a", "branch-b", "--json") assert "✅" not in r.output assert "❌" not in r.output # --------------------------------------------------------------------------- # TypedDicts # --------------------------------------------------------------------------- class TestTypedDicts: def test_merge_tree_json_typeddict_exists(self) -> None: from muse.cli.commands.merge_tree import _MergeTreeJson assert _MergeTreeJson is not None def test_merge_tree_error_json_typeddict_exists(self) -> None: from muse.cli.commands.merge_tree import _MergeTreeErrorJson assert _MergeTreeErrorJson is not None def test_merge_tree_json_has_exit_code_annotation(self) -> None: from muse.cli.commands.merge_tree import _MergeTreeJson hints = get_type_hints(_MergeTreeJson) assert "exit_code" in hints def test_merge_tree_json_has_duration_ms_annotation(self) -> None: from muse.cli.commands.merge_tree import _MergeTreeJson hints = get_type_hints(_MergeTreeJson) assert "duration_ms" in hints def test_merge_tree_error_json_has_required_fields(self) -> None: from muse.cli.commands.merge_tree import _MergeTreeErrorJson hints = get_type_hints(_MergeTreeErrorJson) for field in ("status", "error", "exit_code"): assert field in hints, f"Missing annotation: {field!r}" # --------------------------------------------------------------------------- # Docstring coverage # --------------------------------------------------------------------------- class TestDocstring: def _doc(self) -> str: import muse.cli.commands.merge_tree as mod return mod.__doc__ or "" def test_docstring_documents_exit_code(self) -> None: assert "exit_code" in self._doc() def test_docstring_documents_duration_ms(self) -> None: assert "duration_ms" in self._doc() # --------------------------------------------------------------------------- # Stress # --------------------------------------------------------------------------- class TestStress: def test_100_files_40_pct_conflicts(self, tmp_path: pathlib.Path) -> None: root = _init_repo(tmp_path) n = 100 conflict_n = 40 base_manifest = {f"f{i:03d}.py": _write_obj(root, f"base-{i}".encode()) for i in range(n)} base_cid = _make_commit(root, base_manifest, "main", msg="base") a_manifest = dict(base_manifest) for i in range(conflict_n): a_manifest[f"f{i:03d}.py"] = _write_obj(root, f"a-{i}".encode()) _make_commit(root, a_manifest, "stress-a", parent=base_cid) b_manifest = dict(base_manifest) for i in range(conflict_n): b_manifest[f"f{i:03d}.py"] = _write_obj(root, f"b-{i}".encode()) _make_commit(root, b_manifest, "stress-b", parent=base_cid) r = _mt(root, "stress-a", "stress-b", "--json") assert r.exit_code != 0 d = json.loads(r.output) assert len(d["conflicts"]) == conflict_n assert d["exit_code"] != 0 assert "duration_ms" in d # --------------------------------------------------------------------------- # TestRegisterFlags — argparse-level verification # --------------------------------------------------------------------------- class TestRegisterFlags: """Verify that register() wires --json / -j correctly.""" def _make_parser(self) -> "argparse.ArgumentParser": import argparse from muse.cli.commands.merge_tree import register ap = argparse.ArgumentParser() subs = ap.add_subparsers() register(subs) return ap def test_json_flag_long(self) -> None: ns = self._make_parser().parse_args(["merge-tree", "feat/x", "dev", "--json"]) assert ns.json_out is True def test_j_alias(self) -> None: ns = self._make_parser().parse_args(["merge-tree", "feat/x", "dev", "-j"]) assert ns.json_out is True def test_default_is_text(self) -> None: ns = self._make_parser().parse_args(["merge-tree", "feat/x", "dev"]) assert ns.json_out is False def test_dest_is_json_out(self) -> None: ns = self._make_parser().parse_args(["merge-tree", "feat/x", "dev", "-j"]) assert hasattr(ns, "json_out") assert not hasattr(ns, "fmt")