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