"""TDD — --force flag must appear in the unpack-mpack request body. Gap 2: _run_mpack_path() never includes "force" in the unpack body. The server receives the unpack request with no force field and has no way to distinguish a force push from a normal push. Test plan --------- F1 (RED) _push_mpack with force=True sends {"force": True} in the unpack-mpack POST body. F2 (RED) _push_mpack with force=False sends {"force": False} (or omits it and server defaults to False). F3 _push_mpack without --force does NOT set force=True by accident. F4 The "force" key is present in the unpack body regardless of the force value (protocol stability). """ from __future__ import annotations import asyncio import datetime import json import pathlib from unittest.mock import MagicMock, patch import msgpack import pytest from muse._version import __version__ from muse.core.mpack import PushResult, RemoteInfo from muse.core.object_store import write_object from muse.core.paths import heads_dir, muse_dir _Headers = dict[str, str] # HTTP header map _JsonDict = dict[str, str | int | float | bool | None | list[str]] # JSON object from muse.core.ids import hash_commit as compute_commit_id, hash_snapshot as compute_snapshot_id from muse.core.commits import ( CommitRecord, write_commit, ) from muse.core.snapshots import ( SnapshotRecord, write_snapshot, ) from muse.core.types import Manifest, blob_id # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _bare_repo(tmp_path: pathlib.Path) -> pathlib.Path: muse = muse_dir(tmp_path) for d in ("commits", "snapshots", "objects", "refs/heads", "remotes"): (muse / d).mkdir(parents=True, exist_ok=True) (muse / "HEAD").write_text("ref: refs/heads/main\n") (muse / "repo.json").write_text( json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"}) ) (muse / "config.toml").write_text('[remotes.origin]\nurl = "https://hub.example.com/r"\n') return tmp_path def _make_commit( root: pathlib.Path, label: str, parent_id: str | None = None, content: bytes | None = None, ) -> CommitRecord: raw = content if content is not None else f"content-{label}".encode() oid = blob_id(raw) write_object(root, oid, raw) manifest: Manifest = {"file.txt": oid} snap_id = compute_snapshot_id(manifest) write_snapshot(root, SnapshotRecord(snapshot_id=snap_id, manifest=manifest)) committed_at = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc) parent_ids = [parent_id] if parent_id else [] cid = compute_commit_id( parent_ids=parent_ids, snapshot_id=snap_id, message=f"commit {label}", committed_at_iso=committed_at.isoformat(), ) write_commit(root, CommitRecord( commit_id=cid, branch="main", snapshot_id=snap_id, message=f"commit {label}", committed_at=committed_at, parent_commit_id=parent_id, )) return CommitRecord( commit_id=cid, branch="main", snapshot_id=snap_id, message=f"commit {label}", committed_at=committed_at, parent_commit_id=parent_id, ) def _fake_resp(body: bytes, status: int = 200) -> MagicMock: r = MagicMock() r.status_code = status r.content = body r.headers = {"content-type": "application/x-msgpack"} r.text = "" return r def _run_push_mpack( root: pathlib.Path, local_head: str, remote_head: str | None, force: bool, ) -> MagicMock: """Run _push_mpack with a fake transport and return the transport mock. Callers inspect transport.push_mpack_unpack.call_args to verify the force flag was passed correctly. """ from muse.cli.commands.push import _push_mpack transport = MagicMock() transport.fetch_remote_info.return_value = RemoteInfo( domain="code", default_branch="main", branch_heads={"main": remote_head} if remote_head else {}, ) mock_req = MagicMock() mock_req.headers = {"Authorization": "MSign stub", "Content-Type": "application/x-msgpack"} transport._build_request.return_value = mock_req transport.push_mpack_presign.return_value = { "upload_url": "https://minio.example.com/put?sig=x", "mpack_key": "sha256:fake", } transport.push_mpack_put.return_value = None transport.push_mpack_unpack.return_value = { "job_id": "j", "head": local_head, "branch": "main", "blobs_in_mpack": 0, "commits_in_mpack": 0, } _push_mpack( transport, "https://hub.example.com/repos/r1", None, root, local_head, [], "main", force, ) return transport # --------------------------------------------------------------------------- # F1 — force=True appears in unpack body (currently RED) # --------------------------------------------------------------------------- def test_f1_force_true_sent_in_unpack_body( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """_push_mpack(force=True) must send {"force": True} in the unpack body. This is currently RED: _run_mpack_path does not include "force" in the unpack payload. """ monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) root = _bare_repo(tmp_path) commit = _make_commit(root, "f1", content=b"force flag test") (heads_dir(root) / "main").write_text(commit.commit_id) transport = _run_push_mpack(root, commit.commit_id, remote_head=None, force=True) assert transport.push_mpack_unpack.called, "push_mpack_unpack was not called" kwargs = transport.push_mpack_unpack.call_args.kwargs assert "force" in kwargs, ( f"'force' kwarg missing from push_mpack_unpack call. Got: {list(kwargs.keys())}" ) assert kwargs["force"] is True, ( f"Expected force=True, got {kwargs['force']!r}" ) # --------------------------------------------------------------------------- # F2 — force=False is also sent explicitly (currently RED) # --------------------------------------------------------------------------- def test_f2_force_false_sent_in_unpack_body( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """_push_mpack(force=False) must send {"force": False} in the unpack body.""" monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) root = _bare_repo(tmp_path) commit = _make_commit(root, "f2", content=b"no force test") (heads_dir(root) / "main").write_text(commit.commit_id) transport = _run_push_mpack(root, commit.commit_id, remote_head=None, force=False) assert transport.push_mpack_unpack.called, "push_mpack_unpack was not called" kwargs = transport.push_mpack_unpack.call_args.kwargs assert "force" in kwargs, ( f"'force' kwarg missing from push_mpack_unpack call. Got: {list(kwargs.keys())}" ) assert kwargs["force"] is False, ( f"Expected force=False, got {kwargs['force']!r}" ) # --------------------------------------------------------------------------- # F3 — "force" key is always present regardless of value # --------------------------------------------------------------------------- def test_f3_force_key_present_in_all_pushes( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The 'force' key must always be in the unpack body (protocol stability).""" monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path)) monkeypatch.chdir(tmp_path) root = _bare_repo(tmp_path) commit = _make_commit(root, "f3", content=b"always force key") (heads_dir(root) / "main").write_text(commit.commit_id) for force_value in (True, False): transport = _run_push_mpack(root, commit.commit_id, remote_head=None, force=force_value) assert transport.push_mpack_unpack.called, f"force={force_value}: push_mpack_unpack not called" kwargs = transport.push_mpack_unpack.call_args.kwargs assert "force" in kwargs, f"force={force_value}: 'force' missing from push_mpack_unpack kwargs" assert isinstance(kwargs["force"], bool), ( f"force={force_value}: kwargs['force'] must be bool, got {type(kwargs['force'])}" )