"""Comprehensive hardening tests for ``muse push``.

Covers all changes introduced in the push command review:

Unit
----
- Parser flags: --dry-run, --workers, --format/--json
- Dead-code removal: _current_branch absent
- _all_known_have_anchors: symlink skipping, binary-file safety, missing dir
- _upload_presigned: retry on 5xx/429, non-retriable 4xx propagated immediately
- _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
- --format 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
- invalid --format exits 1 to stderr
- progress prints from _upload_chunk go to stderr

Stress
------
- _push_objects_parallel with 1000 objects (mocked transport)
- _upload_presigned retries exhaust then raise
- 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 tests.cli_test_helper import CliRunner, InvokeResult

if TYPE_CHECKING:
    from muse.cli.commands.push import _PushJson
    from muse.core.pack import ObjectPayload, ObjectsChunkResponse, PackBundle, PushResult, RemoteInfo
    from muse.core.transport import PresignResponse

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())

    def test_presign_retries_constant_defined(self) -> None:
        import muse.cli.commands.push as m
        assert hasattr(m, "_PRESIGN_RETRIES")
        assert isinstance(m._PRESIGN_RETRIES, int)
        assert m._PRESIGN_RETRIES >= 1


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_format_json_shorthand(self) -> None:
        ns = self._parse("--json")
        assert ns.fmt == "json"

    def test_format_flag(self) -> None:
        ns = self._parse("--format", "json")
        assert ns.fmt == "json"

    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 = tmp_path / ".muse" / "remotes" / "origin"
        remotes.mkdir(parents=True)
        (remotes / "main").write_text("abc123\n")
        result = _all_known_have_anchors(tmp_path)
        assert "abc123" in result

    def test_symlinks_are_skipped(self, tmp_path: pathlib.Path) -> None:
        from muse.cli.commands.push import _all_known_have_anchors
        remotes = tmp_path / ".muse" / "remotes" / "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 = tmp_path / ".muse" / "remotes" / "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 = tmp_path / ".muse" / "remotes" / "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
        for name in ["origin", "upstream", "fork"]:
            d = tmp_path / ".muse" / "remotes" / name
            d.mkdir(parents=True)
            (d / "main").write_text(f"commit_{name}\n")
        result = _all_known_have_anchors(tmp_path)
        assert len(result) == 3
        assert "commit_origin" in result


class TestUploadPresigned:
    """Tests for _upload_presigned — always exercise the urllib path (Linux branch)."""

    # Force the urllib path on all platforms so tests can mock urlopen cleanly.
    _linux = patch("muse.cli.commands.push.platform.system", return_value="Linux")

    def test_success_no_retry(self) -> None:
        from muse.cli.commands.push import _upload_presigned

        call_count = 0
        def fake_urlopen(req: urllib.request.Request, timeout: int) -> _FakeResponse:
            nonlocal call_count
            call_count += 1
            return _FakeResponse()

        with self._linux, patch("urllib.request.urlopen", fake_urlopen):
            _upload_presigned("abc" * 20, "http://fake/url", b"data", retries=3)
        assert call_count == 1

    def test_retries_on_503(self) -> None:
        from muse.cli.commands.push import _upload_presigned

        call_count = 0
        def fake_urlopen(req: urllib.request.Request, timeout: int) -> _FakeResponse:
            nonlocal call_count
            call_count += 1
            if call_count < 3:
                raise urllib.error.HTTPError("", 503, "Service Unavailable", http.client.HTTPMessage(), None)
            return _FakeResponse()

        with self._linux, patch("urllib.request.urlopen", fake_urlopen), patch("time.sleep"):
            _upload_presigned("abc" * 20, "http://fake/url", b"data", retries=3)
        assert call_count == 3

    def test_non_retriable_4xx_propagated_immediately(self) -> None:
        from muse.cli.commands.push import _upload_presigned

        call_count = 0
        def fake_urlopen(req: urllib.request.Request, timeout: int) -> _FakeResponse:
            nonlocal call_count
            call_count += 1
            raise urllib.error.HTTPError("", 403, "Forbidden", http.client.HTTPMessage(), None)

        with self._linux, patch("urllib.request.urlopen", fake_urlopen):
            with pytest.raises(urllib.error.HTTPError) as exc_info:
                _upload_presigned("abc" * 20, "http://fake/url", b"data", retries=3)
        assert exc_info.value.code == 403
        assert call_count == 1  # no retries for 4xx (non-429)

    def test_all_retries_exhausted_raises(self) -> None:
        from muse.cli.commands.push import _upload_presigned

        def fake_urlopen(req: urllib.request.Request, timeout: int) -> _FakeResponse:
            raise urllib.error.HTTPError("", 503, "Service Unavailable", http.client.HTTPMessage(), None)

        with self._linux, patch("urllib.request.urlopen", fake_urlopen), patch("time.sleep"):
            with pytest.raises(urllib.error.HTTPError):
                _upload_presigned("abc" * 20, "http://fake/url", b"data", retries=2)


class TestUploadChunk:
    def test_progress_goes_to_stderr(self, capsys: pytest.CaptureFixture[str]) -> None:
        """_upload_chunk must not write to stdout so JSON output stays clean."""
        from muse.cli.commands.push import _upload_chunk

        mock_transport = MagicMock()
        mock_transport.push_objects.return_value = {"stored": 5, "skipped": 0}
        stored, skipped = _upload_chunk(mock_transport, "http://x", None, [], 1, 1)
        captured = capsys.readouterr()
        assert captured.out == ""
        assert "chunk 1/1" in captured.err


# ---------------------------------------------------------------------------
# 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.pack 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 filter_objects(
        self,
        url: str,
        signing: object,
        object_ids: list[str],
        object_hints: dict[str, str] | None = None,
    ) -> "FilterObjectsResult":
        from muse.core.transport import FilterObjectsResult
        return FilterObjectsResult(missing=object_ids, bases={})

    def presign_objects(self, url: str, token: str | None, ids: list[str], op: str) -> "PresignResponse":
        from muse.core.transport import PresignResponse
        return PresignResponse(presigned={}, inline=ids)

    def push_objects(self, url: str, token: str | None, objects: list["ObjectPayload"]) -> "ObjectsChunkResponse":
        from muse.core.pack import ObjectsChunkResponse
        return ObjectsChunkResponse(stored=len(objects), skipped=0)

    def push_object_pack(self, url: str, signing: object, pack: list["ObjectPayload"]) -> "ObjectsChunkResponse":
        from muse.core.pack import ObjectsChunkResponse
        return ObjectsChunkResponse(stored=len(pack), skipped=0)

    def push_pack(self, url: str, token: str | None, bundle: "PackBundle", branch: str, force: bool, local_head: str | None = None) -> "PushResult":
        from muse.core.pack import PushResult
        if self._push_exc is not None:
            raise self._push_exc
        return PushResult(
            ok=self._push_ok,
            message="ok" if self._push_ok else "rejected",
            branch_heads={"main": "deadbeef" * 8},
        )

    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="local://"):
            with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
                with 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.store 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_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 test_push_rejected_to_stderr(self, repo: pathlib.Path) -> None:
        from muse.core.transport import TransportError
        fake_transport = _FakeTransport()
        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=fake_transport):
                    with patch.object(fake_transport, "push_pack") as mock_push:
                        from muse.core.pack import PushResult
                        mock_push.return_value = PushResult(
                            ok=False, message="rejected", branch_heads={}
                        )
                        r = runner.invoke(cli, ["push", "local"], env=_env(repo))
        assert r.exit_code != 0
        assert "rejected" in (r.stderr or "").lower()

    def test_transport_error_409_to_stderr(self, repo: pathlib.Path) -> None:
        from muse.core.transport import TransportError
        exc = TransportError("conflict", status_code=409)
        fake_transport = _FakeTransport(push_exc=exc)
        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=fake_transport):
                    r = runner.invoke(cli, ["push", "local"], env=_env(repo))
        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
        exc = TransportError("unauthorized", status_code=401)
        fake_transport = _FakeTransport(push_exc=exc)
        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=fake_transport):
                    r = runner.invoke(cli, ["push", "local"], env=_env(repo))
        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
        exc = TransportError("not found", status_code=404)
        fake_transport = _FakeTransport(push_exc=exc)
        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=fake_transport):
                    r = runner.invoke(cli, ["push", "local"], env=_env(repo))
        assert r.exit_code != 0
        assert "not found" in (r.stderr or "").lower()

    def test_invalid_format_to_stderr(self, repo: pathlib.Path) -> None:
        r = runner.invoke(cli, ["push", "--format", "xml"], env=_env(repo))
        assert r.exit_code == 1
        assert "xml" in (r.stderr or "").lower()


# ---------------------------------------------------------------------------
# End-to-end with local:// transport
# ---------------------------------------------------------------------------

class TestEndToEnd:
    def test_fresh_push_succeeds(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        r = runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, r.output

    def test_second_push_up_to_date(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False)
        r = runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0
        assert "up to date" in r.output.lower()

    def test_push_json_schema(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        r = runner.invoke(
            cli, ["push", "local", "--json"],
            env=_env(local),
            catch_exceptions=False,
        )
        assert r.exit_code == 0, r.output
        d = _json(r)
        assert d["status"] == "pushed"
        assert isinstance(d["commits_sent"], int) and d["commits_sent"] >= 1
        assert isinstance(d["objects_sent"], int)

    def test_up_to_date_json_schema(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        runner.invoke(cli, ["push", "local"], env=_env(local), catch_exceptions=False)
        r = runner.invoke(cli, ["push", "local", "--json"], env=_env(local), catch_exceptions=False)
        d = _json(r)
        assert d["status"] == "up_to_date"
        assert d["commits_sent"] == 0

    def test_dry_run_does_not_push(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        r = runner.invoke(cli, ["push", "local", "--dry-run"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        assert "dry run" in r.output.lower()
        # Verify nothing was actually pushed by checking remote still needs a push
        r2 = runner.invoke(cli, ["push", "local", "--json"], env=_env(local), catch_exceptions=False)
        d2 = _json(r2)
        assert d2["status"] == "pushed"  # still needs to push — dry run wrote nothing

    def test_dry_run_json_schema(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        r = runner.invoke(
            cli, ["push", "local", "--dry-run", "--json"],
            env=_env(local),
            catch_exceptions=False,
        )
        assert r.exit_code == 0
        d = _json(r)
        assert d["status"] == "dry_run"
        assert d["dry_run"] is True
        assert isinstance(d["commits_sent"], int) and d["commits_sent"] >= 1

    def test_workers_flag_accepted(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        r = runner.invoke(
            cli, ["push", "local", "--workers", "2"],
            env=_env(local),
            catch_exceptions=False,
        )
        assert r.exit_code == 0, r.output

    def test_set_upstream_records_tracking(self, remote_repo: tuple[pathlib.Path, pathlib.Path]) -> None:
        local, remote = remote_repo
        r = runner.invoke(cli, ["push", "local", "-u"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, r.output
        config_path = local / ".muse" / "config.toml"
        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[31mevil\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[31mevil\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 = tmp_path / ".muse" / "remotes" / "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 = tmp_path / ".muse" / "remotes"
        remotes.mkdir(parents=True)
        (remotes / "evil").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:
        """--format json: exactly one JSON line; no progress noise mixed into it."""
        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()):
                    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)


# ---------------------------------------------------------------------------
# Stress
# ---------------------------------------------------------------------------

class TestStress:
    @pytest.mark.slow
    def test_push_objects_as_packs_1000(self, tmp_path: pathlib.Path) -> None:
        """1000 objects via pack endpoint in parallel — all must be stored."""
        from muse.cli.commands.push import _push_objects_as_packs

        uploaded_counts: list[int] = []

        def fake_push_pack(url: str, signing: object, pack: list) -> dict[str, int]:
            uploaded_counts.append(len(pack))
            return {"stored": len(pack), "skipped": 0}

        mock_transport = MagicMock()
        mock_transport.push_object_pack.side_effect = fake_push_pack

        # _push_objects_as_packs reads from the object store; missing OIDs yield b"".
        object_ids = [format(i, "064x") for i in range(1000)]
        stored, skipped = _push_objects_as_packs(
            mock_transport,
            "http://test",
            None,
            object_ids,
            tmp_path,
        )
        assert stored == 1000
        assert skipped == 0
        assert sum(uploaded_counts) == 1000

    @pytest.mark.slow
    def test_upload_presigned_retries_exhaust_raises(self) -> None:
        """Exhausting all retries must raise the last exception."""
        from muse.cli.commands.push import _upload_presigned

        call_count = 0

        def always_503(req: urllib.request.Request, timeout: int) -> _FakeResponse:
            nonlocal call_count
            call_count += 1
            raise urllib.error.HTTPError("", 503, "always fails", http.client.HTTPMessage(), None)

        _linux = patch("muse.cli.commands.push.platform.system", return_value="Linux")
        with _linux, patch("urllib.request.urlopen", always_503), patch("time.sleep"):
            with pytest.raises(urllib.error.HTTPError):
                _upload_presigned("a" * 64, "http://fake", b"data", retries=3)
        assert call_count == 3

    @pytest.mark.slow
    def test_concurrent_push_objects_as_packs_isolated(self, tmp_path: pathlib.Path) -> None:
        """Eight independent ``_push_objects_as_packs`` calls run concurrently.

        Each call gets its own transport mock and accumulates results in an
        isolated counter — verifies there is no shared-state corruption across
        the ThreadPoolExecutor workers used internally.
        """
        from muse.cli.commands.push import _push_objects_as_packs

        N_WORKERS = 8
        N_OBJECTS = 200   # objects per parallel call
        all_results: list[tuple[int, int]] = [(-1, -1)] * N_WORKERS
        errors: list[str] = []

        def run_one(idx: int) -> None:
            uploaded: list[int] = []

            def fake_pack(url: str, signing: object, pack: list) -> dict[str, int]:
                uploaded.append(len(pack))
                return {"stored": len(pack), "skipped": 0}

            mock_t = MagicMock()
            mock_t.push_object_pack.side_effect = fake_pack
            object_ids = [f"{idx:02d}" + format(j, "062x") for j in range(N_OBJECTS)]
            try:
                stored, skipped = _push_objects_as_packs(
                    mock_t, "http://test", None, object_ids, tmp_path
                )
                all_results[idx] = (stored, skipped)
            except Exception as exc:
                errors.append(f"worker {idx}: {exc}")

        threads = [threading.Thread(target=run_one, args=(i,)) for i in range(N_WORKERS)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        assert not errors, f"Concurrent errors: {errors}"
        for idx, (stored, skipped) in enumerate(all_results):
            assert stored == N_OBJECTS, f"worker {idx}: stored={stored}, expected {N_OBJECTS}"
            assert skipped == 0, f"worker {idx}: skipped={skipped}"


from muse.core.pack import PushResult, RemoteInfo
from muse.core._types import Manifest


# ---------------------------------------------------------------------------
# Regression — merge commit push must not re-send second-parent history
# ---------------------------------------------------------------------------

class TestMergeCommitPushBundleSize:
    """After merging branch A into branch B, pushing B must send only the
    merge commit itself — not the entire history of branch A.

    Regression for: push of a merge commit walks parent2's full ancestry
    because ``branch_have`` only contained the target branch's remote HEAD,
    leaving parent2's commits outside the ``seen`` set.
    """

    def _make_two_branch_remote(
        self,
        tmp_path: pathlib.Path,
        monkeypatch: pytest.MonkeyPatch,
        *,
        main_extra_commits: int = 5,
        dev_extra_commits: int = 3,
    ) -> tuple[pathlib.Path, pathlib.Path]:
        """Return (local, remote) where:
        - main has base + *main_extra_commits* commits, pushed to remote
        - dev branches from base, has *dev_extra_commits* extra commits, pushed
        - local HEAD is still on dev (not yet merged)
        """
        local = tmp_path / "local"
        remote = tmp_path / "remote"
        local.mkdir()
        remote.mkdir()

        monkeypatch.chdir(local)
        monkeypatch.setenv("MUSE_REPO_ROOT", str(local))
        runner.invoke(cli, ["init"], 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))
        set_remote("origin", f"file://{remote}", repo_root=local)

        def _commit(name: str, content: str) -> None:
            (local / name).write_text(content)
            runner.invoke(cli, ["code", "add", name], env=_env(local), catch_exceptions=False)
            runner.invoke(cli, ["commit", "-m", f"add {name}"], env=_env(local), catch_exceptions=False)

        # base commit on main
        _commit("base.py", "x = 0\n")

        # dev branches from base
        runner.invoke(cli, ["branch", "dev"], env=_env(local), catch_exceptions=False)

        # extra commits on main
        for i in range(main_extra_commits):
            _commit(f"main_{i}.py", f"v = {i}\n")

        # push main to remote
        r = runner.invoke(cli, ["push", "origin", "--branch", "main"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, f"push main failed: {r.output}"

        # switch to dev, add extra commits, push dev
        runner.invoke(cli, ["checkout", "dev"], env=_env(local), catch_exceptions=False)
        for i in range(dev_extra_commits):
            _commit(f"dev_{i}.py", f"d = {i}\n")

        r = runner.invoke(cli, ["push", "origin", "--branch", "dev"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, f"push dev failed: {r.output}"

        return local, remote

    def test_merge_push_sends_one_commit_exact_heads(
        self,
        tmp_path: pathlib.Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Push of a merge commit sends only the merge commit when both
        branch HEADs are already on the remote (exact remote head match)."""
        local, _remote = self._make_two_branch_remote(
            tmp_path, monkeypatch, main_extra_commits=5, dev_extra_commits=2
        )

        # merge main into dev
        r = runner.invoke(cli, ["merge", "main"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, f"merge failed: {r.output}"

        # push the merge commit — must send only 1 commit
        r = runner.invoke(
            cli, ["push", "origin", "--branch", "dev", "--json"],
            env=_env(local), catch_exceptions=False,
        )
        assert r.exit_code == 0, f"push after merge failed: {r.output}"
        d = _json(r)
        assert d["commits_sent"] == 1, (
            f"Expected 1 commit (the merge commit), got {d['commits_sent']}. "
            "push is re-sending the merged branch's full history."
        )

    def test_merge_push_sends_only_new_commits_when_branch_is_ahead(
        self,
        tmp_path: pathlib.Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """When the merged branch is N commits ahead of the remote, the push
        should send the merge commit + those N new commits, NOT the full history.

        Regression: branch_have only contained the target branch's remote HEAD.
        The BFS followed parent2's chain without a stop anchor, walking the
        entire ancestry of the merged branch instead of stopping at the nearest
        already-remote commit.
        """
        local, _remote = self._make_two_branch_remote(
            tmp_path, monkeypatch, main_extra_commits=5, dev_extra_commits=2
        )

        # Add 2 more commits to main locally, but do NOT push them.
        # Remote main is now 2 commits behind local main.
        runner.invoke(cli, ["checkout", "main"], env=_env(local), catch_exceptions=False)
        for i in range(2):
            (local / f"main_extra_{i}.py").write_text(f"e = {i}\n")
            runner.invoke(cli, ["code", "add", f"main_extra_{i}.py"], env=_env(local), catch_exceptions=False)
            runner.invoke(cli, ["commit", "-m", f"extra main {i}"], env=_env(local), catch_exceptions=False)

        runner.invoke(cli, ["checkout", "dev"], env=_env(local), catch_exceptions=False)

        # merge the (now-ahead) main into dev
        r = runner.invoke(cli, ["merge", "main"], env=_env(local), catch_exceptions=False)
        assert r.exit_code == 0, f"merge failed: {r.output}"

        r = runner.invoke(
            cli, ["push", "origin", "--branch", "dev", "--json"],
            env=_env(local), catch_exceptions=False,
        )
        assert r.exit_code == 0, f"push after merge failed: {r.output}"
        d = _json(r)
        # merge commit + 2 new commits from main = 3, NOT 5+2+1 = 8 full history
        assert d["commits_sent"] == 3, (
            f"Expected 3 commits (merge + 2 new on main), got {d['commits_sent']}. "
            "push walked the merged branch's full history instead of stopping at "
            "the nearest already-remote commit."
        )

    def test_merge_push_succeeds(
        self,
        tmp_path: pathlib.Path,
        monkeypatch: pytest.MonkeyPatch,
    ) -> None:
        """Push of a merge commit must complete without error."""
        local, _remote = self._make_two_branch_remote(
            tmp_path, monkeypatch, main_extra_commits=3, dev_extra_commits=2
        )
        runner.invoke(cli, ["merge", "main"], env=_env(local), catch_exceptions=False)
        r = runner.invoke(
            cli, ["push", "origin", "--branch", "dev"],
            env=_env(local), catch_exceptions=False,
        )
        assert r.exit_code == 0, f"push after merge failed: {r.output}"