"""Supercharge tests for ``muse merge``. Coverage tiers -------------- - JSON envelope: exit_code and duration_ms always present on all outcome types (merged, fast_forward, up_to_date, conflict) - Error payload: errors go to stdout as JSON in --json mode, no dual stderr prose - TypedDicts: _MergeJson and _MergeErrorJson exist with required annotations - Docstring: module docstring covers exit_code and duration_ms - No-prose pollution: no emoji in JSON stdout on success paths """ from __future__ import annotations from collections.abc import Mapping 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, fake_id from muse.core.paths import heads_dir, muse_dir, ref_path runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Mapping[str, str]: return {"MUSE_REPO_ROOT": str(root)} def _init_repo(tmp_path: pathlib.Path) -> tuple[pathlib.Path, str]: 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 def _make_commit( root: pathlib.Path, repo_id: str, branch: str = "main", message: str = "test", manifest: Mapping[str, object] | None = None, ) -> str: ref_file = ref_path(root, branch) parent_id = ref_file.read_text().strip() if ref_file.exists() else None m = manifest or {} snap_id = compute_snapshot_id(m) committed_at = datetime.datetime.now(datetime.timezone.utc) commit_id = compute_commit_id( parent_ids=[parent_id] if parent_id else [], snapshot_id=snap_id, message=message, committed_at_iso=committed_at.isoformat(), ) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=m)) 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, )) ref_file.parent.mkdir(parents=True, exist_ok=True) ref_file.write_text(commit_id, encoding="utf-8") return commit_id def _write_obj(root: pathlib.Path, content: bytes) -> str: oid = blob_id(content) write_object(root, oid, content) return oid def _merge(root: pathlib.Path, *args: str) -> InvokeResult: from muse.cli.app import main as cli return runner.invoke(cli, ["merge", *args], env=_env(root)) # --------------------------------------------------------------------------- # Repo fixtures # --------------------------------------------------------------------------- def _up_to_date_repo(tmp_path: pathlib.Path) -> pathlib.Path: root, repo_id = _init_repo(tmp_path) cid = _make_commit(root, repo_id, branch="main", message="base") (heads_dir(root) / "feature").write_text(cid) return root def _ff_repo(tmp_path: pathlib.Path) -> pathlib.Path: root, repo_id = _init_repo(tmp_path) base = _make_commit(root, repo_id, branch="main", message="base") (heads_dir(root) / "feature").write_text(base) obj = _write_obj(root, b"new file") _make_commit(root, repo_id, branch="feature", message="feat", manifest={"new.py": obj}) return root def _three_way_clean_repo(tmp_path: pathlib.Path) -> pathlib.Path: root, repo_id = _init_repo(tmp_path) base_obj = _write_obj(root, b"base") base = _make_commit(root, repo_id, branch="main", message="base", manifest={"base.py": base_obj}) (heads_dir(root) / "feature").write_text(base) main_obj = _write_obj(root, b"main addition") _make_commit(root, repo_id, branch="main", message="main work", manifest={"base.py": base_obj, "main.py": main_obj}) feat_obj = _write_obj(root, b"feat addition") _make_commit(root, repo_id, branch="feature", message="feat work", manifest={"base.py": base_obj, "feat.py": feat_obj}) # Write working tree to match main HEAD so require_clean_workdir passes. (root / "base.py").write_bytes(b"base") (root / "main.py").write_bytes(b"main addition") return root def _conflict_repo(tmp_path: pathlib.Path) -> pathlib.Path: root, repo_id = _init_repo(tmp_path) shared_v1 = _write_obj(root, b"shared v1") base = _make_commit(root, repo_id, branch="main", message="base", manifest={"shared.py": shared_v1}) (heads_dir(root) / "feature").write_text(base) shared_main = _write_obj(root, b"shared main version") _make_commit(root, repo_id, branch="main", message="main mod", manifest={"shared.py": shared_main}) shared_feat = _write_obj(root, b"shared feature version") _make_commit(root, repo_id, branch="feature", message="feat mod", manifest={"shared.py": shared_feat}) # Write working tree to match main HEAD so require_clean_workdir passes. (root / "shared.py").write_bytes(b"shared main version") return root # --------------------------------------------------------------------------- # JSON envelope — exit_code and duration_ms on all outcomes # --------------------------------------------------------------------------- class TestJsonEnvelopeExitCode: """exit_code is present and correct across all merge outcome types.""" def test_up_to_date_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _up_to_date_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "exit_code" in d, "exit_code missing from up_to_date envelope" assert d["exit_code"] == 0 def test_fast_forward_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _ff_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "exit_code" in d, "exit_code missing from fast_forward envelope" assert d["exit_code"] == 0 def test_three_way_clean_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _three_way_clean_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "exit_code" in d, "exit_code missing from merged envelope" assert d["exit_code"] == 0 def test_conflict_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _conflict_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "exit_code" in d, "exit_code missing from conflict envelope" assert d["exit_code"] != 0 def test_dry_run_merged_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _three_way_clean_repo(tmp_path) r = _merge(root, "feature", "--dry-run", "--json") d = json.loads(r.output) assert "exit_code" in d assert d["exit_code"] == 0 def test_dry_run_conflict_has_exit_code(self, tmp_path: pathlib.Path) -> None: root = _conflict_repo(tmp_path) r = _merge(root, "feature", "--dry-run", "--json") d = json.loads(r.output) assert "exit_code" in d assert d["exit_code"] != 0 class TestJsonEnvelopeDurationMs: """duration_ms is present and is a non-negative float on all outcome types.""" def test_up_to_date_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _up_to_date_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "duration_ms" in d assert isinstance(d["duration_ms"], float) assert d["duration_ms"] >= 0.0 def test_fast_forward_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _ff_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "duration_ms" in d assert isinstance(d["duration_ms"], float) def test_three_way_clean_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _three_way_clean_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "duration_ms" in d assert isinstance(d["duration_ms"], float) def test_conflict_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _conflict_repo(tmp_path) r = _merge(root, "feature", "--json") d = json.loads(r.output) assert "duration_ms" in d assert isinstance(d["duration_ms"], float) def test_dry_run_has_duration_ms(self, tmp_path: pathlib.Path) -> None: root = _ff_repo(tmp_path) r = _merge(root, "feature", "--dry-run", "--json") d = json.loads(r.output) assert "duration_ms" in d # --------------------------------------------------------------------------- # Error payload — errors go to stdout as JSON in --json mode # --------------------------------------------------------------------------- class TestErrorPayload: """Errors emit {status: "error", error: "...", exit_code: N} on stdout in --json mode.""" def test_no_branch_error_is_json(self, tmp_path: pathlib.Path) -> None: root, _ = _init_repo(tmp_path) r = _merge(root, "--json") # no branch arg assert r.exit_code != 0 d = json.loads(r.output) assert d["status"] == "error" def test_self_merge_error_is_json(self, tmp_path: pathlib.Path) -> None: root, repo_id = _init_repo(tmp_path) _make_commit(root, repo_id, branch="main") r = _merge(root, "main", "--json") assert r.exit_code != 0 d = json.loads(r.output) assert d["status"] == "error" def test_self_merge_no_duplicate_stderr_prose(self, tmp_path: pathlib.Path) -> None: """In --json mode, errors should not also print emoji prose to stderr.""" root, repo_id = _init_repo(tmp_path) _make_commit(root, repo_id, branch="main") r = _merge(root, "main", "--json") assert "❌" not in r.stderr def test_error_payload_has_status_error(self, tmp_path: pathlib.Path) -> None: root, _ = _init_repo(tmp_path) r = _merge(root, "--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 = _merge(root, "--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 = _merge(root, "--json") d = json.loads(r.output) assert "error" in d assert d["error"] # non-empty message def test_unknown_branch_error_is_json(self, tmp_path: pathlib.Path) -> None: root, repo_id = _init_repo(tmp_path) _make_commit(root, repo_id, branch="main") r = _merge(root, "no-such-branch", "--json") assert r.exit_code != 0 d = json.loads(r.output) assert d["status"] == "error" # --------------------------------------------------------------------------- # No-prose pollution # --------------------------------------------------------------------------- class TestNoProsePollution: def test_up_to_date_stdout_valid_json(self, tmp_path: pathlib.Path) -> None: root = _up_to_date_repo(tmp_path) r = _merge(root, "feature", "--json") json.loads(r.output) # must not raise def test_fast_forward_stdout_valid_json(self, tmp_path: pathlib.Path) -> None: root = _ff_repo(tmp_path) r = _merge(root, "feature", "--json") json.loads(r.output) def test_merged_stdout_valid_json(self, tmp_path: pathlib.Path) -> None: root = _three_way_clean_repo(tmp_path) r = _merge(root, "feature", "--json") json.loads(r.output) def test_no_emoji_in_merged_json(self, tmp_path: pathlib.Path) -> None: root = _three_way_clean_repo(tmp_path) r = _merge(root, "feature", "--json") assert "✅" not in r.output assert "❌" not in r.output # --------------------------------------------------------------------------- # TypedDicts # --------------------------------------------------------------------------- class TestTypedDicts: def test_merge_json_typeddict_exists(self) -> None: from muse.cli.commands.merge import _MergeJson assert _MergeJson is not None def test_merge_error_json_typeddict_exists(self) -> None: from muse.cli.commands.merge import _MergeErrorJson assert _MergeErrorJson is not None def test_merge_json_has_exit_code_annotation(self) -> None: from muse.cli.commands.merge import _MergeJson hints = get_type_hints(_MergeJson) assert "exit_code" in hints def test_merge_json_has_duration_ms_annotation(self) -> None: from muse.cli.commands.merge import _MergeJson hints = get_type_hints(_MergeJson) assert "duration_ms" in hints def test_merge_error_json_has_required_fields(self) -> None: from muse.cli.commands.merge import _MergeErrorJson hints = get_type_hints(_MergeErrorJson) 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 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()