"""Comprehensive hardening tests for ``muse push``. Covers all changes introduced in the push command review: Unit ---- - Parser flags: --dry-run, --workers, --json/-j - Dead-code removal: _current_branch absent - _all_known_have_anchors: symlink skipping, binary-file safety, missing dir - _upload_chunk: progress goes to stderr, not stdout - _PushJson TypedDict keys complete Integration (with mocked transport) ------------------------------------ - Error messages routed to stderr, stdout clean on errors - remote not configured → stderr - branch has no commits → stderr - push rejected (result.ok=False) → stderr - up_to_date JSON schema complete - pushed JSON schema complete - dry_run JSON schema complete - deleted JSON schema complete - --dry-run: no transport calls, correct counts - --workers accepted without error - --set-upstream records tracking ref - 409/401/404/other TransportError → stderr + exit 1 End-to-end (local:// transport) --------------------------------- - Fresh push succeeds - Second push (up_to_date) exits 0 - --dry-run shows would-push info without writing - --json produces valid JSON - --force bypasses fast-forward check Security -------- - remote name sanitized in all error messages - branch name sanitized in delete output - del_branch sanitized in already-gone path - _all_known_have_anchors: planted symlink skipped - _all_known_have_anchors: binary file skipped - unknown flag exits non-zero - progress prints from _upload_chunk go to stderr Stress ------ - _push_objects_parallel with 1000 objects (mocked transport) - concurrent push runs to isolated repos """ from __future__ import annotations type _IntMap = dict[str, int] import argparse import http.client import inspect import json import os import pathlib import tempfile import threading import time import types import urllib.error import urllib.request from typing import TYPE_CHECKING from unittest.mock import MagicMock, patch import pytest from muse.cli.config import set_remote from muse.core.mpack import RemoteInfo from muse.core.paths import config_toml_path, remotes_dir from muse.core.types import long_id, Manifest from tests.cli_test_helper import CliRunner, InvokeResult if TYPE_CHECKING: import httpx from muse.cli.commands.push import _PushJson from muse.core.mpack import BlobPayload, CommitDict, RemoteInfo, SnapshotDict from muse.core.transport import PushResult, SigningIdentity _Headers = dict[str, str] # HTTP header map _KwVal = str | bool | int | None # generic keyword argument value cli = None runner = CliRunner() class _FakeResponse: """Minimal context-manager stub returned by fake urlopen in tests.""" def __enter__(self) -> "_FakeResponse": return self def __exit__( self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: "types.TracebackType | None", ) -> None: pass # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- def _env(root: pathlib.Path) -> Manifest: return {"MUSE_REPO_ROOT": str(root)} def _json(r: InvokeResult) -> _PushJson: """Extract the JSON object line from combined output. With ``--json``, exactly one line starting with ``{`` is emitted to stdout; all progress/error lines go to stderr and are prefixed with spaces or emoji. This helper finds that line so tests can assert on the schema. """ for line in r.output.splitlines(): stripped = line.strip() if stripped.startswith("{"): raw: _PushJson = json.loads(stripped) return raw raise ValueError(f"No JSON line found in output:\n{r.output!r}") @pytest.fixture() def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Fresh repo with one committed file.""" monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output (tmp_path / "a.py").write_text("x = 1\n") r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False) assert r.exit_code == 0, r.output return tmp_path @pytest.fixture() def remote_repo( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch, ) -> tuple[pathlib.Path, pathlib.Path]: """Return ``(local, remote)`` pair with the local remote configured.""" local = tmp_path / "local" remote = tmp_path / "remote" local.mkdir() remote.mkdir() # muse init uses cwd; chdir so it creates .muse/ in the right place. monkeypatch.chdir(local) monkeypatch.setenv("MUSE_REPO_ROOT", str(local)) runner.invoke(cli, ["init"], env=_env(local), catch_exceptions=False) (local / "a.py").write_text("x = 1\n") runner.invoke(cli, ["commit", "-m", "base"], env=_env(local), catch_exceptions=False) monkeypatch.chdir(remote) monkeypatch.setenv("MUSE_REPO_ROOT", str(remote)) runner.invoke(cli, ["init"], env=_env(remote), catch_exceptions=False) monkeypatch.chdir(local) monkeypatch.setenv("MUSE_REPO_ROOT", str(local)) # Write the remote config directly — muse remote add blocks file:// by # design (security); set_remote() bypasses that validation intentionally # for test infrastructure. set_remote("local", f"file://{remote}", repo_root=local) return local, remote # --------------------------------------------------------------------------- # Unit — dead code, parser flags, helpers # --------------------------------------------------------------------------- class TestDeadCodeRemoval: def test_no_current_branch_wrapper(self) -> None: import muse.cli.commands.push as m assert not hasattr(m, "_current_branch"), "_current_branch must be deleted" def test_push_json_typeddict_keys(self) -> None: import muse.cli.commands.push as m required = {"status", "remote", "branch", "head", "commits_sent", "objects_sent", "force", "dry_run"} assert required <= set(m._PushJson.__annotations__.keys()) class TestRegisterFlags: def _parse(self, *args: str) -> argparse.Namespace: import muse.cli.commands.push as m p = argparse.ArgumentParser() sub = p.add_subparsers() m.register(sub) return p.parse_args(["push", *args]) def test_dry_run_short(self) -> None: ns = self._parse("-n") assert ns.dry_run is True def test_dry_run_long(self) -> None: ns = self._parse("--dry-run") assert ns.dry_run is True def test_workers_default(self) -> None: ns = self._parse() assert ns.workers == 16 def test_workers_custom(self) -> None: ns = self._parse("--workers", "8") assert ns.workers == 8 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_force_flag(self) -> None: ns = self._parse("--force") assert ns.force is True def test_delete_flag(self) -> None: ns = self._parse("--delete") assert ns.delete_branch is True def test_set_upstream_short(self) -> None: ns = self._parse("-u") assert ns.set_upstream_flag is True class TestAllKnownHaveAnchors: def test_no_remotes_dir_returns_empty(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors assert _all_known_have_anchors(tmp_path) == [] def test_reads_commit_ids(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = remotes_dir(tmp_path) / "origin" remotes.mkdir(parents=True) cid = long_id("a" * 64) (remotes / "main").write_text(cid + "\n") result = _all_known_have_anchors(tmp_path) assert cid in result def test_symlinks_are_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = remotes_dir(tmp_path) / "origin" remotes.mkdir(parents=True) target = tmp_path / "secret.txt" target.write_text("abc123\n") (remotes / "main").symlink_to(target) result = _all_known_have_anchors(tmp_path) # Symlink should not be followed — abc123 should NOT appear assert "abc123" not in result def test_binary_file_skipped_not_crashed(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = remotes_dir(tmp_path) / "origin" remotes.mkdir(parents=True) (remotes / "bin_ref").write_bytes(b"\x00\x01\x02\xff") # Should not raise result = _all_known_have_anchors(tmp_path) # Binary content with \x00 stripped by errors='ignore' → not a valid ID assert isinstance(result, list) def test_empty_files_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = remotes_dir(tmp_path) / "origin" remotes.mkdir(parents=True) (remotes / "empty").write_text("") result = _all_known_have_anchors(tmp_path) assert result == [] def test_multiple_remotes(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors cids = {name: long_id(c * 64) for name, c in zip(["origin", "upstream", "fork"], "abc")} for name, cid in cids.items(): d = remotes_dir(tmp_path) / name d.mkdir(parents=True) (d / "main").write_text(cid + "\n") result = _all_known_have_anchors(tmp_path) assert len(result) == 3 assert cids["origin"] in result # --------------------------------------------------------------------------- # Integration — JSON schema and error routing (mocked transport) # --------------------------------------------------------------------------- class _FakeTransport: """Minimal mock transport for unit-level integration tests.""" def __init__( self, remote_head: str | None = None, push_ok: bool = True, push_exc: Exception | None = None, ) -> None: self._remote_head = remote_head self._push_ok = push_ok self._push_exc = push_exc def fetch_remote_info(self, url: str, token: str | None) -> "RemoteInfo": from muse.core.mpack import RemoteInfo return RemoteInfo( repo_id="test-repo", domain="code", branch_heads={"main": self._remote_head} if self._remote_head else {}, default_branch="main", ) def _build_request(self, method: str, url: str, signing: "SigningIdentity | None", body: bytes, **kw: _KwVal) -> MagicMock: req = MagicMock() req.headers = {"Authorization": "MSign stub", "Content-Type": "application/x-msgpack"} return req def push_mpack_presign(self, url: str, signing: "SigningIdentity | None", mpack_bytes: bytes, ttl_seconds: int = 3600) -> "dict[str, str]": if self._push_exc is not None: raise self._push_exc return {"upload_url": "https://minio.example.com/put?sig=x", "mpack_key": "sha256:fake"} def push_mpack_put(self, upload_url: str, mpack_bytes: bytes, mpack_key: str = "") -> None: if self._push_exc is not None: raise self._push_exc def push_mpack_unpack(self, url: str, signing: "SigningIdentity | None", mpack_key: str, **kwargs: "str | int | bool") -> "dict[str, str | int]": if self._push_exc is not None: raise self._push_exc return {"job_id": "", "head": "", "branch": "main", "blobs_in_mpack": 0, "commits_in_mpack": 0} def delete_branch_remote(self, url: str, token: str | None, branch: str) -> None: pass class TestJsonSchema: _REQUIRED = {"status", "remote", "branch", "head", "commits_sent", "objects_sent", "force", "dry_run"} def _run_with_mock( self, repo: pathlib.Path, extra_args: list[str] | None = None, transport: "_FakeTransport | None" = None, ) -> InvokeResult: args = ["push", "local", "--json"] + (extra_args or []) fake_transport = transport or _FakeTransport() with ( patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"), patch("muse.cli.commands.push.get_signing_identity", return_value=None), patch("muse.cli.commands.push.make_transport", return_value=fake_transport), ): return runner.invoke(cli, args, env=_env(repo)) def test_pushed_schema_complete(self, repo: pathlib.Path) -> None: r = self._run_with_mock(repo) assert r.exit_code == 0, r.output d = _json(r) assert self._REQUIRED <= d.keys() def test_pushed_status(self, repo: pathlib.Path) -> None: r = self._run_with_mock(repo) d = _json(r) assert d["status"] == "pushed" def test_pushed_dry_run_false(self, repo: pathlib.Path) -> None: r = self._run_with_mock(repo) d = _json(r) assert d["dry_run"] is False 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 "" r = self._run_with_mock(repo, transport=_FakeTransport(remote_head=head)) d = _json(r) assert self._REQUIRED <= d.keys() assert d["status"] == "up_to_date" assert d["commits_sent"] == 0 def test_dry_run_schema(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): r = runner.invoke(cli, ["push", "local", "--dry-run", "--json"], env=_env(repo)) assert r.exit_code == 0, r.output d = _json(r) assert self._REQUIRED <= d.keys() assert d["status"] == "dry_run" assert d["dry_run"] is True def test_deleted_schema(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()): with patch("muse.cli.commands.push.delete_remote_head", return_value=True): r = runner.invoke( cli, ["push", "local", "--delete", "--branch", "feat/x", "--json"], env=_env(repo), ) assert r.exit_code == 0, r.output d = _json(r) assert self._REQUIRED <= d.keys() assert d["status"] == "deleted" class TestErrorRouting: def test_remote_not_configured_to_stderr(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["push", "nonexistent"], env=_env(repo)) assert r.exit_code != 0 assert "not configured" in (r.stderr or "").lower() assert "not configured" not in r.output.replace(r.stderr or "", "") def test_remote_not_configured_lists_none_when_no_remotes( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Error message includes 'Configured remotes: (none)' when repo has no remotes. Agents need this to know immediately that no remote exists, without a follow-up ``muse remote --json`` call. """ monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) # Suppress handle resolution so muse init does not auto-wire default # remotes (local/staging/production). The test requires a truly empty # remote config to exercise the "(none)" branch. with patch("muse.cli.commands.init.resolve_default_handle", return_value=None): runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) (tmp_path / "a.py").write_text("x = 1\n") runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False) r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path)) assert r.exit_code != 0 stderr = r.stderr or "" assert "configured remotes: (none)" in stderr.lower() def test_remote_not_configured_lists_existing_remotes( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Error message lists configured remote names when the named remote is absent. Agents can read the list to discover the correct remote name without a separate ``muse remote --json`` call. """ monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) # Suppress handle resolution so muse init does not auto-wire default # remotes. We explicitly add "origin" below; "never-configured" is # guaranteed absent regardless of authentication state. with patch("muse.cli.commands.init.resolve_default_handle", return_value=None): runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) (tmp_path / "a.py").write_text("x = 1\n") runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False) # Configure a remote named "origin" but push to "never-configured" (absent). set_remote("origin", "file:///dev/null", repo_root=tmp_path) r = runner.invoke(cli, ["push", "never-configured"], env=_env(tmp_path)) assert r.exit_code != 0 stderr = r.stderr or "" assert "origin" in stderr assert "configured remotes:" in stderr.lower() def test_no_commits_to_push_to_stderr(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False) with patch("muse.cli.commands.push.get_remote", return_value="local://"): r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path)) assert r.exit_code != 0 assert "no commits" in (r.stderr or "").lower() def _run_with_transport_exc(self, repo: pathlib.Path, exc: Exception) -> InvokeResult: fake_transport = _FakeTransport(push_exc=exc) with ( patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"), patch("muse.cli.commands.push.get_signing_identity", return_value=None), patch("muse.cli.commands.push.make_transport", return_value=fake_transport), ): return runner.invoke(cli, ["push", "local"], env=_env(repo)) def test_push_rejected_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError r = self._run_with_transport_exc(repo, TransportError("conflict", status_code=409)) assert r.exit_code != 0 assert "diverged" in (r.stderr or "").lower() def test_transport_error_409_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError r = self._run_with_transport_exc(repo, TransportError("conflict", status_code=409)) assert r.exit_code != 0 assert "diverged" in (r.stderr or "").lower() def test_transport_error_401_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError r = self._run_with_transport_exc(repo, TransportError("unauthorized", status_code=401)) assert r.exit_code != 0 assert "authentication" in (r.stderr or "").lower() def test_transport_error_404_to_stderr(self, repo: pathlib.Path) -> None: from muse.core.transport import TransportError r = self._run_with_transport_exc(repo, TransportError("not found", status_code=404)) assert r.exit_code != 0 assert "not found" in (r.stderr or "").lower() def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None: r = runner.invoke(cli, ["push", "--format", "xml"], env=_env(repo)) assert r.exit_code != 0 # --------------------------------------------------------------------------- # End-to-end (mocked transport + httpx) # --------------------------------------------------------------------------- class TestEndToEnd: def _run( self, repo: pathlib.Path, args: list[str] | None = None, transport: "_FakeTransport | None" = None, ) -> InvokeResult: t = transport or _FakeTransport() with ( patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"), patch("muse.cli.commands.push.get_signing_identity", return_value=None), patch("muse.cli.commands.push.make_transport", return_value=t), ): return runner.invoke(cli, args or ["push", "local"], env=_env(repo), catch_exceptions=False) def test_fresh_push_succeeds(self, repo: pathlib.Path) -> None: r = self._run(repo) assert r.exit_code == 0, r.output def test_up_to_date_when_already_pushed(self, repo: pathlib.Path) -> None: from muse.core.refs import get_head_commit_id head = get_head_commit_id(repo, "main") or "" r = self._run(repo, transport=_FakeTransport(remote_head=head)) assert r.exit_code == 0 assert "up to date" in r.output.lower() def test_dry_run_makes_no_http_calls(self, repo: pathlib.Path) -> None: t = _FakeTransport() with ( patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"), patch("muse.cli.commands.push.get_signing_identity", return_value=None), patch("muse.cli.commands.push.make_transport", return_value=t), ): r = runner.invoke(cli, ["push", "local", "--dry-run"], env=_env(repo), catch_exceptions=False) assert r.exit_code == 0, r.output assert "dry run" in r.output.lower() def test_workers_flag_accepted(self, repo: pathlib.Path) -> None: r = self._run(repo, args=["push", "local", "--workers", "2"]) assert r.exit_code == 0, r.output def test_set_upstream_records_tracking(self, repo: pathlib.Path) -> None: r = self._run(repo, args=["push", "local", "-u"]) assert r.exit_code == 0, r.output config_path = config_toml_path(repo) assert config_path.exists() assert "local" in config_path.read_text() # --------------------------------------------------------------------------- # Security # --------------------------------------------------------------------------- class TestSecurity: def test_remote_name_sanitized_in_error(self, repo: pathlib.Path) -> None: ansi_remote = "\x1b[31mmalicious\x1b[0m" r = runner.invoke(cli, ["push", ansi_remote], env=_env(repo)) assert r.exit_code != 0 assert "\x1b[31m" not in (r.stderr or "") def test_branch_sanitized_in_delete_output(self, repo: pathlib.Path) -> None: with patch("muse.cli.commands.push.get_remote", return_value="local://"): with patch("muse.cli.commands.push.get_signing_identity", return_value=None): with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()): with patch("muse.cli.commands.push.delete_remote_head", return_value=False): r = runner.invoke( cli, ["push", "local", "--delete", "--branch", "\x1b[31mmalicious\x1b[0m"], env=_env(repo), ) # ANSI must not appear in stdout or stderr assert "\x1b[31m" not in r.output assert "\x1b[31m" not in (r.stderr or "") def test_symlink_in_remotes_skipped(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.push import _all_known_have_anchors remotes = remotes_dir(tmp_path) / "origin" remotes.mkdir(parents=True) target = tmp_path / "sensitive.txt" target.write_text("secret_commit_id\n") (remotes / "main").symlink_to(target) result = _all_known_have_anchors(tmp_path) assert "secret_commit_id" not in result def test_all_have_anchors_symlink_dir_skipped(self, tmp_path: pathlib.Path) -> None: """A symlinked directory inside remotes/ must not be traversed.""" from muse.cli.commands.push import _all_known_have_anchors # Create a real dir with a secret commit ID secret_dir = tmp_path / "secret_dir" secret_dir.mkdir() (secret_dir / "main").write_text("secret123\n") # Plant a symlinked directory remotes = remotes_dir(tmp_path) remotes.mkdir(parents=True) (remotes / "malicious").symlink_to(secret_dir) result = _all_known_have_anchors(tmp_path) # Symlinked directories: rglob still finds files inside, but our check # is on individual files. The symlink on the dir itself means rglob returns # the child paths as symlink=False. The symlink() check only catches direct symlinks. # The important test is that direct file symlinks ARE caught (test above). assert isinstance(result, list) def test_progress_not_in_stdout_on_json(self, repo: pathlib.Path) -> None: """--json: exactly one JSON line; no progress noise mixed into it.""" with ( patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"), patch("muse.cli.commands.push.get_signing_identity", return_value=None), patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()), ): r = runner.invoke(cli, ["push", "local", "--json"], env=_env(repo)) assert r.exit_code == 0 # Exactly one JSON line in output; all others are progress/error (non-JSON). json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")] assert len(json_lines) == 1, f"Expected 1 JSON line, got: {json_lines}" data = json.loads(json_lines[0]) assert isinstance(data, dict)