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