gabriel / muse public
test_clone_shallow.py python
279 lines 11.4 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
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=1 is passed to transport.fetch_stream in the request body.
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 objects_written reflects only objects 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 "objects_received": objects_received,
64 "shallow_commits": shallow_commits or [],
65 }
66
67
68 def _fake_remote_info(branch_heads: BranchHeads | None = None) -> MsgpackDict:
69 heads = branch_heads or {"main": long_id("b" * 64)}
70 return {
71 "repo_id": long_id("a" * 64),
72 "domain": "code",
73 "default_branch": "main",
74 "branch_heads": heads,
75 }
76
77
78 # ---------------------------------------------------------------------------
79 # C1 — --depth flag is registered
80 # ---------------------------------------------------------------------------
81
82 class TestDepthFlagRegistered:
83 def test_depth_flag_exists(self) -> None:
84 parser = _clone_parser()
85 args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x", "--depth", "1"])
86 assert args.depth == 1
87
88 def test_depth_defaults_to_none(self) -> None:
89 parser = _clone_parser()
90 args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x"])
91 assert args.depth is None
92
93 def test_depth_shortflag(self) -> None:
94 parser = _clone_parser()
95 args = parser.parse_args(["clone", "https://example.com/repo", "/tmp/x", "--depth", "5"])
96 assert args.depth == 5
97
98
99 # ---------------------------------------------------------------------------
100 # C2 — depth is forwarded to transport.fetch_stream
101 # ---------------------------------------------------------------------------
102
103 class TestDepthForwardedToTransport:
104 def test_depth_1_passed_to_fetch_stream(self, tmp_path: pathlib.Path) -> None:
105 dest = str(tmp_path / "cloned")
106 tip = long_id("b" * 64)
107
108 mock_transport = MagicMock()
109 mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip})
110 mock_transport.fetch_stream.return_value = _fake_fetch_result(
111 branch_heads={"main": tip},
112 shallow_commits=[tip],
113 )
114
115 with (
116 patch("muse.cli.commands.clone.make_transport", return_value=mock_transport),
117 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
118 patch("muse.cli.commands.clone.apply_mpack", return_value={"commits_written": 0}),
119 patch("muse.cli.commands.clone._restore_working_tree"),
120 patch("muse.cli.commands.clone.set_remote"),
121 patch("muse.cli.commands.clone.set_remote_head"),
122 patch("muse.cli.commands.clone.set_upstream"),
123 patch("muse.cli.commands.clone.write_branch_ref"),
124 ):
125 parser = _clone_parser()
126 args = parser.parse_args(["clone", "https://example.com/repo", dest, "--depth", "1"])
127 args.func(args)
128
129 mock_transport.fetch_stream.assert_called_once()
130 _, kwargs = mock_transport.fetch_stream.call_args
131 assert kwargs.get("depth") == 1
132
133 def test_no_depth_passes_none_to_fetch_stream(self, tmp_path: pathlib.Path) -> None:
134 dest = str(tmp_path / "cloned")
135 tip = long_id("b" * 64)
136
137 mock_transport = MagicMock()
138 mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip})
139 mock_transport.fetch_presign_or_stream.return_value = _fake_fetch_result(branch_heads={"main": tip})
140
141 with (
142 patch("muse.cli.commands.clone.make_transport", return_value=mock_transport),
143 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
144 patch("muse.cli.commands.clone.apply_mpack", return_value={"commits_written": 0}),
145 patch("muse.cli.commands.clone._restore_working_tree"),
146 patch("muse.cli.commands.clone.set_remote"),
147 patch("muse.cli.commands.clone.set_remote_head"),
148 patch("muse.cli.commands.clone.set_upstream"),
149 patch("muse.cli.commands.clone.write_branch_ref"),
150 ):
151 parser = _clone_parser()
152 args = parser.parse_args(["clone", "https://example.com/repo", dest])
153 args.func(args)
154
155 mock_transport.fetch_presign_or_stream.assert_called_once()
156
157
158 # ---------------------------------------------------------------------------
159 # C3 — .muse/shallow written when depth is active
160 # ---------------------------------------------------------------------------
161
162 class TestShallowFileWritten:
163 def test_shallow_file_written_on_depth_clone(self, tmp_path: pathlib.Path) -> None:
164 dest = tmp_path / "cloned"
165 tip = long_id("b" * 64)
166
167 mock_transport = MagicMock()
168 mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip})
169 mock_transport.fetch_stream.return_value = _fake_fetch_result(
170 branch_heads={"main": tip},
171 shallow_commits=[tip],
172 )
173
174 with (
175 patch("muse.cli.commands.clone.make_transport", return_value=mock_transport),
176 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
177 patch("muse.cli.commands.clone.apply_mpack", return_value={"commits_written": 1}),
178 patch("muse.cli.commands.clone._restore_working_tree"),
179 patch("muse.cli.commands.clone.set_remote"),
180 patch("muse.cli.commands.clone.set_remote_head"),
181 patch("muse.cli.commands.clone.set_upstream"),
182 patch("muse.cli.commands.clone.write_branch_ref"),
183 ):
184 parser = _clone_parser()
185 args = parser.parse_args(["clone", "https://example.com/repo", str(dest), "--depth", "1"])
186 args.func(args)
187
188 shallow_file = muse_dir(dest) / "shallow"
189 assert shallow_file.exists(), ".muse/shallow must be written for a shallow clone"
190 contents = shallow_file.read_text().strip().splitlines()
191 assert tip in contents
192
193
194 # ---------------------------------------------------------------------------
195 # C4 — no .muse/shallow on full clone
196 # ---------------------------------------------------------------------------
197
198 class TestNoShallowFileOnFullClone:
199 def test_no_shallow_file_on_full_clone(self, tmp_path: pathlib.Path) -> None:
200 dest = tmp_path / "cloned"
201 tip = long_id("b" * 64)
202
203 mock_transport = MagicMock()
204 mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip})
205 mock_transport.fetch_presign_or_stream.return_value = _fake_fetch_result(branch_heads={"main": tip})
206
207 with (
208 patch("muse.cli.commands.clone.make_transport", return_value=mock_transport),
209 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
210 patch("muse.cli.commands.clone.apply_mpack", return_value={"commits_written": 1}),
211 patch("muse.cli.commands.clone._restore_working_tree"),
212 patch("muse.cli.commands.clone.set_remote"),
213 patch("muse.cli.commands.clone.set_remote_head"),
214 patch("muse.cli.commands.clone.set_upstream"),
215 patch("muse.cli.commands.clone.write_branch_ref"),
216 ):
217 parser = _clone_parser()
218 args = parser.parse_args(["clone", "https://example.com/repo", str(dest)])
219 args.func(args)
220
221 shallow_file = muse_dir(dest) / "shallow"
222 assert not shallow_file.exists(), ".muse/shallow must NOT exist for a full clone"
223
224
225 # ---------------------------------------------------------------------------
226 # C6 — depth=0 rejected
227 # ---------------------------------------------------------------------------
228
229 class TestInvalidDepth:
230 def test_depth_zero_rejected(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
231 dest = str(tmp_path / "cloned")
232 parser = _clone_parser()
233 args = parser.parse_args(["clone", "https://example.com/repo", dest, "--depth", "0"])
234
235 with pytest.raises(SystemExit) as exc_info:
236 args.func(args)
237
238 assert exc_info.value.code != 0
239 captured = capsys.readouterr()
240 assert "depth" in (captured.err + captured.out).lower()
241
242
243 # ---------------------------------------------------------------------------
244 # C7 — JSON output includes shallow_commits when depth active
245 # ---------------------------------------------------------------------------
246
247 class TestShallowJsonOutput:
248 def test_json_includes_shallow_commits(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
249 dest = tmp_path / "cloned"
250 tip = long_id("b" * 64)
251
252 mock_transport = MagicMock()
253 mock_transport.fetch_remote_info.return_value = _fake_remote_info({"main": tip})
254 mock_transport.fetch_stream.return_value = _fake_fetch_result(
255 branch_heads={"main": tip},
256 shallow_commits=[tip],
257 )
258
259 with (
260 patch("muse.cli.commands.clone.make_transport", return_value=mock_transport),
261 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
262 patch("muse.cli.commands.clone.apply_mpack", return_value={"commits_written": 1}),
263 patch("muse.cli.commands.clone._restore_working_tree"),
264 patch("muse.cli.commands.clone.set_remote"),
265 patch("muse.cli.commands.clone.set_remote_head"),
266 patch("muse.cli.commands.clone.set_upstream"),
267 patch("muse.cli.commands.clone.write_branch_ref"),
268 ):
269 parser = _clone_parser()
270 args = parser.parse_args([
271 "clone", "https://example.com/repo", str(dest),
272 "--depth", "1", "--json",
273 ])
274 args.func(args)
275
276 out = capsys.readouterr().out.strip()
277 data = json.loads(out)
278 assert "shallow_commits" in data
279 assert tip in data["shallow_commits"]
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago