"""TDD — clone with corrupt objects must exit PARTIAL (exit code 3), not INTERNAL_ERROR. C1 When fetch_presign_or_stream delivers objects and some fail the integrity check, clone must: - write the objects that are clean - skip the corrupt ones with a warning - exit with ExitCode.PARTIAL (3), not ExitCode.INTERNAL_ERROR (3 currently mis-mapped to INTERNAL_ERROR) - document this in the help text and ExitCode enum C2 The --json envelope must include a "skipped_objects" count when > 0. """ from __future__ import annotations import json import pathlib import unittest.mock import pytest from muse.core.errors import ExitCode from muse.core.pack import MPackBundle, ObjectPayload, RemoteInfo from muse.core.snapshot import compute_commit_id, compute_snapshot_id from muse.core.types import blob_id, long_id from tests.cli_test_helper import CliRunner runner = CliRunner() cli = None def _make_remote_info(repo_id: str = "repo-partial") -> RemoteInfo: return RemoteInfo( repo_id=repo_id, domain="code", default_branch="main", branch_heads={"main": long_id("c" * 64)}, ) def _make_bundle_with_corrupt_object() -> tuple[MPackBundle, str, str]: """Return (bundle, good_oid, corrupt_oid). The bundle contains two objects: - good_oid: content matches its ID - corrupt_oid: content is wrong bytes (sha256 mismatch) """ committed_at = "2026-01-01T00:00:00+00:00" good_content = b"good content" good_oid = blob_id(good_content) corrupt_content = b"wrong bytes" # Use an OID that does NOT match corrupt_content corrupt_oid = blob_id(b"the real content that should be here") snap_id = compute_snapshot_id({"good.txt": good_oid, "bad.txt": corrupt_oid}) cid = compute_commit_id( parent_ids=[], snapshot_id=snap_id, message="test commit", committed_at_iso=committed_at, author="gabriel", ) branch_heads = {"main": cid} bundle = MPackBundle( commits=[{ "commit_id": cid, "repo_id": "repo-partial", "branch": "main", "snapshot_id": snap_id, "message": "test commit", "committed_at": committed_at, "parent_commit_id": None, "parent2_commit_id": None, "author": "gabriel", "metadata": {}, "structured_delta": None, "sem_ver_bump": "none", "breaking_changes": [], "agent_id": "", "model_id": "", "toolchain_id": "", "prompt_hash": "", "signature": "", "signer_key_id": "", "reviewed_by": [], "test_runs": 0, }], snapshots=[{ "snapshot_id": snap_id, "manifest": {"good.txt": good_oid, "bad.txt": corrupt_oid}, "created_at": committed_at, }], objects=[ ObjectPayload(object_id=good_oid, content=good_content), ObjectPayload(object_id=corrupt_oid, content=corrupt_content), ], branch_heads=branch_heads, ) return bundle, good_oid, corrupt_oid def _mock_transport(info: RemoteInfo, bundle: MPackBundle) -> unittest.mock.MagicMock: mock = unittest.mock.MagicMock() # Use the bundle's real content-addressed branch heads so apply_mpack and # _restore_working_tree can find the commit after the fetch. real_heads = bundle["branch_heads"] patched_info = {**info, "branch_heads": real_heads} mock.fetch_remote_info.return_value = patched_info fetch_result = { "repo_id": info["repo_id"], "domain": info["domain"], "default_branch": info["default_branch"], "branch_heads": real_heads, "commits": bundle["commits"], "snapshots": bundle["snapshots"], "objects_received": len(bundle["objects"]), "shallow_commits": [], } def _fake_fetch_presign_or_stream(url, signing, *, want, have, on_object=None, **kw): if on_object is not None: for obj in bundle["objects"]: on_object(obj) return fetch_result mock.fetch_stream.return_value = fetch_result mock.fetch_presign_or_stream.side_effect = _fake_fetch_presign_or_stream return mock class TestClonePartial: def test_exit_code_is_partial_when_objects_skipped( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Clone with a corrupt object must exit PARTIAL, not INTERNAL_ERROR.""" monkeypatch.chdir(tmp_path) info = _make_remote_info() bundle, good_oid, corrupt_oid = _make_bundle_with_corrupt_object() mock = _mock_transport(info, bundle) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): result = runner.invoke(cli, ["clone", "https://hub.example.com/repo-partial", "cloned"]) assert result.exit_code == ExitCode.PARTIAL def test_good_object_is_written_despite_corrupt_sibling( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """The clean object must be written even when another object is corrupt.""" monkeypatch.chdir(tmp_path) info = _make_remote_info() bundle, good_oid, corrupt_oid = _make_bundle_with_corrupt_object() mock = _mock_transport(info, bundle) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): runner.invoke(cli, ["clone", "https://hub.example.com/repo-partial", "cloned"]) from muse.core.object_store import object_path from muse.core.paths import server_objects_dir cloned = tmp_path / "cloned" good_path = object_path(cloned, good_oid) corrupt_path = object_path(cloned, corrupt_oid) assert good_path.exists(), "clean object must be written" assert not corrupt_path.exists(), "corrupt object must not be written" def test_json_envelope_includes_skipped_objects_count( self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """--json output must include skipped_objects count.""" monkeypatch.chdir(tmp_path) info = _make_remote_info() bundle, good_oid, corrupt_oid = _make_bundle_with_corrupt_object() mock = _mock_transport(info, bundle) with unittest.mock.patch("muse.cli.commands.clone.make_transport", return_value=mock): result = runner.invoke(cli, ["clone", "https://hub.example.com/repo-partial", "cloned", "--json"]) lines = [l for l in result.output.strip().splitlines() if l.startswith("{")] assert lines, "expected JSON output" d = json.loads(lines[0]) assert "skipped_objects" in d, "JSON envelope must include skipped_objects" assert d["skipped_objects"] == 1