test_clone_shallow.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """TDD — muse clone --depth N (shallow clone). |
| 2 | |
| 3 | Test plan |
| 4 | --------- |
| 5 | C1 --depth flag is registered on the clone subcommand. |
| 6 | C2 --depth clone calls fetch_mpack (presign-only policy; no stream path). |
| 7 | C3 clone with depth=1 writes a .muse/shallow file listing boundary commits. |
| 8 | C4 clone with depth=None (full) does NOT write .muse/shallow. |
| 9 | C5 blobs_written reflects only blobs from depth-limited commits. |
| 10 | C6 --depth 0 is rejected with a user-friendly error (depth must be >= 1). |
| 11 | C7 JSON output includes a shallow_commits field when depth is active. |
| 12 | """ |
| 13 | from __future__ import annotations |
| 14 | |
| 15 | import argparse |
| 16 | import json |
| 17 | import pathlib |
| 18 | from unittest.mock import MagicMock, patch, call |
| 19 | |
| 20 | import pytest |
| 21 | |
| 22 | from muse.cli.commands.clone import register |
| 23 | from muse.core.types import MsgpackDict, long_id |
| 24 | from muse.core.paths import muse_dir |
| 25 | |
| 26 | type BranchHeads = dict[str, str] |
| 27 | |
| 28 | |
| 29 | # --------------------------------------------------------------------------- |
| 30 | # Helpers |
| 31 | # --------------------------------------------------------------------------- |
| 32 | |
| 33 | def _make_parser() -> tuple[argparse.ArgumentParser, argparse.ArgumentParser]: |
| 34 | root = argparse.ArgumentParser() |
| 35 | sub = root.add_subparsers() |
| 36 | clone_parser = sub.add_parser("clone") |
| 37 | register(sub) |
| 38 | return root |
| 39 | |
| 40 | |
| 41 | def _clone_parser() -> argparse.ArgumentParser: |
| 42 | root = argparse.ArgumentParser() |
| 43 | subs = root.add_subparsers(dest="cmd") |
| 44 | register(subs) |
| 45 | return root |
| 46 | |
| 47 | |
| 48 | def _fake_fetch_result( |
| 49 | *, |
| 50 | commits: list[MsgpackDict] | None = None, |
| 51 | snapshots: list[MsgpackDict] | None = None, |
| 52 | objects_received: int = 0, |
| 53 | branch_heads: BranchHeads | None = None, |
| 54 | shallow_commits: list[str] | None = None, |
| 55 | ) -> MsgpackDict: |
| 56 | return { |
| 57 | "repo_id": long_id("a" * 64), |
| 58 | "domain": "code", |
| 59 | "default_branch": "main", |
| 60 | "branch_heads": branch_heads or {"main": long_id("b" * 64)}, |
| 61 | "commits": commits or [], |
| 62 | "snapshots": snapshots or [], |
| 63 | "blobs": [], |
| 64 | "blobs_received": objects_received, |
| 65 | "shallow_commits": shallow_commits or [], |
| 66 | } |
| 67 | |
| 68 | |
| 69 | def _fake_apply_result(commits_written: int = 0) -> MsgpackDict: |
| 70 | return { |
| 71 | "commits_written": commits_written, |
| 72 | "snapshots_written": 0, |
| 73 | "blobs_written": 0, |
| 74 | "blobs_skipped": 0, |
| 75 | "tags_written": 0, |
| 76 | "failed_blobs": [], |
| 77 | "skipped_snapshots": [], |
| 78 | } |
| 79 | |
| 80 | |
| 81 | def _fake_remote_info(branch_heads: BranchHeads | None = None) -> MsgpackDict: |
| 82 | heads = branch_heads or {"main": long_id("b" * 64)} |
| 83 | return { |
| 84 | "repo_id": long_id("a" * 64), |
| 85 | "domain": "code", |
| 86 | "default_branch": "main", |
| 87 | "branch_heads": heads, |
| 88 | } |
| 89 | |
| 90 | |
| 91 | # --------------------------------------------------------------------------- |
| 92 | # C1 — --depth flag is registered |
| 93 | # --------------------------------------------------------------------------- |
| 94 | |
| 95 | class TestDepthFlagRegistered: |
| 96 | def test_depth_flag_exists(self) -> None: |
| 97 | parser = _clone_parser() |
| 98 | args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x", "--depth", "1"]) |
| 99 | assert args.depth == 1 |
| 100 | |
| 101 | def test_depth_defaults_to_none(self) -> None: |
| 102 | parser = _clone_parser() |
| 103 | args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x"]) |
| 104 | assert args.depth is None |
| 105 | |
| 106 | def test_depth_shortflag(self) -> None: |
| 107 | parser = _clone_parser() |
| 108 | args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x", "--depth", "5"]) |
| 109 | assert args.depth == 5 |
| 110 | |
| 111 | |
| 112 | # --------------------------------------------------------------------------- |
| 113 | # C2 — --depth clone uses fetch_mpack (presign-only policy) |
| 114 | # --------------------------------------------------------------------------- |
| 115 | |
| 116 | class TestDepthForwardedToTransport: |
| 117 | def test_depth_1_calls_fetch_mpack(self, tmp_path: pathlib.Path) -> None: |
| 118 | dest = str(tmp_path / "cloned") |
| 119 | tip = long_id("b" * 64) |
| 120 | |
| 121 | mock_transport = MagicMock() |
| 122 | mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) |
| 123 | mock_transport.fetch_mpack.return_value = _fake_fetch_result( |
| 124 | branch_heads={"main": tip}, |
| 125 | shallow_commits=[tip], |
| 126 | ) |
| 127 | |
| 128 | with ( |
| 129 | patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), |
| 130 | patch("muse.cli.commands.clone.get_signing_identity", return_value=None), |
| 131 | patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(0)), |
| 132 | patch("muse.cli.commands.clone._restore_working_tree"), |
| 133 | patch("muse.cli.commands.clone.set_remote"), |
| 134 | patch("muse.cli.commands.clone.set_remote_head"), |
| 135 | patch("muse.cli.commands.clone.set_upstream"), |
| 136 | patch("muse.cli.commands.clone.write_branch_ref"), |
| 137 | ): |
| 138 | parser = _clone_parser() |
| 139 | args = parser.parse_args(["clone", "https://example.com/repo", dest, "--depth", "1"]) |
| 140 | args.func(args) |
| 141 | |
| 142 | mock_transport.fetch_mpack.assert_called_once() |
| 143 | |
| 144 | def test_no_depth_calls_fetch_mpack(self, tmp_path: pathlib.Path) -> None: |
| 145 | dest = str(tmp_path / "cloned") |
| 146 | tip = long_id("b" * 64) |
| 147 | |
| 148 | mock_transport = MagicMock() |
| 149 | mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) |
| 150 | mock_transport.fetch_mpack.return_value = _fake_fetch_result(branch_heads={"main": tip}) |
| 151 | |
| 152 | with ( |
| 153 | patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), |
| 154 | patch("muse.cli.commands.clone.get_signing_identity", return_value=None), |
| 155 | patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(0)), |
| 156 | patch("muse.cli.commands.clone._restore_working_tree"), |
| 157 | patch("muse.cli.commands.clone.set_remote"), |
| 158 | patch("muse.cli.commands.clone.set_remote_head"), |
| 159 | patch("muse.cli.commands.clone.set_upstream"), |
| 160 | patch("muse.cli.commands.clone.write_branch_ref"), |
| 161 | ): |
| 162 | parser = _clone_parser() |
| 163 | args = parser.parse_args(["clone", "https://example.com/repo", dest]) |
| 164 | args.func(args) |
| 165 | |
| 166 | mock_transport.fetch_mpack.assert_called_once() |
| 167 | |
| 168 | |
| 169 | # --------------------------------------------------------------------------- |
| 170 | # C3 — .muse/shallow written when depth is active |
| 171 | # --------------------------------------------------------------------------- |
| 172 | |
| 173 | class TestShallowFileWritten: |
| 174 | def test_shallow_file_written_on_depth_clone(self, tmp_path: pathlib.Path) -> None: |
| 175 | dest = tmp_path / "cloned" |
| 176 | tip = long_id("b" * 64) |
| 177 | |
| 178 | mock_transport = MagicMock() |
| 179 | mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) |
| 180 | mock_transport.fetch_mpack.return_value = _fake_fetch_result( |
| 181 | branch_heads={"main": tip}, |
| 182 | shallow_commits=[tip], |
| 183 | ) |
| 184 | |
| 185 | with ( |
| 186 | patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), |
| 187 | patch("muse.cli.commands.clone.get_signing_identity", return_value=None), |
| 188 | patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(1)), |
| 189 | patch("muse.cli.commands.clone._restore_working_tree"), |
| 190 | patch("muse.cli.commands.clone.set_remote"), |
| 191 | patch("muse.cli.commands.clone.set_remote_head"), |
| 192 | patch("muse.cli.commands.clone.set_upstream"), |
| 193 | patch("muse.cli.commands.clone.write_branch_ref"), |
| 194 | ): |
| 195 | parser = _clone_parser() |
| 196 | args = parser.parse_args(["clone", "https://example.com/repo", str(dest), "--depth", "1"]) |
| 197 | args.func(args) |
| 198 | |
| 199 | shallow_file = muse_dir(dest) / "shallow" |
| 200 | assert shallow_file.exists(), ".muse/shallow must be written for a shallow clone" |
| 201 | contents = shallow_file.read_text().strip().splitlines() |
| 202 | assert tip in contents |
| 203 | |
| 204 | |
| 205 | # --------------------------------------------------------------------------- |
| 206 | # C4 — no .muse/shallow on full clone |
| 207 | # --------------------------------------------------------------------------- |
| 208 | |
| 209 | class TestNoShallowFileOnFullClone: |
| 210 | def test_no_shallow_file_on_full_clone(self, tmp_path: pathlib.Path) -> None: |
| 211 | dest = tmp_path / "cloned" |
| 212 | tip = long_id("b" * 64) |
| 213 | |
| 214 | mock_transport = MagicMock() |
| 215 | mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) |
| 216 | mock_transport.fetch_mpack.return_value = _fake_fetch_result(branch_heads={"main": tip}) |
| 217 | |
| 218 | with ( |
| 219 | patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), |
| 220 | patch("muse.cli.commands.clone.get_signing_identity", return_value=None), |
| 221 | patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(1)), |
| 222 | patch("muse.cli.commands.clone._restore_working_tree"), |
| 223 | patch("muse.cli.commands.clone.set_remote"), |
| 224 | patch("muse.cli.commands.clone.set_remote_head"), |
| 225 | patch("muse.cli.commands.clone.set_upstream"), |
| 226 | patch("muse.cli.commands.clone.write_branch_ref"), |
| 227 | ): |
| 228 | parser = _clone_parser() |
| 229 | args = parser.parse_args(["clone", "https://example.com/repo", str(dest)]) |
| 230 | args.func(args) |
| 231 | |
| 232 | shallow_file = muse_dir(dest) / "shallow" |
| 233 | assert not shallow_file.exists(), ".muse/shallow must NOT exist for a full clone" |
| 234 | |
| 235 | |
| 236 | # --------------------------------------------------------------------------- |
| 237 | # C6 — depth=0 rejected |
| 238 | # --------------------------------------------------------------------------- |
| 239 | |
| 240 | class TestInvalidDepth: |
| 241 | def test_depth_zero_rejected(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 242 | dest = str(tmp_path / "cloned") |
| 243 | parser = _clone_parser() |
| 244 | args = parser.parse_args(["clone", "https://example.com/repo", dest, "--depth", "0"]) |
| 245 | |
| 246 | with pytest.raises(SystemExit) as exc_info: |
| 247 | args.func(args) |
| 248 | |
| 249 | assert exc_info.value.code != 0 |
| 250 | captured = capsys.readouterr() |
| 251 | assert "depth" in (captured.err + captured.out).lower() |
| 252 | |
| 253 | |
| 254 | # --------------------------------------------------------------------------- |
| 255 | # C7 — JSON output includes shallow_commits when depth active |
| 256 | # --------------------------------------------------------------------------- |
| 257 | |
| 258 | class TestShallowJsonOutput: |
| 259 | def test_json_includes_shallow_commits(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None: |
| 260 | dest = tmp_path / "cloned" |
| 261 | tip = long_id("b" * 64) |
| 262 | |
| 263 | mock_transport = MagicMock() |
| 264 | mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip}) |
| 265 | mock_transport.fetch_mpack.return_value = _fake_fetch_result( |
| 266 | branch_heads={"main": tip}, |
| 267 | shallow_commits=[tip], |
| 268 | ) |
| 269 | |
| 270 | with ( |
| 271 | patch("muse.cli.commands.clone.make_transport", return_value=mock_transport), |
| 272 | patch("muse.cli.commands.clone.get_signing_identity", return_value=None), |
| 273 | patch("muse.cli.commands.clone.apply_mpack", return_value=_fake_apply_result(1)), |
| 274 | patch("muse.cli.commands.clone._restore_working_tree"), |
| 275 | patch("muse.cli.commands.clone.set_remote"), |
| 276 | patch("muse.cli.commands.clone.set_remote_head"), |
| 277 | patch("muse.cli.commands.clone.set_upstream"), |
| 278 | patch("muse.cli.commands.clone.write_branch_ref"), |
| 279 | ): |
| 280 | parser = _clone_parser() |
| 281 | args = parser.parse_args([ |
| 282 | "clone", "https://example.com/repo", str(dest), |
| 283 | "--depth", "1", "--json", |
| 284 | ]) |
| 285 | args.func(args) |
| 286 | |
| 287 | out = capsys.readouterr().out.strip() |
| 288 | data = json.loads(out) |
| 289 | assert "shallow_commits" in data |
| 290 | assert tip in data["shallow_commits"] |