"""TDD — muse clone --depth N (shallow clone). Test plan --------- C1 --depth flag is registered on the clone subcommand. C2 --depth clone calls fetch_mpack (presign-only policy; no stream path). C3 clone with depth=1 writes a .muse/shallow file listing boundary commits. C4 clone with depth=None (full) does NOT write .muse/shallow. C5 objects_written reflects only objects from depth-limited commits. C6 --depth 0 is rejected with a user-friendly error (depth must be >= 1). C7 JSON output includes a shallow_commits field when depth is active. """ from __future__ import annotations import argparse import json import pathlib from unittest.mock import MagicMock, patch, call import pytest from muse.cli.commands.clone import register from muse.core.types import MsgpackDict, long_id from muse.core.paths import muse_dir type BranchHeads = dict[str, str] # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_parser() -> tuple[argparse.ArgumentParser, argparse.ArgumentParser]: root = argparse.ArgumentParser() sub = root.add_subparsers() clone_parser = sub.add_parser("clone") register(sub) return root def _clone_parser() -> argparse.ArgumentParser: root = argparse.ArgumentParser() subs = root.add_subparsers(dest="cmd") register(subs) return root def _fake_fetch_result( *, commits: list[MsgpackDict] | None = None, snapshots: list[MsgpackDict] | None = None, objects_received: int = 0, branch_heads: BranchHeads | None = None, shallow_commits: list[str] | None = None, ) -> MsgpackDict: return { "repo_id": long_id("a" * 64), "domain": "code", "default_branch": "main", "branch_heads": branch_heads or {"main": long_id("b" * 64)}, "commits": commits or [], "snapshots": snapshots or [], "objects": [], "objects_received": objects_received, "shallow_commits": shallow_commits or [], } def _fake_apply_result(commits_written: int = 0) -> MsgpackDict: return { "commits_written": commits_written, "snapshots_written": 0, "objects_written": 0, "objects_skipped": 0, "tags_written": 0, "failed_objects": [], "skipped_snapshots": [], } def _fake_remote_info(branch_heads: BranchHeads | None = None) -> MsgpackDict: heads = branch_heads or {"main": long_id("b" * 64)} return { "repo_id": long_id("a" * 64), "domain": "code", "default_branch": "main", "branch_heads": heads, } # --------------------------------------------------------------------------- # C1 — --depth flag is registered # --------------------------------------------------------------------------- class TestDepthFlagRegistered: def test_depth_flag_exists(self) -> None: parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x", "--depth", "1"]) assert args.depth == 1 def test_depth_defaults_to_none(self) -> None: parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x"]) assert args.depth is None def test_depth_shortflag(self) -> None: parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x", "--depth", "5"]) assert args.depth == 5 # --------------------------------------------------------------------------- # C2 — --depth clone uses fetch_mpack (presign-only policy) # --------------------------------------------------------------------------- class TestDepthForwardedToTransport: def test_depth_1_calls_fetch_mpack(self, tmp_path: pathlib.Path) -> None: dest = str(tmp_path / "cloned") tip = long_id("b" * 64) mock_transport = MagicMock() mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) mock_transport.fetch_mpack.return_value = _fake_fetch_result( branch_heads={"main": tip}, shallow_commits=[tip], ) with ( patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), patch("muse.cli.commands.clone.get_signing_identity", return_value=None), patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(0)), patch("muse.cli.commands.clone._restore_working_tree"), patch("muse.cli.commands.clone.set_remote"), patch("muse.cli.commands.clone.set_remote_head"), patch("muse.cli.commands.clone.set_upstream"), patch("muse.cli.commands.clone.write_branch_ref"), ): parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", dest, "--depth", "1"]) args.func(args) mock_transport.fetch_mpack.assert_called_once() def test_no_depth_calls_fetch_mpack(self, tmp_path: pathlib.Path) -> None: dest = str(tmp_path / "cloned") tip = long_id("b" * 64) mock_transport = MagicMock() mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) mock_transport.fetch_mpack.return_value = _fake_fetch_result(branch_heads={"main": tip}) with ( patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), patch("muse.cli.commands.clone.get_signing_identity", return_value=None), patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(0)), patch("muse.cli.commands.clone._restore_working_tree"), patch("muse.cli.commands.clone.set_remote"), patch("muse.cli.commands.clone.set_remote_head"), patch("muse.cli.commands.clone.set_upstream"), patch("muse.cli.commands.clone.write_branch_ref"), ): parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", dest]) args.func(args) mock_transport.fetch_mpack.assert_called_once() # --------------------------------------------------------------------------- # C3 — .muse/shallow written when depth is active # --------------------------------------------------------------------------- class TestShallowFileWritten: def test_shallow_file_written_on_depth_clone(self, tmp_path: pathlib.Path) -> None: dest = tmp_path / "cloned" tip = long_id("b" * 64) mock_transport = MagicMock() mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) mock_transport.fetch_mpack.return_value = _fake_fetch_result( branch_heads={"main": tip}, shallow_commits=[tip], ) with ( patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), patch("muse.cli.commands.clone.get_signing_identity", return_value=None), patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(1)), patch("muse.cli.commands.clone._restore_working_tree"), patch("muse.cli.commands.clone.set_remote"), patch("muse.cli.commands.clone.set_remote_head"), patch("muse.cli.commands.clone.set_upstream"), patch("muse.cli.commands.clone.write_branch_ref"), ): parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", str(dest), "--depth", "1"]) args.func(args) shallow_file = muse_dir(dest) / "shallow" assert shallow_file.exists(), ".muse/shallow must be written for a shallow clone" contents = shallow_file.read_text().strip().splitlines() assert tip in contents # --------------------------------------------------------------------------- # C4 — no .muse/shallow on full clone # --------------------------------------------------------------------------- class TestNoShallowFileOnFullClone: def test_no_shallow_file_on_full_clone(self, tmp_path: pathlib.Path) -> None: dest = tmp_path / "cloned" tip = long_id("b" * 64) mock_transport = MagicMock() mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) mock_transport.fetch_mpack.return_value = _fake_fetch_result(branch_heads={"main": tip}) with ( patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), patch("muse.cli.commands.clone.get_signing_identity", return_value=None), patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(1)), patch("muse.cli.commands.clone._restore_working_tree"), patch("muse.cli.commands.clone.set_remote"), patch("muse.cli.commands.clone.set_remote_head"), patch("muse.cli.commands.clone.set_upstream"), patch("muse.cli.commands.clone.write_branch_ref"), ): parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", str(dest)]) args.func(args) shallow_file = muse_dir(dest) / "shallow" assert not shallow_file.exists(), ".muse/shallow must NOT exist for a full clone" # --------------------------------------------------------------------------- # C6 — depth=0 rejected # --------------------------------------------------------------------------- class TestInvalidDepth: def test_depth_zero_rejected(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: dest = str(tmp_path / "cloned") parser = _clone_parser() args = parser.parse_args(["clone", "https://example.com/repo", dest, "--depth", "0"]) with pytest.raises(SystemExit) as exc_info: args.func(args) assert exc_info.value.code != 0 captured = capsys.readouterr() assert "depth" in (captured.err + captured.out).lower() # --------------------------------------------------------------------------- # C7 — JSON output includes shallow_commits when depth active # --------------------------------------------------------------------------- class TestShallowJsonOutput: def test_json_includes_shallow_commits(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: dest = tmp_path / "cloned" tip = long_id("b" * 64) mock_transport = MagicMock() mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) mock_transport.fetch_mpack.return_value = _fake_fetch_result( branch_heads={"main": tip}, shallow_commits=[tip], ) with ( patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), patch("muse.cli.commands.clone.get_signing_identity", return_value=None), patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(1)), patch("muse.cli.commands.clone._restore_working_tree"), patch("muse.cli.commands.clone.set_remote"), patch("muse.cli.commands.clone.set_remote_head"), patch("muse.cli.commands.clone.set_upstream"), patch("muse.cli.commands.clone.write_branch_ref"), ): parser = _clone_parser() args = parser.parse_args([ "clone", "https://example.com/repo", str(dest), "--depth", "1", "--json", ]) args.func(args) out = capsys.readouterr().out.strip() data = json.loads(out) assert "shallow_commits" in data assert tip in data["shallow_commits"]