gabriel / muse public
test_cmd_clone_hardening.py python
880 lines 37.3 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 121 days ago
1 """Hardening tests for ``muse clone``.
2
3 Coverage
4 --------
5 Unit
6 - _infer_dir_name: normal URL, trailing slash, query/fragment stripped,
7 path-traversal blocked, bare host, dot/double-dot fallback
8 - _init_muse_dir: all _CLONE_SUBDIRS created, repo.json written, HEAD set,
9 config.toml written, tags/ regression, superset of init subdirs
10 - _restore_working_tree: commit None warns, snapshot None warns, happy path
11
12 Integration (mocked transport — uses fetch_stream / on_object callback)
13 - already-exists guard raises USER_ERROR
14 - signing identity forwarded to fetch_remote_info and fetch_stream
15 - domain fallback to "code" (not "midi")
16 - branch fallback to first available when requested branch is missing
17 - commits_received comes from apply_result["commits_written"], not bundle
18 - --no-checkout skips apply_manifest
19 - branch refs written for every remote branch
20 - target directory removed on fetch_stream failure
21 - target directory removed on _init_muse_dir failure
22
23 Security
24 - ANSI injection in URL stripped in stderr output
25 - ANSI injection in branch name stripped
26 - Path-traversal URL blocked by _infer_dir_name
27 - All progress/error messages go to stderr, not stdout
28 - already-exists error on stderr, stdout empty
29
30 E2E (via CliRunner)
31 - --dry-run exits 0, no filesystem changes
32 - --dry-run --json correct schema
33 - --json cloned schema correct (all keys present, objects_written from on_object)
34 - --format json equivalent to --json
35 - --no-checkout flag accepted
36 - transport error exits non-zero
37 - empty repository exits non-zero
38
39 Performance / Stress
40 - 8 concurrent clones into isolated directories do not interfere
41 """
42
43 from __future__ import annotations
44
45 import json
46 import pathlib
47 import threading
48 from collections.abc import Callable
49 from typing import TYPE_CHECKING
50 from unittest.mock import MagicMock, patch
51
52 import pytest
53
54 from tests.cli_test_helper import CliRunner, InvokeResult
55 from muse.core.types import Manifest, blob_id
56
57 if TYPE_CHECKING:
58 from muse.cli.commands.clone import _CloneJson
59 from muse.core.pack import ApplyResult, RemoteInfo
60 from muse.core.transport import FetchStreamResult, SigningIdentity
61
62 cli = None
63 runner = CliRunner()
64
65 from muse.core.types import long_id
66 COMMIT_ID = long_id("a" * 64)
67 SNAP_ID = long_id("b" * 64)
68 REPO_ID = "test-repo-id"
69
70
71 # ── typed helpers ─────────────────────────────────────────────────────────────
72
73 def _make_apply_result(
74 commits_written: int = 5,
75 objects_written: int = 11,
76 ) -> "ApplyResult":
77 from muse.core.pack import ApplyResult
78 return ApplyResult(
79 commits_written=commits_written,
80 snapshots_written=commits_written,
81 objects_written=objects_written,
82 objects_skipped=0,
83 )
84
85
86 def _make_remote_info(
87 branch_heads: Manifest | None = None,
88 domain: str = "code",
89 default_branch: str = "main",
90 repo_id: str = REPO_ID,
91 ) -> "RemoteInfo":
92 effective_heads: Manifest = (
93 {"main": COMMIT_ID} if branch_heads is None else branch_heads
94 )
95 return {
96 "repo_id": repo_id,
97 "domain": domain,
98 "default_branch": default_branch,
99 "branch_heads": effective_heads,
100 }
101
102
103 def _make_fetch_stream_result(
104 commits: list[str] | None = None,
105 snapshots: list[str] | None = None,
106 objects_received: int = 0,
107 ) -> "FetchStreamResult":
108 return {
109 "commits": commits or [],
110 "snapshots": snapshots or [],
111 "objects_received": objects_received,
112 }
113
114
115 def _make_transport_mock(
116 branch_heads: Manifest | None = None,
117 domain: str = "code",
118 objects_count: int = 11,
119 ) -> MagicMock:
120 """Return a transport mock whose fetch_stream dispatches objects via on_object."""
121 t = MagicMock()
122 t.fetch_remote_info.return_value = _make_remote_info(branch_heads, domain=domain)
123
124 def _fetch_stream(url: str, signing: "SigningIdentity | None", want: list[str], have: list[str], on_object: Callable[..., None] | None = None, **kwargs: str) -> "FetchStreamResult":
125 if callable(on_object):
126 for i in range(objects_count):
127 content = f"fake-blob-{i}".encode()
128 oid = blob_id(content)
129 on_object({"object_id": oid, "content": content, "path": f"file{i}.txt"})
130 return _make_fetch_stream_result(objects_received=objects_count)
131
132 t.fetch_stream.side_effect = _fetch_stream
133 t.fetch_presign_or_stream.side_effect = _fetch_stream
134 return t
135
136
137 def _json_line(result: InvokeResult) -> "_CloneJson":
138 for line in result.output.splitlines():
139 stripped = line.strip()
140 if stripped.startswith("{"):
141 parsed: _CloneJson = json.loads(stripped)
142 return parsed
143 raise ValueError(f"No JSON line in output:\n{result.output!r}")
144
145
146 # ── Unit: _infer_dir_name ─────────────────────────────────────────────────────
147
148 class TestInferDirName:
149 def test_last_url_segment(self) -> None:
150 from muse.cli.commands.clone import _infer_dir_name
151 assert _infer_dir_name("http://hub.muse.ai/gabriel/my-repo") == "my-repo"
152
153 def test_trailing_slash_stripped(self) -> None:
154 from muse.cli.commands.clone import _infer_dir_name
155 assert _infer_dir_name("http://hub.muse.ai/gabriel/my-repo/") == "my-repo"
156
157 def test_query_string_stripped(self) -> None:
158 from muse.cli.commands.clone import _infer_dir_name
159 assert _infer_dir_name("http://hub.muse.ai/repo?token=abc") == "repo"
160
161 def test_fragment_stripped(self) -> None:
162 from muse.cli.commands.clone import _infer_dir_name
163 assert _infer_dir_name("http://hub.muse.ai/repo#section") == "repo"
164
165 def test_bare_host_uses_hostname(self) -> None:
166 from muse.cli.commands.clone import _infer_dir_name
167 result = _infer_dir_name("http://hub.muse.ai/")
168 assert result == "hub.muse.ai"
169 assert ".." not in result
170
171 def test_path_traversal_blocked(self) -> None:
172 from muse.cli.commands.clone import _infer_dir_name
173 result = _infer_dir_name("http://attacker.example.com/../../../../etc/passwd")
174 assert ".." not in result
175 assert "/" not in result
176 assert result != ""
177
178 def test_dot_only_falls_back(self) -> None:
179 from muse.cli.commands.clone import _infer_dir_name
180 result = _infer_dir_name("http://example.com/.")
181 assert result not in (".", "..")
182
183 def test_double_dot_falls_back(self) -> None:
184 from muse.cli.commands.clone import _infer_dir_name
185 result = _infer_dir_name("http://example.com/..")
186 assert result not in (".", "..")
187 assert result == "muse-repo"
188
189
190 # ── Unit: _init_muse_dir ──────────────────────────────────────────────────────
191
192 class TestInitMuseDir:
193 def test_all_clone_subdirs_created(self, tmp_path: pathlib.Path) -> None:
194 from muse.cli.commands.clone import _CLONE_SUBDIRS, _init_muse_dir
195 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
196 dot_muse = muse_dir(tmp_path)
197 for subdir in _CLONE_SUBDIRS:
198 assert (dot_muse / subdir).is_dir(), f"Missing subdir: {subdir}"
199
200 def test_tags_dir_created(self, tmp_path: pathlib.Path) -> None:
201 from muse.cli.commands.clone import _init_muse_dir
202 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
203 assert (tags_dir(tmp_path)).is_dir()
204
205 def test_repo_json_written(self, tmp_path: pathlib.Path) -> None:
206 from muse.cli.commands.clone import _init_muse_dir
207 _init_muse_dir(tmp_path, "my-repo-id", "code", "main")
208 meta = json.loads((repo_json_path(tmp_path)).read_text())
209 assert meta["repo_id"] == "my-repo-id"
210 assert meta["domain"] == "code"
211 assert "schema_version" in meta
212 assert "created_at" in meta
213
214 def test_head_file_written(self, tmp_path: pathlib.Path) -> None:
215 from muse.cli.commands.clone import _init_muse_dir
216 _init_muse_dir(tmp_path, REPO_ID, "code", "dev")
217 head = (head_path(tmp_path)).read_text()
218 assert "dev" in head
219
220 def test_default_branch_ref_created(self, tmp_path: pathlib.Path) -> None:
221 from muse.cli.commands.clone import _init_muse_dir
222 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
223 assert (heads_dir(tmp_path) / "main").exists()
224
225 def test_config_toml_written(self, tmp_path: pathlib.Path) -> None:
226 from muse.cli.commands.clone import _init_muse_dir
227 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
228 config = (config_toml_path(tmp_path)).read_text()
229 assert "[remotes]" in config
230
231 def test_superset_of_init_subdirs(self, tmp_path: pathlib.Path) -> None:
232 from muse.cli.commands.clone import _CLONE_SUBDIRS
233 from muse.cli.commands.init import _INIT_SUBDIRS
234 clone_set = set(_CLONE_SUBDIRS)
235 for subdir in _INIT_SUBDIRS:
236 assert subdir in clone_set, (
237 f"_INIT_SUBDIRS has '{subdir}' but _CLONE_SUBDIRS does not"
238 )
239
240
241 # ── Unit: _restore_working_tree ───────────────────────────────────────────────
242
243 class TestRestoreWorkingTree:
244 def test_warns_when_commit_not_found(
245 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
246 ) -> None:
247 from muse.cli.commands.clone import _restore_working_tree
248 with patch("muse.cli.commands.clone.read_commit", return_value=None):
249 with patch("muse.cli.commands.clone.apply_manifest") as am:
250 _restore_working_tree(tmp_path, long_id("a" * 64))
251 am.assert_not_called()
252
253 def test_warns_when_snapshot_not_found(
254 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
255 ) -> None:
256 from muse.cli.commands.clone import _restore_working_tree
257 fake_commit = MagicMock()
258 fake_commit.snapshot_id = SNAP_ID
259 with (
260 patch("muse.cli.commands.clone.read_commit", return_value=fake_commit),
261 patch("muse.cli.commands.clone.read_snapshot", return_value=None),
262 ):
263 with patch("muse.cli.commands.clone.apply_manifest") as am:
264 _restore_working_tree(tmp_path, COMMIT_ID)
265 am.assert_not_called()
266
267 def test_happy_path_calls_apply_manifest(self, tmp_path: pathlib.Path) -> None:
268 from muse.cli.commands.clone import _restore_working_tree
269 fake_commit = MagicMock()
270 fake_commit.snapshot_id = SNAP_ID
271 fake_snap = MagicMock()
272 fake_snap.manifest = {"file.txt": "aabbcc"}
273 with (
274 patch("muse.cli.commands.clone.read_commit", return_value=fake_commit),
275 patch("muse.cli.commands.clone.read_snapshot", return_value=fake_snap),
276 ):
277 with patch("muse.cli.commands.clone.apply_manifest") as am:
278 _restore_working_tree(tmp_path, COMMIT_ID)
279 am.assert_called_once_with(tmp_path, {}, {"file.txt": "aabbcc"})
280
281
282 # ── Integration: run() with mocked transport ──────────────────────────────────
283
284 def _invoke_run(
285 tmp_path: pathlib.Path,
286 url: str = "http://localhost:19999/gabriel/repo",
287 branch: str | None = None,
288 dry_run: bool = False,
289 no_checkout: bool = False,
290 json_out: bool = False,
291 transport: MagicMock | None = None,
292 apply_result: "ApplyResult | None" = None,
293 signing: "SigningIdentity | None" = None,
294 objects_count: int = 11,
295 ) -> None:
296 import argparse
297 from muse.cli.commands.clone import run
298 t = transport or _make_transport_mock(objects_count=objects_count)
299 ar = apply_result or _make_apply_result()
300 args = argparse.Namespace(
301 url=url,
302 directory=str(tmp_path / "cloned"),
303 branch=branch,
304 dry_run=dry_run,
305 no_checkout=no_checkout,
306 json_out=json_out,
307 )
308 with (
309 patch("muse.cli.commands.clone.get_signing_identity", return_value=signing),
310 patch("muse.cli.commands.clone.make_transport", return_value=t),
311 patch("muse.cli.commands.clone.apply_mpack", return_value=ar),
312 patch("muse.cli.commands.clone.write_object", return_value=True),
313 patch("muse.cli.commands.clone.set_remote"),
314 patch("muse.cli.commands.clone.set_remote_head"),
315 patch("muse.cli.commands.clone.set_upstream"),
316 patch("muse.cli.commands.clone.apply_manifest"),
317 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
318 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
319 ):
320 run(args)
321
322
323 class TestRunIntegration:
324 def test_already_exists_raises_user_error(self, tmp_path: pathlib.Path) -> None:
325 import argparse
326 from muse.cli.commands.clone import run
327 from muse.core.errors import ExitCode
328 target = tmp_path / "existing"
329 muse_dir(target).mkdir(parents=True)
330 args = argparse.Namespace(
331 url="http://localhost:19999/repo",
332 directory=str(target),
333 branch=None,
334 dry_run=False,
335 no_checkout=False,
336 json_out=False,
337 )
338 with pytest.raises(SystemExit) as exc:
339 with patch("muse.cli.commands.clone.get_signing_identity", return_value=None):
340 run(args)
341 assert exc.value.code == ExitCode.USER_ERROR
342
343 def test_signing_forwarded_to_transport(self, tmp_path: pathlib.Path) -> None:
344 """Signing identity must reach both fetch_remote_info and fetch_presign_or_stream."""
345 from muse.core.transport import SigningIdentity
346 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
347 signing = SigningIdentity("alice", Ed25519PrivateKey.generate())
348 t = _make_transport_mock()
349 _invoke_run(tmp_path, transport=t, signing=signing)
350 t.fetch_remote_info.assert_called_once_with(
351 "http://localhost:19999/gabriel/repo", signing=signing
352 )
353 t.fetch_presign_or_stream.assert_called_once()
354 call_kwargs = t.fetch_presign_or_stream.call_args
355 # signing is the second positional arg
356 assert call_kwargs.args[1] is signing or call_kwargs.kwargs.get("signing") is signing
357
358 def test_domain_defaults_to_code_not_midi(self, tmp_path: pathlib.Path) -> None:
359 t = _make_transport_mock()
360 t.fetch_remote_info.return_value = _make_remote_info(domain="")
361 target = tmp_path / "cloned"
362 import argparse
363 from muse.cli.commands.clone import run
364 args = argparse.Namespace(
365 url="http://localhost:19999/repo",
366 directory=str(target),
367 branch=None,
368 dry_run=False,
369 no_checkout=False,
370 json_out=False,
371 )
372 with (
373 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
374 patch("muse.cli.commands.clone.make_transport", return_value=t),
375 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
376 patch("muse.cli.commands.clone.write_object", return_value=True),
377 patch("muse.cli.commands.clone.set_remote"),
378 patch("muse.cli.commands.clone.set_remote_head"),
379 patch("muse.cli.commands.clone.set_upstream"),
380 patch("muse.cli.commands.clone.apply_manifest"),
381 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
382 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
383 ):
384 run(args)
385 meta = json.loads((repo_json_path(target)).read_text())
386 assert meta["domain"] == "code", f"Expected 'code', got '{meta['domain']}'"
387
388 def test_commits_received_from_apply_result_not_bundle(
389 self, tmp_path: pathlib.Path
390 ) -> None:
391 """commits_received in JSON output must come from apply_result['commits_written']."""
392 output_lines: list[str] = []
393 import argparse
394 from muse.cli.commands.clone import run
395 ar = _make_apply_result(commits_written=7, objects_written=23)
396 t = _make_transport_mock(objects_count=0)
397 args = argparse.Namespace(
398 url="http://localhost:19999/repo",
399 directory=str(tmp_path / "cloned"),
400 branch=None,
401 dry_run=False,
402 no_checkout=False,
403 json_out=True,
404 )
405 with (
406 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
407 patch("muse.cli.commands.clone.make_transport", return_value=t),
408 patch("muse.cli.commands.clone.apply_mpack", return_value=ar),
409 patch("muse.cli.commands.clone.write_object", return_value=True),
410 patch("muse.cli.commands.clone.set_remote"),
411 patch("muse.cli.commands.clone.set_remote_head"),
412 patch("muse.cli.commands.clone.set_upstream"),
413 patch("muse.cli.commands.clone.apply_manifest"),
414 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
415 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
416 patch("builtins.print", side_effect=lambda *a, **kw: output_lines.append(str(a[0]) if a else "")),
417 ):
418 run(args)
419 json_out = next((l for l in output_lines if l.strip().startswith("{")), None)
420 assert json_out is not None
421 data: _CloneJson = json.loads(json_out)
422 assert data["commits_received"] == 7
423
424 def test_branch_fallback_to_first_available(
425 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
426 ) -> None:
427 import argparse
428 from muse.cli.commands.clone import run
429 t = _make_transport_mock(branch_heads={"dev": COMMIT_ID})
430 args = argparse.Namespace(
431 url="http://localhost:19999/repo",
432 directory=str(tmp_path / "cloned"),
433 branch="nonexistent",
434 dry_run=False,
435 no_checkout=True,
436 json_out=False,
437 )
438 with (
439 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
440 patch("muse.cli.commands.clone.make_transport", return_value=t),
441 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
442 patch("muse.cli.commands.clone.write_object", return_value=True),
443 patch("muse.cli.commands.clone.set_remote"),
444 patch("muse.cli.commands.clone.set_remote_head"),
445 patch("muse.cli.commands.clone.set_upstream"),
446 ):
447 run(args)
448 assert "nonexistent" in capsys.readouterr().err
449
450 def test_target_dir_removed_on_fetch_failure(
451 self, tmp_path: pathlib.Path
452 ) -> None:
453 """If fetch_stream raises TransportError, the target directory is removed."""
454 from muse.core.transport import TransportError
455 import argparse
456 from muse.cli.commands.clone import run
457 t = _make_transport_mock()
458 t.fetch_stream.side_effect = TransportError("connection refused", 503)
459 t.fetch_presign_or_stream.side_effect = TransportError("connection refused", 503)
460 target = tmp_path / "cloned"
461 args = argparse.Namespace(
462 url="http://localhost:19999/repo",
463 directory=str(target),
464 branch=None,
465 dry_run=False,
466 no_checkout=False,
467 json_out=False,
468 )
469 with (
470 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
471 patch("muse.cli.commands.clone.make_transport", return_value=t),
472 pytest.raises(SystemExit),
473 ):
474 run(args)
475 assert not target.exists(), "Target directory must be removed on failure"
476
477 def test_target_dir_removed_on_init_failure(
478 self, tmp_path: pathlib.Path
479 ) -> None:
480 import argparse
481 from muse.cli.commands.clone import run
482 args = argparse.Namespace(
483 url="http://localhost:19999/repo",
484 directory=str(tmp_path / "cloned"),
485 branch=None,
486 dry_run=False,
487 no_checkout=False,
488 json_out=False,
489 )
490 with (
491 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
492 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
493 patch("muse.cli.commands.clone._init_muse_dir", side_effect=OSError("disk full")),
494 pytest.raises(SystemExit),
495 ):
496 run(args)
497 assert not (tmp_path / "cloned").exists()
498
499 def test_no_checkout_skips_apply_manifest(self, tmp_path: pathlib.Path) -> None:
500 t = _make_transport_mock()
501 apply_mock = MagicMock()
502 import argparse
503 from muse.cli.commands.clone import run
504 args = argparse.Namespace(
505 url="http://localhost:19999/repo",
506 directory=str(tmp_path / "cloned"),
507 branch=None,
508 dry_run=False,
509 no_checkout=True,
510 json_out=False,
511 )
512 with (
513 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
514 patch("muse.cli.commands.clone.make_transport", return_value=t),
515 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
516 patch("muse.cli.commands.clone.write_object", return_value=True),
517 patch("muse.cli.commands.clone.set_remote"),
518 patch("muse.cli.commands.clone.set_remote_head"),
519 patch("muse.cli.commands.clone.set_upstream"),
520 patch("muse.cli.commands.clone.apply_manifest", apply_mock),
521 ):
522 run(args)
523 apply_mock.assert_not_called()
524
525 def test_branch_refs_written_for_every_remote_branch(
526 self, tmp_path: pathlib.Path
527 ) -> None:
528 branches = {"main": long_id("a" * 64), "dev": long_id("b" * 64), "feat/x": long_id("c" * 64)}
529 t = _make_transport_mock(branch_heads=branches)
530 import argparse
531 from muse.cli.commands.clone import run
532 target = tmp_path / "cloned"
533 args = argparse.Namespace(
534 url="http://localhost:19999/repo",
535 directory=str(target),
536 branch=None,
537 dry_run=False,
538 no_checkout=True,
539 json_out=False,
540 )
541 with (
542 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
543 patch("muse.cli.commands.clone.make_transport", return_value=t),
544 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
545 patch("muse.cli.commands.clone.write_object", return_value=True),
546 patch("muse.cli.commands.clone.set_remote"),
547 patch("muse.cli.commands.clone.set_remote_head"),
548 patch("muse.cli.commands.clone.set_upstream"),
549 ):
550 run(args)
551 for branch, cid in branches.items():
552 ref_file = ref_path(target, branch)
553 assert ref_file.exists(), f"Missing ref file for branch {branch}"
554 assert ref_file.read_text() == cid
555
556
557 # ── Security ──────────────────────────────────────────────────────────────────
558
559 class TestSecurity:
560 def test_ansi_in_url_stripped_in_stderr(
561 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
562 ) -> None:
563 malicious_url = "\x1b[31mhttp://attacker.example.com/repo\x1b[0m"
564 import argparse
565 from muse.cli.commands.clone import run
566 from muse.core.transport import TransportError
567 t = MagicMock()
568 t.fetch_remote_info.side_effect = TransportError("refused", 503)
569 args = argparse.Namespace(
570 url=malicious_url,
571 directory=str(tmp_path / "cloned"),
572 branch=None,
573 dry_run=False,
574 no_checkout=False,
575 json_out=False,
576 )
577 with (
578 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
579 patch("muse.cli.commands.clone.make_transport", return_value=t),
580 pytest.raises(SystemExit),
581 ):
582 run(args)
583 assert "\x1b[" not in capsys.readouterr().err
584
585 def test_ansi_in_branch_name_stripped(
586 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
587 ) -> None:
588 malicious_branch = "\x1b[32mHACKED\x1b[0m"
589 t = _make_transport_mock(branch_heads={"main": COMMIT_ID})
590 import argparse
591 from muse.cli.commands.clone import run
592 args = argparse.Namespace(
593 url="http://localhost:19999/repo",
594 directory=str(tmp_path / "cloned"),
595 branch=malicious_branch,
596 dry_run=False,
597 no_checkout=True,
598 json_out=False,
599 )
600 with (
601 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
602 patch("muse.cli.commands.clone.make_transport", return_value=t),
603 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
604 patch("muse.cli.commands.clone.write_object", return_value=True),
605 patch("muse.cli.commands.clone.set_remote"),
606 patch("muse.cli.commands.clone.set_remote_head"),
607 patch("muse.cli.commands.clone.set_upstream"),
608 ):
609 run(args)
610 assert "\x1b[" not in capsys.readouterr().err
611
612 def test_path_traversal_url_blocked(self) -> None:
613 from muse.cli.commands.clone import _infer_dir_name
614 result = _infer_dir_name("http://attacker.example.com/../../../etc/passwd")
615 assert ".." not in result
616 assert "etc" not in result or result == "etc"
617
618 def test_all_diagnostics_go_to_stderr_not_stdout(
619 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
620 ) -> None:
621 t = _make_transport_mock()
622 import argparse
623 from muse.cli.commands.clone import run
624 args = argparse.Namespace(
625 url="http://localhost:19999/repo",
626 directory=str(tmp_path / "cloned"),
627 branch=None,
628 dry_run=False,
629 no_checkout=True,
630 json_out=False,
631 )
632 with (
633 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
634 patch("muse.cli.commands.clone.make_transport", return_value=t),
635 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
636 patch("muse.cli.commands.clone.write_object", return_value=True),
637 patch("muse.cli.commands.clone.set_remote"),
638 patch("muse.cli.commands.clone.set_remote_head"),
639 patch("muse.cli.commands.clone.set_upstream"),
640 ):
641 run(args)
642 assert capsys.readouterr().out == ""
643
644 def test_already_exists_message_on_stderr(
645 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
646 ) -> None:
647 import argparse
648 from muse.cli.commands.clone import run
649 target = tmp_path / "existing"
650 muse_dir(target).mkdir(parents=True)
651 args = argparse.Namespace(
652 url="http://localhost:19999/repo",
653 directory=str(target),
654 branch=None,
655 dry_run=False,
656 no_checkout=False,
657 json_out=False,
658 )
659 with (
660 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
661 pytest.raises(SystemExit),
662 ):
663 run(args)
664 cap = capsys.readouterr()
665 assert "already" in cap.err.lower()
666 assert cap.out == ""
667
668
669 # ── E2E: via CliRunner ────────────────────────────────────────────────────────
670
671 def _invoke(*args: str, transport: MagicMock | None = None, tmp_path: pathlib.Path | None = None) -> InvokeResult:
672 t = transport or _make_transport_mock()
673 with (
674 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
675 patch("muse.cli.commands.clone.make_transport", return_value=t),
676 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
677 patch("muse.cli.commands.clone.write_object", return_value=True),
678 patch("muse.cli.commands.clone.set_remote"),
679 patch("muse.cli.commands.clone.set_remote_head"),
680 patch("muse.cli.commands.clone.set_upstream"),
681 patch("muse.cli.commands.clone.apply_manifest"),
682 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
683 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
684 ):
685 return runner.invoke(
686 cli,
687 ["clone", "http://localhost:19999/gabriel/repo", *args],
688 )
689
690
691 class TestCLIClone:
692 def test_dry_run_exits_zero_no_fs_changes(self, tmp_path: pathlib.Path) -> None:
693 target = tmp_path / "should-not-exist"
694 result = runner.invoke(
695 cli,
696 ["clone", "http://localhost:19999/repo", str(target), "--dry-run"],
697 )
698 assert not target.exists()
699
700 def test_dry_run_json_schema(self, tmp_path: pathlib.Path) -> None:
701 with (
702 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
703 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
704 ):
705 result = runner.invoke(
706 cli,
707 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--dry-run", "--json"],
708 )
709 assert result.exit_code == 0, result.output
710 data = _json_line(result)
711 assert data["status"] == "dry_run"
712 assert data["dry_run"] is True
713 assert data["head"] == COMMIT_ID
714 assert data["domain"] == "code"
715
716 def test_json_cloned_schema(self, tmp_path: pathlib.Path) -> None:
717 """All expected keys present; objects_written counted via on_object callback."""
718 with (
719 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
720 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock(objects_count=11)),
721 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result(commits_written=5)),
722 patch("muse.cli.commands.clone.write_object", return_value=True),
723 patch("muse.cli.commands.clone.set_remote"),
724 patch("muse.cli.commands.clone.set_remote_head"),
725 patch("muse.cli.commands.clone.set_upstream"),
726 patch("muse.cli.commands.clone.apply_manifest"),
727 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
728 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
729 ):
730 result = runner.invoke(
731 cli,
732 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--json"],
733 )
734 assert result.exit_code == 0, result.output
735 data = _json_line(result)
736 for key in ("status", "url", "directory", "branch", "commits_received", "objects_written", "head", "domain", "dry_run"):
737 assert key in data, f"Missing key: {key}"
738 assert data["status"] == "cloned"
739 assert data["dry_run"] is False
740 assert data["commits_received"] == 5
741 assert data["objects_written"] == 11
742
743 def test_no_checkout_flag_accepted(self, tmp_path: pathlib.Path) -> None:
744 with (
745 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
746 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
747 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
748 patch("muse.cli.commands.clone.write_object", return_value=True),
749 patch("muse.cli.commands.clone.set_remote"),
750 patch("muse.cli.commands.clone.set_remote_head"),
751 patch("muse.cli.commands.clone.set_upstream"),
752 ):
753 result = runner.invoke(
754 cli,
755 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--no-checkout"],
756 )
757 assert result.exit_code == 0, result.output
758
759 def test_transport_error_exits_nonzero(self) -> None:
760 from muse.core.transport import TransportError
761 t = MagicMock()
762 t.fetch_remote_info.side_effect = TransportError("refused", 503)
763 with (
764 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
765 patch("muse.cli.commands.clone.make_transport", return_value=t),
766 ):
767 result = runner.invoke(cli, ["clone", "http://localhost:19999/repo", "/tmp/muse-test-clone-xxx"])
768 assert result.exit_code != 0
769
770 def test_empty_repository_exits_nonzero(self) -> None:
771 t = MagicMock()
772 t.fetch_remote_info.return_value = _make_remote_info(branch_heads={})
773 with (
774 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
775 patch("muse.cli.commands.clone.make_transport", return_value=t),
776 ):
777 result = runner.invoke(cli, ["clone", "http://localhost:19999/repo", "/tmp/muse-test-clone-empty"])
778 assert result.exit_code != 0
779
780 def test_json_on_stdout_parseable(self, tmp_path: pathlib.Path) -> None:
781 with (
782 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
783 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
784 ):
785 result = runner.invoke(
786 cli,
787 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--dry-run", "--json"],
788 )
789 assert result.exit_code == 0
790 data = _json_line(result)
791 assert "status" in data
792
793
794 # ── Stress: concurrent clones into isolated directories ───────────────────────
795
796 class TestStressConcurrent:
797 def test_8_concurrent_clones_isolated(self, tmp_path: pathlib.Path) -> None:
798 """8 concurrent clone calls into separate directories must not interfere."""
799 import argparse
800 from muse.cli.commands.clone import run
801 errors: list[str] = []
802
803 def _do(idx: int) -> None:
804 try:
805 target = tmp_path / f"repo{idx}"
806 t = _make_transport_mock()
807 args = argparse.Namespace(
808 url=f"http://localhost:19999/repo{idx}",
809 directory=str(target),
810 branch=None,
811 dry_run=False,
812 no_checkout=True,
813 json_out=False,
814 )
815 with (
816 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
817 patch("muse.cli.commands.clone.make_transport", return_value=t),
818 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
819 patch("muse.cli.commands.clone.write_object", return_value=True),
820 patch("muse.cli.commands.clone.set_remote"),
821 patch("muse.cli.commands.clone.set_remote_head"),
822 patch("muse.cli.commands.clone.set_upstream"),
823 ):
824 run(args)
825 assert muse_dir(target).is_dir()
826 assert (tags_dir(target)).is_dir()
827 meta = json.loads((repo_json_path(target)).read_text())
828 assert meta["domain"] == "code"
829 except Exception as exc:
830 errors.append(f"Thread {idx}: {exc}")
831
832 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
833 for th in threads:
834 th.start()
835 for th in threads:
836 th.join()
837 assert errors == [], f"Concurrent clone failures:\n{'\n'.join(errors)}"
838
839
840 # ---------------------------------------------------------------------------
841 # Flag registration tests
842 # ---------------------------------------------------------------------------
843
844 import argparse as _argparse
845 from muse.cli.commands.clone import register as _register_clone
846 from muse.core.paths import config_toml_path, head_path, heads_dir, muse_dir, ref_path, repo_json_path, tags_dir
847
848
849 def _parse_clone(*args: str) -> _argparse.Namespace:
850 root_p = _argparse.ArgumentParser()
851 subs = root_p.add_subparsers(dest="cmd")
852 _register_clone(subs)
853 return root_p.parse_args(["clone", *args])
854
855
856 class TestRegisterFlags:
857 def test_default_json_out_is_false(self) -> None:
858 ns = _parse_clone("https://example.com/repo")
859 assert ns.json_out is False
860
861 def test_json_flag_sets_json_out(self) -> None:
862 ns = _parse_clone("https://example.com/repo", "--json")
863 assert ns.json_out is True
864
865 def test_j_shorthand_sets_json_out(self) -> None:
866 ns = _parse_clone("https://example.com/repo", "-j")
867 assert ns.json_out is True
868
869 def test_dry_run_flag(self) -> None:
870 ns = _parse_clone("https://example.com/repo", "--dry-run")
871 assert ns.dry_run is True
872
873 def test_n_shorthand_for_dry_run(self) -> None:
874 ns = _parse_clone("https://example.com/repo", "-n")
875 assert ns.dry_run is True
876
877 def test_format_flag_no_longer_exists(self) -> None:
878 import pytest
879 with pytest.raises(SystemExit):
880 _parse_clone("https://example.com/repo", "--format", "json")
File History 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 121 days ago