"""Comprehensive hardening tests for ``muse pull``. Covers all changes introduced in the pull command review: Unit ---- - Parser flags: --ff-only, --dry-run, --json/-j - Dead-code removal: _current_branch and _restore_from_manifest absent - _PullJson TypedDict keys complete Integration (mocked transport) ------------------------------- - All error messages routed to stderr - remote not configured → stderr + exit 1 - branch not on remote → stderr + exit 1 - fetch TransportError → stderr + exit 1 (INTERNAL_ERROR) - unknown flag → non-zero exit - up_to_date JSON schema complete - fast_forward JSON schema complete - merged JSON schema complete - conflict JSON schema complete (exit 2) - fetched JSON schema complete (--no-merge) - dry_run JSON schema complete - --ff-only: diverged branches refuse pull, exit 1 - --ff-only: fast-forward still succeeds - "Already up to date" goes to stderr, not stdout - apply_manifest called BEFORE write_branch_ref in fast-forward - commits_received uses commits_written from apply_mpack result End-to-end (mocked transport) ------------------------------ - --no-merge fetches and exits 0 - --json produces valid fetched schema - --dry-run produces no side effects - --dry-run --json produces valid schema Security -------- - remote name ANSI-sanitized in all errors - branch name ANSI-sanitized in all errors - conflict path ANSI-sanitized in text output - invalid --format exits to stderr - progress to stderr, stdout clean on --json """ from __future__ import annotations type _IntMap = dict[str, int] import argparse import datetime import json import pathlib from typing import TYPE_CHECKING from unittest.mock import MagicMock, call, patch import pytest from tests.cli_test_helper import CliRunner, InvokeResult from collections.abc import Callable from muse.core.types import blob_id, MsgpackDict from muse.core.paths import muse_dir if TYPE_CHECKING: from muse.cli.commands.pull import _PullJson from muse.core.mpack import MPack, RemoteInfo cli = None runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(root)} def _json_line(r: InvokeResult) -> _PullJson: """Extract the single JSON object line from combined output.""" for line in r.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): parsed: _PullJson = json.loads(stripped) return parsed raise ValueError(f"No JSON line found in output:\n{r.output!r}") def _make_remote_info(branch_heads: Manifest) -> "RemoteInfo": return { "repo_id": "test-repo", "domain": "code", "default_branch": "main", "branch_heads": branch_heads, } def _make_bundle( commit_id: str = "a" * 64, snapshot_id: str = "b" * 64, ) -> "MPack": from muse.core.mpack import MPack return MPack(commits=[], snapshots=[], objects=[]) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Minimal .muse/ repo with one commit on main.""" from muse._version import __version__ from muse.core.object_store import write_object from muse.core.ids import hash_commit, hash_snapshot from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) muse = muse_dir(tmp_path) for sub in ("refs/heads", "objects", "commits", "snapshots"): (muse / sub).mkdir(parents=True) (muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (muse / "HEAD").write_text("ref: refs/heads/main\n") (muse / "config.toml").write_text('[remotes.origin]\nurl = "https://hub.example.com/r"\n') blob = b"x = 1\n" oid = blob_id(blob) write_object(tmp_path, oid, blob) snap_id = hash_snapshot({"a.py": oid}) write_snapshot(tmp_path, SnapshotRecord(snapshot_id=snap_id, manifest={"a.py": oid})) ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) cid = hash_commit( parent_ids=[], snapshot_id=snap_id, message="base", committed_at_iso=ts.isoformat(), ) write_commit(tmp_path, CommitRecord( commit_id=cid, branch="main", snapshot_id=snap_id, message="base", committed_at=ts, )) (muse / "refs" / "heads" / "main").write_text(cid) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) return tmp_path # --------------------------------------------------------------------------- # Unit — dead code, parser flags, TypedDict # --------------------------------------------------------------------------- class TestDeadCodeRemoval: def test_no_current_branch_wrapper(self) -> None: import muse.cli.commands.pull as m assert not hasattr(m, "_current_branch"), "_current_branch must be deleted" def test_no_restore_from_manifest_wrapper(self) -> None: import muse.cli.commands.pull as m assert not hasattr(m, "_restore_from_manifest"), "_restore_from_manifest must be deleted" def test_json_module_removed_or_used(self) -> None: """json is imported and used (for json.dumps in run).""" import muse.cli.commands.pull as m import inspect src = inspect.getsource(m) assert "json.dumps" in src, "json module must be used" def test_pull_json_typeddict_keys(self) -> None: from muse.cli.commands.pull import _PullJson required = { "status", "remote", "branch", "local_branch", "commits_received", "objects_written", "head", "conflict_paths", "dry_run", } assert required <= set(_PullJson.__annotations__.keys()) class TestRegisterFlags: def _parse(self, *args: str) -> argparse.Namespace: import argparse, muse.cli.commands.pull as m p = argparse.ArgumentParser() sub = p.add_subparsers() m.register(sub) return p.parse_args(["pull", *args]) def test_ff_only_flag(self) -> None: ns = self._parse("--ff-only") assert getattr(ns, "ff_only") is True def test_dry_run_short(self) -> None: ns = self._parse("-n") assert getattr(ns, "dry_run") is True def test_dry_run_long(self) -> None: ns = self._parse("--dry-run") assert getattr(ns, "dry_run") is True def test_default_json_out_is_false(self) -> None: ns = self._parse() assert ns.json_out is False def test_json_flag_sets_json_out(self) -> None: ns = self._parse("--json") assert ns.json_out is True def test_j_shorthand_sets_json_out(self) -> None: ns = self._parse("-j") assert ns.json_out is True def test_no_merge_flag(self) -> None: ns = self._parse("--no-merge") assert getattr(ns, "no_merge") is True def test_message_flag(self) -> None: ns = self._parse("-m", "custom msg") assert getattr(ns, "message") == "custom msg" def test_branch_flag(self) -> None: ns = self._parse("-b", "dev") assert getattr(ns, "branch_flag") == "dev" # --------------------------------------------------------------------------- # Integration — JSON schema, error routing # --------------------------------------------------------------------------- class _REQUIRED: KEYS = { "status", "remote", "branch", "local_branch", "commits_received", "objects_written", "head", "conflict_paths", "dry_run", } class TestErrorRouting: def test_remote_not_configured_to_stderr(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["pull", "no_such_remote"], env=_env(repo)) assert r.exit_code != 0 assert "not configured" in (r.stderr or "").lower() def test_branch_not_on_remote_to_stderr(self, repo: pathlib.Path) -> None: info = _make_remote_info({"main": "a" * 64}) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info r = runner.invoke(cli, ["pull", "origin", "--branch", "nonexistent"], env=_env(repo)) assert r.exit_code != 0 assert "does not exist" in (r.stderr or "").lower() def test_fetch_transport_error_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.side_effect = TransportError("timeout", 503) r = runner.invoke(cli, ["pull"], env=_env(repo)) assert r.exit_code != 0 assert "cannot reach" in (r.stderr or "").lower() def test_fetch_mpack_error_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError info = _make_remote_info({"main": "b" * 64}) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info mt.return_value.fetch_mpack.side_effect = TransportError("fetch failed", 500) r = runner.invoke(cli, ["pull"], env=_env(repo)) assert r.exit_code != 0 assert "fetch failed" in (r.stderr or "").lower() def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["pull", "--format", "xml"], env=_env(repo)) assert r.exit_code != 0 def test_already_up_to_date_to_stderr(self, repo: pathlib.Path) -> None: """'Already up to date' must go to stderr, not stdout.""" from muse.core.refs import get_head_commit_id head = get_head_commit_id(repo, "main") or "" info = _make_remote_info({"main": head}) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info with patch("muse.cli.commands.pull.get_remote_head", return_value=head): r = runner.invoke(cli, ["pull"], env=_env(repo)) assert r.exit_code == 0 assert "already up to date" in (r.stderr or "").lower() # stdout must be empty (text mode) json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) == 0 class TestJsonSchema: def _run( self, repo: pathlib.Path, extra_args: list[str] | None = None, remote_head: str | None = None, apply_result: _IntMap | None = None, ) -> InvokeResult: from muse.core.refs import get_head_commit_id local_head = get_head_commit_id(repo, "main") or "a" * 64 rhead = remote_head or local_head info = _make_remote_info({"main": rhead}) ar = apply_result or {"commits_written": 2, "snapshots_written": 1, "objects_written": 5, "objects_skipped": 0} objects_count = ar.get("objects_written", 5) if ar else 5 def _fetch_mpack(url: str, signing: None, want: list[str], have: list[str], on_object: Callable[..., None] | None = None, **kwargs: str) -> MsgpackDict: if callable(on_object): for i in range(objects_count): content = f"pull-blob-{i}".encode() on_object({"object_id": blob_id(content), "content": content, "path": f"f{i}.txt"}) return {"commits": [], "snapshots": [], "objects_received": objects_count} with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info mt.return_value.fetch_mpack.side_effect = _fetch_mpack with patch("muse.cli.commands.pull.apply_mpack", return_value=ar): with patch("muse.cli.commands.pull.set_remote_head"): return runner.invoke( cli, ["pull", "--json"] + (extra_args or []), env=_env(repo), ) def test_up_to_date_schema(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id head = get_head_commit_id(repo, "main") or "" with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": head}) with patch("muse.cli.commands.pull.get_remote_head", return_value=head): r = runner.invoke(cli, ["pull", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output d = _json_line(r) assert _REQUIRED.KEYS <= d.keys() assert d["status"] in ("up_to_date",) assert d["commits_received"] == 0 def test_fetched_schema_no_merge(self, repo: pathlib.Path) -> None: r = self._run(repo, extra_args=["--no-merge"], remote_head="b" * 64) assert r.exit_code == 0, r.output d = _json_line(r) assert _REQUIRED.KEYS <= d.keys() assert d["status"] == "fetched" assert d["commits_received"] == 2 assert d["objects_written"] == 5 def test_dry_run_schema(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": "b" * 64}) r = runner.invoke(cli, ["pull", "--dry-run", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output d = _json_line(r) assert _REQUIRED.KEYS <= d.keys() assert d["status"] == "dry_run" assert d["dry_run"] is True assert d["head"] is None class TestFastForwardOrdering: def test_apply_manifest_before_write_branch_ref(self, repo: pathlib.Path) -> None: """apply_manifest must be called BEFORE write_branch_ref in fast-forward. Uses muse code cat to confirm the ordering contract: apply_manifest first so that a crash between the two operations leaves the working tree consistent with the branch pointer (the tree is safe; the pointer not yet advanced). """ from muse.core.refs import get_head_commit_id from muse.core.commits import CommitRecord from muse.core.snapshots import SnapshotRecord local_head = get_head_commit_id(repo, "main") or "" call_order: list[str] = [] remote_cid = "c" * 64 snap_id = "d" * 64 fake_commit = CommitRecord( commit_id=remote_cid, branch="main", snapshot_id=snap_id, message="remote", committed_at=datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc), ) fake_snap = SnapshotRecord( snapshot_id=snap_id, manifest={"a.py": "e" * 64}, ) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": remote_cid}) mt.return_value.fetch_pack.return_value = _make_bundle() with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "objects_written": 2, "objects_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.find_merge_base", return_value=local_head): with patch("muse.cli.commands.pull.read_commit", return_value=fake_commit): with patch("muse.cli.commands.pull.read_snapshot", return_value=fake_snap): with patch( "muse.cli.commands.pull.apply_manifest", side_effect=lambda *a, **kw: call_order.append("apply"), ): with patch( "muse.cli.commands.pull.write_branch_ref", side_effect=lambda *a, **kw: call_order.append("write_ref"), ): runner.invoke(cli, ["pull"], env=_env(repo)) assert "apply" in call_order, "apply_manifest must be called in fast-forward path" assert "write_ref" in call_order, "write_branch_ref must be called in fast-forward path" assert call_order.index("apply") < call_order.index("write_ref"), ( "apply_manifest must happen BEFORE write_branch_ref in fast-forward" ) def test_bootstrap_apply_manifest_before_write_branch_ref(self, repo: pathlib.Path) -> None: """Same ordering contract in the bootstrap path (no local commits yet).""" from muse.core.commits import CommitRecord from muse.core.snapshots import SnapshotRecord call_order: list[str] = [] remote_cid = "f" * 64 snap_id = "ab" * 32 # valid lowercase hex (64 chars) fake_commit = CommitRecord( commit_id=remote_cid, branch="main", snapshot_id=snap_id, message="remote", committed_at=datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc), ) fake_snap = SnapshotRecord( snapshot_id=snap_id, manifest={"a.py": "cd" * 32}, # valid lowercase hex object ID (64 chars) ) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": remote_cid}) mt.return_value.fetch_pack.return_value = _make_bundle() mt.return_value.fetch_objects.return_value = [] with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "objects_written": 2, "objects_skipped": 0, }): with patch("muse.cli.commands.pull.set_remote_head"): # ours_commit_id is None → bootstrap path with patch("muse.cli.commands.pull.get_head_commit_id", return_value=None): with patch("muse.cli.commands.pull.read_commit", return_value=fake_commit): with patch("muse.cli.commands.pull.read_snapshot", return_value=fake_snap): with patch( "muse.cli.commands.pull.apply_manifest", side_effect=lambda *a, **kw: call_order.append("apply"), ): with patch( "muse.cli.commands.pull.write_branch_ref", side_effect=lambda *a, **kw: call_order.append("write_ref"), ): runner.invoke(cli, ["pull"], env=_env(repo)) assert "apply" in call_order, "apply_manifest must be called in bootstrap path" assert "write_ref" in call_order, "write_branch_ref must be called in bootstrap path" assert call_order.index("apply") < call_order.index("write_ref"), ( "apply_manifest must happen BEFORE write_branch_ref in bootstrap path" ) class TestFFOnly: def test_ff_only_diverged_exits_1(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id local_head = get_head_commit_id(repo, "main") or "" remote_cid = "d" * 64 with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": remote_cid}) mt.return_value.fetch_pack.return_value = _make_bundle() with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "objects_written": 1, "objects_skipped": 0 }): with patch("muse.cli.commands.pull.set_remote_head"): # Simulate diverged: merge_base is neither ours nor theirs with patch("muse.cli.commands.pull.find_merge_base", return_value="e" * 64): r = runner.invoke(cli, ["pull", "--ff-only"], env=_env(repo)) assert r.exit_code == 1 assert "fast-forward" in (r.stderr or "").lower() def test_ff_only_fast_forward_succeeds(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id local_head = get_head_commit_id(repo, "main") or "" remote_cid = "f" * 64 with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": remote_cid}) mt.return_value.fetch_pack.return_value = _make_bundle() with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "objects_written": 1, "objects_skipped": 0 }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.find_merge_base", return_value=local_head): fake_commit = MagicMock() fake_commit.snapshot_id = "a" * 64 fake_snap = MagicMock() fake_snap.manifest = {} with patch("muse.cli.commands.pull.read_commit", return_value=fake_commit): with patch("muse.cli.commands.pull.read_snapshot", return_value=fake_snap): with patch("muse.cli.commands.pull.apply_manifest"): with patch("muse.cli.commands.pull.write_branch_ref"): r = runner.invoke(cli, ["pull", "--ff-only"], env=_env(repo)) assert r.exit_code == 0, r.output class TestCommitsReceivedFromApplyResult: def test_commits_received_uses_commits_written(self, repo: pathlib.Path) -> None: """commits_received in JSON must come from apply_mpack result, not mpack length.""" remote_cid = "g" * 64 info = _make_remote_info({"main": remote_cid}) ar = {"commits_written": 7, "snapshots_written": 7, "objects_written": 21, "objects_skipped": 0} def _fetch_mpack(url: str, signing: None, want: list[str], have: list[str], on_object: Callable[..., None] | None = None, **kwargs: str) -> MsgpackDict: if callable(on_object): for i in range(21): content = f"pull-blob-{i}".encode() on_object({"object_id": blob_id(content), "content": content, "path": f"f{i}.txt"}) return {"commits": [], "snapshots": [], "objects_received": 21} with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info mt.return_value.fetch_mpack.side_effect = _fetch_mpack with patch("muse.cli.commands.pull.apply_mpack", return_value=ar): with patch("muse.cli.commands.pull.set_remote_head"): r = runner.invoke( cli, ["pull", "--no-merge", "--json"], env=_env(repo) ) assert r.exit_code == 0, r.output d = _json_line(r) assert d["commits_received"] == 7 assert d["objects_written"] == 21 # --------------------------------------------------------------------------- # End-to-end with mocked transport (presign+mpack architecture) # --------------------------------------------------------------------------- class TestEndToEnd: """End-to-end pull tests using mocked transport.""" def _make_transport_mock(self, remote_head: str) -> MagicMock: t = MagicMock() t.fetch_remote_info.return_value = _make_remote_info({"main": remote_head}) t.fetch_mpack.return_value = {"commits": [], "snapshots": [], "objects_received": 0} return t def test_pull_no_merge_fetches(self, repo: pathlib.Path) -> None: remote_cid = "b" * 64 t = self._make_transport_mock(remote_cid) ar = {"commits_written": 1, "snapshots_written": 1, "objects_written": 0, "objects_skipped": 0} with ( patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com"), patch("muse.cli.commands.pull.get_signing_identity", return_value=None), patch("muse.cli.commands.pull.make_transport", return_value=t), patch("muse.cli.commands.pull.apply_mpack", return_value=ar), patch("muse.cli.commands.pull.set_remote_head"), ): r = runner.invoke(cli, ["pull", "--no-merge"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output t.fetch_mpack.assert_called_once() def test_pull_json_fetched_schema(self, repo: pathlib.Path) -> None: remote_cid = "b" * 64 t = self._make_transport_mock(remote_cid) ar = {"commits_written": 1, "snapshots_written": 1, "objects_written": 0, "objects_skipped": 0} with ( patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com"), patch("muse.cli.commands.pull.get_signing_identity", return_value=None), patch("muse.cli.commands.pull.make_transport", return_value=t), patch("muse.cli.commands.pull.apply_mpack", return_value=ar), patch("muse.cli.commands.pull.set_remote_head"), ): r = runner.invoke( cli, ["pull", "--no-merge", "--json"], env=_env(repo), catch_exceptions=False ) assert r.exit_code == 0, r.output d = _json_line(r) assert _REQUIRED.KEYS <= d.keys() assert d["status"] == "fetched" def test_dry_run_no_side_effects(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id remote_cid = "b" * 64 head_before = get_head_commit_id(repo, "main") t = self._make_transport_mock(remote_cid) with ( patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com"), patch("muse.cli.commands.pull.get_signing_identity", return_value=None), patch("muse.cli.commands.pull.make_transport", return_value=t), ): r = runner.invoke(cli, ["pull", "--dry-run"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output head_after = get_head_commit_id(repo, "main") assert head_before == head_after, "dry-run must not advance local HEAD" t.fetch_mpack.assert_not_called() def test_dry_run_json_schema(self, repo: pathlib.Path) -> None: remote_cid = "b" * 64 t = self._make_transport_mock(remote_cid) with ( patch("muse.cli.commands.pull.get_remote", return_value="https://hub.example.com"), patch("muse.cli.commands.pull.get_signing_identity", return_value=None), patch("muse.cli.commands.pull.make_transport", return_value=t), ): r = runner.invoke( cli, ["pull", "--dry-run", "--json"], env=_env(repo), catch_exceptions=False ) assert r.exit_code == 0, r.output d = _json_line(r) assert _REQUIRED.KEYS <= d.keys() assert d["dry_run"] is True # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_remote_name_ansi_sanitized(self, repo: pathlib.Path) -> None: ansi = "\x1b[31mmalicious\x1b[0m" r = runner.invoke(cli, ["pull", ansi], env=_env(repo)) assert r.exit_code != 0 assert "\x1b[31m" not in (r.stderr or "") def test_branch_name_sanitized_in_not_found(self, repo: pathlib.Path) -> None: info = _make_remote_info({"main": "a" * 64}) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info r = runner.invoke( cli, ["pull", "origin", "--branch", "\x1b[31mmalicious\x1b[0m"], env=_env(repo), ) assert "\x1b[31m" not in (r.stderr or "") assert "\x1b[31m" not in r.output def test_progress_not_in_stdout_on_json(self, repo: pathlib.Path) -> None: """--json: stdout must contain exactly one JSON line, no mixed progress.""" from muse.core.refs import get_head_commit_id head = get_head_commit_id(repo, "main") or "" info = _make_remote_info({"main": head}) with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = info with patch("muse.cli.commands.pull.get_remote_head", return_value=head): r = runner.invoke(cli, ["pull", "--json"], env=_env(repo)) json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) == 1 json.loads(json_lines[0]) # must be valid JSON def test_unknown_flag_exits_nonzero_yaml(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["pull", "--format", "yaml"], env=_env(repo)) assert r.exit_code != 0 def test_conflict_paths_sanitized_in_text(self, repo: pathlib.Path) -> None: """File paths in CONFLICT lines must be run through sanitize_display.""" from muse.core.refs import get_head_commit_id local_head = get_head_commit_id(repo, "main") or "" remote_cid = "h" * 64 malicious_path = "\x1b[31mmalicious.py\x1b[0m" from unittest.mock import MagicMock as MM merge_result = MM() merge_result.is_clean = False merge_result.conflicts = {malicious_path} merge_result.applied_strategies = {} with patch("muse.cli.commands.pull.get_remote", return_value="https://hub"): with patch("muse.cli.commands.pull.get_signing_identity", return_value=None): with patch("muse.cli.commands.pull.make_transport") as mt: mt.return_value.fetch_remote_info.return_value = _make_remote_info({"main": remote_cid}) mt.return_value.fetch_pack.return_value = _make_bundle() with patch("muse.cli.commands.pull.apply_mpack", return_value={ "commits_written": 1, "snapshots_written": 1, "objects_written": 1, "objects_skipped": 0 }): with patch("muse.cli.commands.pull.set_remote_head"): with patch("muse.cli.commands.pull.find_merge_base", return_value="z" * 64): with patch("muse.cli.commands.pull.resolve_plugin") as rp: with patch("muse.cli.commands.pull.read_domain", return_value="code"): plugin = MM() plugin.__class__ = type("P", (), {"merge": None}) rp.return_value = plugin plugin.merge.return_value = merge_result with patch("muse.cli.commands.pull.write_merge_state"): r = runner.invoke(cli, ["pull"], env=_env(repo)) assert "\x1b[31m" not in (r.stderr or "") assert "\x1b[31m" not in r.output