gabriel / muse public
test_cmd_clone_hardening.py python
996 lines 42.2 KB
Raw
sha256:a5f8a020b53ddab68f231cf27489fe865525f85ee924133034262d39b332076e fix(clone,pull): never claim success when a snapshot hash m… Sonnet 5 patch 1 day 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 """Happy path: every branch's tip commit landed cleanly, so every
525 branch ref gets written (muse#35's guard only withholds the ref when
526 a commit was NOT actually written — read_commit is mocked here to
527 simulate every commit having landed, since apply_mpack itself is
528 also mocked and never really writes anything to the store)."""
529 branches = {"main": long_id("a" * 64), "dev": long_id("b" * 64), "feat/x": long_id("c" * 64)}
530 t = _make_transport_mock(branch_heads=branches)
531 import argparse
532 from muse.cli.commands.clone import run
533 target = tmp_path / "cloned"
534 args = argparse.Namespace(
535 url="http://localhost:19999/repo",
536 directory=str(target),
537 branch=None,
538 dry_run=False,
539 no_checkout=True,
540 json_out=False,
541 )
542 with (
543 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
544 patch("muse.cli.commands.clone.make_transport", return_value=t),
545 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
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 patch("muse.cli.commands.clone.read_commit", return_value=object()),
550 ):
551 run(args)
552 for branch, cid in branches.items():
553 ref_file = ref_path(target, branch)
554 assert ref_file.exists(), f"Missing ref file for branch {branch}"
555 assert ref_file.read_text() == cid
556
557
558 # ── Security ──────────────────────────────────────────────────────────────────
559
560 class TestSecurity:
561 def test_ansi_in_url_stripped_in_stderr(
562 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
563 ) -> None:
564 malicious_url = "\x1b[31mhttp://attacker.example.com/repo\x1b[0m"
565 import argparse
566 from muse.cli.commands.clone import run
567 from muse.core.transport import TransportError
568 t = MagicMock()
569 t.fetch_remote_info.side_effect = TransportError("refused", 503)
570 args = argparse.Namespace(
571 url=malicious_url,
572 directory=str(tmp_path / "cloned"),
573 branch=None,
574 dry_run=False,
575 no_checkout=False,
576 json_out=False,
577 )
578 with (
579 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
580 patch("muse.cli.commands.clone.make_transport", return_value=t),
581 pytest.raises(SystemExit),
582 ):
583 run(args)
584 assert "\x1b[" not in capsys.readouterr().err
585
586 def test_ansi_in_branch_name_stripped(
587 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
588 ) -> None:
589 malicious_branch = "\x1b[32mHACKED\x1b[0m"
590 t = _make_transport_mock(branch_heads={"main": COMMIT_ID})
591 import argparse
592 from muse.cli.commands.clone import run
593 args = argparse.Namespace(
594 url="http://localhost:19999/repo",
595 directory=str(tmp_path / "cloned"),
596 branch=malicious_branch,
597 dry_run=False,
598 no_checkout=True,
599 json_out=False,
600 )
601 with (
602 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
603 patch("muse.cli.commands.clone.make_transport", return_value=t),
604 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
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.set_remote"),
637 patch("muse.cli.commands.clone.set_remote_head"),
638 patch("muse.cli.commands.clone.set_upstream"),
639 ):
640 run(args)
641 assert capsys.readouterr().out == ""
642
643 def test_already_exists_message_on_stderr(
644 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
645 ) -> None:
646 import argparse
647 from muse.cli.commands.clone import run
648 target = tmp_path / "existing"
649 muse_dir(target).mkdir(parents=True)
650 args = argparse.Namespace(
651 url="http://localhost:19999/repo",
652 directory=str(target),
653 branch=None,
654 dry_run=False,
655 no_checkout=False,
656 json_out=False,
657 )
658 with (
659 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
660 pytest.raises(SystemExit),
661 ):
662 run(args)
663 cap = capsys.readouterr()
664 assert "already" in cap.err.lower()
665 assert cap.out == ""
666
667
668 # ── E2E: via CliRunner ────────────────────────────────────────────────────────
669
670 def _invoke(*args: str, transport: MagicMock | None = None, tmp_path: pathlib.Path | None = None) -> InvokeResult:
671 t = transport or _make_transport_mock()
672 with (
673 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
674 patch("muse.cli.commands.clone.make_transport", return_value=t),
675 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
676 patch("muse.cli.commands.clone.set_remote"),
677 patch("muse.cli.commands.clone.set_remote_head"),
678 patch("muse.cli.commands.clone.set_upstream"),
679 patch("muse.cli.commands.clone.apply_manifest"),
680 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
681 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
682 ):
683 return runner.invoke(
684 cli,
685 ["clone", "http://localhost:19999/gabriel/repo", *args],
686 )
687
688
689 class TestCLIClone:
690 def test_dry_run_exits_zero_no_fs_changes(self, tmp_path: pathlib.Path) -> None:
691 target = tmp_path / "should-not-exist"
692 result = runner.invoke(
693 cli,
694 ["clone", "http://localhost:19999/repo", str(target), "--dry-run"],
695 )
696 assert not target.exists()
697
698 def test_dry_run_json_schema(self, tmp_path: pathlib.Path) -> None:
699 with (
700 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
701 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
702 ):
703 result = runner.invoke(
704 cli,
705 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--dry-run", "--json"],
706 )
707 assert result.exit_code == 0, result.output
708 data = _json_line(result)
709 assert data["status"] == "dry_run"
710 assert data["dry_run"] is True
711 assert data["head"] == COMMIT_ID
712 assert data["domain"] == "code"
713
714 def test_json_cloned_schema(self, tmp_path: pathlib.Path) -> None:
715 """All expected keys present; blobs_written counted via on_object callback."""
716 with (
717 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
718 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock(objects_count=11)),
719 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result(commits_written=5)),
720 patch("muse.cli.commands.clone.set_remote"),
721 patch("muse.cli.commands.clone.set_remote_head"),
722 patch("muse.cli.commands.clone.set_upstream"),
723 patch("muse.cli.commands.clone.apply_manifest"),
724 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
725 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
726 ):
727 result = runner.invoke(
728 cli,
729 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--json"],
730 )
731 assert result.exit_code == 0, result.output
732 data = _json_line(result)
733 for key in ("status", "url", "directory", "branch", "commits_received", "blobs_written", "head", "domain", "dry_run"):
734 assert key in data, f"Missing key: {key}"
735 assert data["status"] == "cloned"
736 assert data["dry_run"] is False
737 assert data["commits_received"] == 5
738 assert data["blobs_written"] == 11
739
740 def test_no_checkout_flag_accepted(self, tmp_path: pathlib.Path) -> None:
741 with (
742 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
743 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
744 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
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
796 Patches are applied once at the test level (not per-thread) to avoid
797 unittest.mock.patch thread-safety issues: concurrent patch/unpatch of the
798 same attribute causes stale mocks to leak into subsequent tests.
799 """
800 import argparse
801 from muse.cli.commands.clone import run
802 errors: list[str] = []
803
804 # Per-thread transport mocks are safe because they are independent objects.
805 transports = [_make_transport_mock() for _ in range(8)]
806
807 def _do(idx: int) -> None:
808 try:
809 target = tmp_path / f"repo{idx}"
810 args = argparse.Namespace(
811 url=f"http://localhost:19999/repo{idx}",
812 directory=str(target),
813 branch=None,
814 dry_run=False,
815 no_checkout=True,
816 json_out=False,
817 )
818 run(args)
819 assert muse_dir(target).is_dir()
820 assert (tags_dir(target)).is_dir()
821 meta = json.loads((repo_json_path(target)).read_text())
822 assert meta["domain"] == "code"
823 except Exception as exc:
824 errors.append(f"Thread {idx}: {exc}")
825
826 with (
827 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
828 patch("muse.cli.commands.clone.make_transport", side_effect=transports),
829 patch("muse.cli.commands.clone.apply_mpack", return_value=_make_apply_result()),
830 patch("muse.cli.commands.clone.set_remote"),
831 patch("muse.cli.commands.clone.set_remote_head"),
832 patch("muse.cli.commands.clone.set_upstream"),
833 ):
834 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
835 for th in threads:
836 th.start()
837 for th in threads:
838 th.join()
839 assert errors == [], f"Concurrent clone failures:\n{'\n'.join(errors)}"
840
841
842 # ---------------------------------------------------------------------------
843 # Flag registration tests
844 # ---------------------------------------------------------------------------
845
846 import argparse as _argparse
847 from muse.cli.commands.clone import register as _register_clone
848 from muse.core.paths import config_toml_path, head_path, heads_dir, muse_dir, ref_path, repo_json_path, tags_dir
849
850
851 def _parse_clone(*args: str) -> _argparse.Namespace:
852 root_p = _argparse.ArgumentParser()
853 subs = root_p.add_subparsers(dest="cmd")
854 _register_clone(subs)
855 return root_p.parse_args(["clone", *args])
856
857
858 class TestRegisterFlags:
859 def test_default_json_out_is_false(self) -> None:
860 ns = _parse_clone("https://example.com/repo")
861 assert ns.json_out is False
862
863 def test_json_flag_sets_json_out(self) -> None:
864 ns = _parse_clone("https://example.com/repo", "--json")
865 assert ns.json_out is True
866
867 def test_j_shorthand_sets_json_out(self) -> None:
868 ns = _parse_clone("https://example.com/repo", "-j")
869 assert ns.json_out is True
870
871 def test_dry_run_flag(self) -> None:
872 ns = _parse_clone("https://example.com/repo", "--dry-run")
873 assert ns.dry_run is True
874
875 def test_n_shorthand_for_dry_run(self) -> None:
876 ns = _parse_clone("https://example.com/repo", "-n")
877 assert ns.dry_run is True
878
879 def test_format_flag_no_longer_exists(self) -> None:
880 import pytest
881 with pytest.raises(SystemExit):
882 _parse_clone("https://example.com/repo", "--format", "json")
883
884
885 # ---------------------------------------------------------------------------
886 # muse#35 — clone must not falsely report success when a branch tip commit
887 # was silently dropped (e.g. by a snapshot hash mismatch in apply_mpack).
888 # ---------------------------------------------------------------------------
889
890 from muse.core.mpack import ApplyResult
891
892
893 class TestSnapshotHashMismatchGuard:
894 def test_branch_ref_not_written_when_commit_missing_from_store(
895 self, tmp_path: pathlib.Path
896 ) -> None:
897 """When apply_mpack reports commits_written below what the server
898 sent, and the branch tip commit is not actually readable from the
899 local store, clone must NOT write that branch's ref — writing it
900 anyway would point the ref at a nonexistent commit and make the
901 corruption permanent (the next pull would see the tracking ref
902 already matching the remote and skip re-fetching)."""
903 branches = {"main": COMMIT_ID}
904 t = _make_transport_mock(branch_heads=branches)
905 t.fetch_mpack.side_effect = lambda *a, **kw: _make_fetch_mpack_result(
906 commits=[COMMIT_ID], blobs_received=11,
907 )
908 ar = ApplyResult(
909 commits_written=0, snapshots_written=0, blobs_written=11,
910 blobs_skipped=0, tags_written=0, failed_blobs=[],
911 skipped_snapshots=[SNAP_ID],
912 )
913 import argparse
914 from muse.cli.commands.clone import run
915 target = tmp_path / "cloned"
916 args = argparse.Namespace(
917 url="http://localhost:19999/gabriel/repo",
918 directory=str(target), branch=None, dry_run=False,
919 no_checkout=True, json_out=True,
920 )
921 with (
922 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
923 patch("muse.cli.commands.clone.make_transport", return_value=t),
924 patch("muse.cli.commands.clone.apply_mpack", return_value=ar),
925 patch("muse.cli.commands.clone.set_remote"),
926 patch("muse.cli.commands.clone.set_remote_head") as mock_set_remote_head,
927 patch("muse.cli.commands.clone.set_upstream"),
928 patch("muse.cli.commands.clone.read_commit", return_value=None),
929 pytest.raises(SystemExit) as exc_info,
930 ):
931 run(args)
932
933 from muse.core.errors import ExitCode
934 assert exc_info.value.code == ExitCode.PARTIAL
935 # _init_dot_muse always pre-creates an empty placeholder ref for the
936 # default branch (matching `muse init`'s convention for a fresh,
937 # commit-less branch) — the guard's job is to leave it EMPTY rather
938 # than advance it to point at a commit that was never written.
939 ref_content = ref_path(target, "main").read_text()
940 assert ref_content != COMMIT_ID, (
941 "branch ref must not be advanced to a commit that is missing from the store"
942 )
943 assert ref_content == "", (
944 f"expected the untouched placeholder (empty) ref, got {ref_content!r}"
945 )
946 mock_set_remote_head.assert_not_called()
947
948 def test_json_reports_partial_status_not_cloned(
949 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
950 ) -> None:
951 branches = {"main": COMMIT_ID}
952 t = _make_transport_mock(branch_heads=branches)
953 t.fetch_mpack.side_effect = lambda *a, **kw: _make_fetch_mpack_result(
954 commits=[COMMIT_ID], blobs_received=11,
955 )
956 ar = ApplyResult(
957 commits_written=0, snapshots_written=0, blobs_written=11,
958 blobs_skipped=0, tags_written=0, failed_blobs=[],
959 skipped_snapshots=[SNAP_ID],
960 )
961 import argparse
962 from muse.cli.commands.clone import run
963 target = tmp_path / "cloned"
964 args = argparse.Namespace(
965 url="http://localhost:19999/gabriel/repo",
966 directory=str(target), branch=None, dry_run=False,
967 no_checkout=True, json_out=True,
968 )
969 with (
970 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
971 patch("muse.cli.commands.clone.make_transport", return_value=t),
972 patch("muse.cli.commands.clone.apply_mpack", return_value=ar),
973 patch("muse.cli.commands.clone.set_remote"),
974 patch("muse.cli.commands.clone.set_remote_head"),
975 patch("muse.cli.commands.clone.set_upstream"),
976 patch("muse.cli.commands.clone.read_commit", return_value=None),
977 pytest.raises(SystemExit),
978 ):
979 run(args)
980
981 out = capsys.readouterr().out
982 line = next(l for l in out.splitlines() if l.strip().startswith("{"))
983 data = json.loads(line)
984 assert data["status"] == "partial"
985 assert data["commits_received"] == 0
986 assert data["commits_missing"] >= 1
987 assert data["skipped_snapshots"] >= 1
988
989 def test_happy_path_still_reports_cloned_and_writes_ref(
990 self, tmp_path: pathlib.Path
991 ) -> None:
992 """Regression guard: when the commit WAS actually written, the
993 guard must not withhold the ref or report a false partial."""
994 _invoke_run(tmp_path, json_out=False)
995 target = tmp_path / "cloned"
996 assert ref_path(target, "main").exists()
File History 1 commit
sha256:a5f8a020b53ddab68f231cf27489fe865525f85ee924133034262d39b332076e fix(clone,pull): never claim success when a snapshot hash m… Sonnet 5 patch 1 day ago