gabriel / muse public
test_cmd_clone_hardening.py python
852 lines 36.2 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
1 """Comprehensive 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, no-path
8 - _init_muse_dir: all _CLONE_SUBDIRS created, repo.json written, HEAD set,
9 config.toml written, missing tags/ regression
10 - _restore_working_tree: commit None warns, snapshot None warns, happy path
11
12 Integration (mocked transport)
13 - run: already-exists guard, auth token forwarded, domain fallback "code",
14 branch fallback to first available, directory creation on error removed,
15 commits_received from apply_result not bundle, no-checkout skips manifest,
16 branch ref files written atomically, remote tracking pointer set
17
18 Security
19 - ANSI injection in URL stripped in stderr output
20 - ANSI injection in branch name stripped
21 - Path-traversal URL blocked by _infer_dir_name
22 - All progress/error messages go to stderr, not stdout
23 - target directory removed on fetch failure
24 - target directory removed on _init_muse_dir failure
25
26 E2E (via CliRunner)
27 - --dry-run exits 0, no filesystem changes
28 - --dry-run --json correct schema
29 - --json cloned schema correct
30 - --format json equivalent to --json
31 - --no-checkout skips working tree
32 - --branch selects branch
33 - already-exists exits non-zero
34 - transport error exits non-zero
35 - empty repository exits non-zero
36 - unknown --branch falls back with warning
37
38 Performance / Stress
39 - 8 concurrent clones into isolated directories do not interfere
40 """
41
42 from __future__ import annotations
43
44 import json
45 import pathlib
46 import threading
47 from typing import TYPE_CHECKING
48 from unittest.mock import MagicMock, call, patch
49
50 import pytest
51
52 from tests.cli_test_helper import CliRunner, InvokeResult
53 from muse.core._types import Manifest
54
55 if TYPE_CHECKING:
56 from muse.cli.commands.clone import _CloneJson
57 from muse.core.pack import ApplyResult, PackBundle, RemoteInfo
58 from muse.core.transport import SigningIdentity
59
60 cli = None
61 runner = CliRunner()
62
63 COMMIT_ID = "a" * 64
64 SNAP_ID = "b" * 64
65 REPO_ID = "test-repo-id"
66
67
68 # ── typed helpers ─────────────────────────────────────────────────────────────
69
70 def _make_bundle() -> "PackBundle":
71 from muse.core.pack import PackBundle
72 return PackBundle(commits=[], snapshots=[], objects=[])
73
74
75 def _make_apply_result(
76 commits_written: int = 5,
77 objects_written: int = 11,
78 ) -> "ApplyResult":
79 from muse.core.pack import ApplyResult
80 return ApplyResult(
81 commits_written=commits_written,
82 snapshots_written=commits_written,
83 objects_written=objects_written,
84 objects_skipped=0,
85 )
86
87
88 def _make_remote_info(
89 branch_heads: Manifest | None = None,
90 domain: str = "code",
91 default_branch: str = "main",
92 repo_id: str = REPO_ID,
93 ) -> RemoteInfo:
94 # Use `is not None` so that callers can explicitly pass {} for an empty repo.
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_transport_mock(
107 branch_heads: Manifest | None = None,
108 domain: str = "code",
109 ) -> MagicMock:
110 t = MagicMock()
111 t.fetch_remote_info.return_value = _make_remote_info(branch_heads, domain=domain)
112 t.fetch_pack.return_value = _make_bundle()
113 return t
114
115
116 def _json_line(result: InvokeResult) -> "_CloneJson":
117 """Extract the JSON object from cli_test_helper's combined stdout+stderr."""
118 for line in result.output.splitlines():
119 stripped = line.strip()
120 if stripped.startswith("{"):
121 parsed: _CloneJson = json.loads(stripped)
122 return parsed
123 raise ValueError(f"No JSON line in output:\n{result.output!r}")
124
125
126 # ── Unit: _infer_dir_name ─────────────────────────────────────────────────────
127
128 class TestInferDirName:
129 def test_last_url_segment(self) -> None:
130 from muse.cli.commands.clone import _infer_dir_name
131 assert _infer_dir_name("http://hub.muse.ai/gabriel/my-repo") == "my-repo"
132
133 def test_trailing_slash_stripped(self) -> None:
134 from muse.cli.commands.clone import _infer_dir_name
135 assert _infer_dir_name("http://hub.muse.ai/gabriel/my-repo/") == "my-repo"
136
137 def test_query_string_stripped(self) -> None:
138 from muse.cli.commands.clone import _infer_dir_name
139 assert _infer_dir_name("http://hub.muse.ai/repo?token=abc") == "repo"
140
141 def test_fragment_stripped(self) -> None:
142 from muse.cli.commands.clone import _infer_dir_name
143 assert _infer_dir_name("http://hub.muse.ai/repo#section") == "repo"
144
145 def test_bare_host_uses_hostname(self) -> None:
146 """A URL with no path segment uses the hostname as directory name."""
147 from muse.cli.commands.clone import _infer_dir_name
148 result = _infer_dir_name("http://hub.muse.ai/")
149 # hostname is a safe, non-empty, non-traversal name
150 assert result == "hub.muse.ai"
151 assert ".." not in result
152
153 def test_path_traversal_blocked(self) -> None:
154 """A crafted URL must not produce '..' as the directory name."""
155 from muse.cli.commands.clone import _infer_dir_name
156 result = _infer_dir_name("http://evil.example.com/../../../../etc/passwd")
157 assert ".." not in result
158 assert "/" not in result
159 assert result != ""
160
161 def test_dot_only_falls_back(self) -> None:
162 from muse.cli.commands.clone import _infer_dir_name
163 result = _infer_dir_name("http://example.com/.")
164 assert result not in (".", "..")
165
166 def test_double_dot_falls_back(self) -> None:
167 from muse.cli.commands.clone import _infer_dir_name
168 result = _infer_dir_name("http://example.com/..")
169 assert result not in (".", "..")
170 assert result == "muse-repo"
171
172
173 # ── Unit: _init_muse_dir ──────────────────────────────────────────────────────
174
175 class TestInitMuseDir:
176 def test_all_clone_subdirs_created(self, tmp_path: pathlib.Path) -> None:
177 from muse.cli.commands.clone import _CLONE_SUBDIRS, _init_muse_dir
178 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
179 muse_dir = tmp_path / ".muse"
180 for subdir in _CLONE_SUBDIRS:
181 assert (muse_dir / subdir).is_dir(), f"Missing subdir: {subdir}"
182
183 def test_tags_dir_created(self, tmp_path: pathlib.Path) -> None:
184 """Regression: old clone.py omitted 'tags/' from the directory set."""
185 from muse.cli.commands.clone import _init_muse_dir
186 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
187 assert (tmp_path / ".muse" / "tags").is_dir()
188
189 def test_repo_json_written(self, tmp_path: pathlib.Path) -> None:
190 from muse.cli.commands.clone import _init_muse_dir
191 _init_muse_dir(tmp_path, "my-repo-id", "code", "main")
192 meta = json.loads((tmp_path / ".muse" / "repo.json").read_text())
193 assert meta["repo_id"] == "my-repo-id"
194 assert meta["domain"] == "code"
195 assert "schema_version" in meta
196 assert "created_at" in meta
197
198 def test_head_file_written(self, tmp_path: pathlib.Path) -> None:
199 from muse.cli.commands.clone import _init_muse_dir
200 _init_muse_dir(tmp_path, REPO_ID, "code", "dev")
201 head = (tmp_path / ".muse" / "HEAD").read_text()
202 assert "dev" in head
203
204 def test_default_branch_ref_created(self, tmp_path: pathlib.Path) -> None:
205 from muse.cli.commands.clone import _init_muse_dir
206 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
207 assert (tmp_path / ".muse" / "refs" / "heads" / "main").exists()
208
209 def test_config_toml_written(self, tmp_path: pathlib.Path) -> None:
210 from muse.cli.commands.clone import _init_muse_dir
211 _init_muse_dir(tmp_path, REPO_ID, "code", "main")
212 config = (tmp_path / ".muse" / "config.toml").read_text()
213 assert "[remotes]" in config
214
215 def test_superset_of_init_subdirs(self, tmp_path: pathlib.Path) -> None:
216 """_CLONE_SUBDIRS must be a superset of muse init's _INIT_SUBDIRS."""
217 from muse.cli.commands.clone import _CLONE_SUBDIRS
218 from muse.cli.commands.init import _INIT_SUBDIRS
219 clone_set = set(_CLONE_SUBDIRS)
220 for subdir in _INIT_SUBDIRS:
221 assert subdir in clone_set, (
222 f"_INIT_SUBDIRS has '{subdir}' but _CLONE_SUBDIRS does not — "
223 "clone would produce an incomplete repository layout"
224 )
225
226
227 # ── Unit: _restore_working_tree ───────────────────────────────────────────────
228
229 class TestRestoreWorkingTree:
230 def test_warns_when_commit_not_found(
231 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
232 ) -> None:
233 from muse.cli.commands.clone import _restore_working_tree
234 with patch("muse.cli.commands.clone.read_commit", return_value=None):
235 with patch("muse.cli.commands.clone.apply_manifest") as am:
236 _restore_working_tree(tmp_path, "a" * 64)
237 am.assert_not_called()
238 assert "not found" in caplog.text.lower() or "not found" in "".join(
239 r.message for r in caplog.records
240 )
241
242 def test_warns_when_snapshot_not_found(
243 self, tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
244 ) -> None:
245 from muse.cli.commands.clone import _restore_working_tree
246 fake_commit = MagicMock()
247 fake_commit.snapshot_id = SNAP_ID
248 with (
249 patch("muse.cli.commands.clone.read_commit", return_value=fake_commit),
250 patch("muse.cli.commands.clone.read_snapshot", return_value=None),
251 ):
252 with patch("muse.cli.commands.clone.apply_manifest") as am:
253 _restore_working_tree(tmp_path, COMMIT_ID)
254 am.assert_not_called()
255
256 def test_happy_path_calls_apply_manifest(self, tmp_path: pathlib.Path) -> None:
257 from muse.cli.commands.clone import _restore_working_tree
258 fake_commit = MagicMock()
259 fake_commit.snapshot_id = SNAP_ID
260 fake_snap = MagicMock()
261 fake_snap.manifest = {"file.txt": "aabbcc"}
262 with (
263 patch("muse.cli.commands.clone.read_commit", return_value=fake_commit),
264 patch("muse.cli.commands.clone.read_snapshot", return_value=fake_snap),
265 ):
266 with patch("muse.cli.commands.clone.apply_manifest") as am:
267 _restore_working_tree(tmp_path, COMMIT_ID)
268 am.assert_called_once_with(tmp_path, {"file.txt": "aabbcc"})
269
270
271 # ── Integration: run() with mocked transport ─────────────────────────────────
272
273 def _base_patches(
274 transport: MagicMock | None = None,
275 apply_result: "ApplyResult | None" = None,
276 ) -> tuple[MagicMock, "ApplyResult"]:
277 """Return (transport_mock, apply_result_mock) ready for use in tests."""
278 t = transport or _make_transport_mock()
279 ar = apply_result or _make_apply_result()
280 return t, ar
281
282
283 class TestRunIntegration:
284 def _invoke_run(
285 self,
286 tmp_path: pathlib.Path,
287 url: str = "http://localhost:19999/gabriel/repo",
288 branch: str | None = None,
289 dry_run: bool = False,
290 no_checkout: bool = False,
291 fmt: str = "text",
292 transport: MagicMock | None = None,
293 apply_result: "ApplyResult | None" = None,
294 signing: "SigningIdentity | None" = None,
295 ) -> None:
296 """Invoke run() directly with all external dependencies mocked."""
297 import argparse
298 from muse.cli.commands.clone import run
299
300 t, ar = _base_patches(transport, apply_result)
301 args = argparse.Namespace(
302 url=url,
303 directory=str(tmp_path / "cloned"),
304 branch=branch,
305 dry_run=dry_run,
306 no_checkout=no_checkout,
307 format=fmt,
308 )
309 with (
310 patch("muse.cli.commands.clone.get_signing_identity", return_value=signing),
311 patch("muse.cli.commands.clone.make_transport", return_value=t),
312 patch("muse.cli.commands.clone.apply_pack", return_value=ar),
313 patch("muse.cli.commands.clone.set_remote"),
314 patch("muse.cli.commands.clone.set_remote_head"),
315 patch("muse.cli.commands.clone.set_upstream"),
316 patch("muse.cli.commands.clone.apply_manifest"),
317 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
318 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
319 ):
320 run(args)
321
322 def test_already_exists_raises_user_error(self, tmp_path: pathlib.Path) -> None:
323 import argparse
324 from muse.cli.commands.clone import run
325 from muse.core.errors import ExitCode
326 target = tmp_path / "existing"
327 (target / ".muse").mkdir(parents=True)
328 args = argparse.Namespace(
329 url="http://localhost:19999/repo",
330 directory=str(target),
331 branch=None,
332 dry_run=False,
333 no_checkout=False,
334 format="text",
335 )
336 with pytest.raises(SystemExit) as exc:
337 with patch("muse.cli.commands.clone.get_signing_identity", return_value=None):
338 run(args)
339 assert exc.value.code == ExitCode.USER_ERROR
340
341 def test_signing_forwarded_to_transport(self, tmp_path: pathlib.Path) -> None:
342 """Signing identity from get_signing_identity must reach both fetch_remote_info and fetch_pack."""
343 from muse.core.transport import SigningIdentity
344 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
345 signing = SigningIdentity("alice", Ed25519PrivateKey.generate())
346 t = _make_transport_mock()
347 self._invoke_run(tmp_path, transport=t, signing=signing)
348 t.fetch_remote_info.assert_called_once_with(
349 "http://localhost:19999/gabriel/repo", signing=signing
350 )
351 t.fetch_pack.assert_called_once()
352 _, kwargs = t.fetch_pack.call_args
353 assert kwargs.get("signing") is signing
354
355 def test_domain_defaults_to_code_not_midi(self, tmp_path: pathlib.Path) -> None:
356 """Regression: old code fell back to 'midi'; correct fallback is 'code'."""
357 t = _make_transport_mock()
358 t.fetch_remote_info.return_value = _make_remote_info(domain="")
359 target = tmp_path / "cloned"
360 import argparse
361 from muse.cli.commands.clone import run
362 args = argparse.Namespace(
363 url="http://localhost:19999/repo",
364 directory=str(target),
365 branch=None,
366 dry_run=False,
367 no_checkout=False,
368 format="text",
369 )
370 with (
371 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
372 patch("muse.cli.commands.clone.make_transport", return_value=t),
373 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
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 """Regression: commits_received must use 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()
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 format="json",
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_pack", return_value=ar),
406 patch("muse.cli.commands.clone.set_remote"),
407 patch("muse.cli.commands.clone.set_remote_head"),
408 patch("muse.cli.commands.clone.set_upstream"),
409 patch("muse.cli.commands.clone.apply_manifest"),
410 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
411 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
412 patch("builtins.print", side_effect=lambda *a, **kw: output_lines.append(str(a[0]) if a else "")),
413 ):
414 run(args)
415 json_out = next((l for l in output_lines if l.strip().startswith("{")), None)
416 assert json_out is not None
417 data: _CloneJson = json.loads(json_out)
418 assert data["commits_received"] == 7
419 assert data["objects_written"] == 23
420
421 def test_branch_fallback_to_first_available(
422 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
423 ) -> None:
424 """When requested branch is missing, clone falls back to first branch."""
425 import argparse
426 from muse.cli.commands.clone import run
427 t = _make_transport_mock(branch_heads={"dev": COMMIT_ID})
428 args = argparse.Namespace(
429 url="http://localhost:19999/repo",
430 directory=str(tmp_path / "cloned"),
431 branch="nonexistent",
432 dry_run=False,
433 no_checkout=True,
434 format="text",
435 )
436 with (
437 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
438 patch("muse.cli.commands.clone.make_transport", return_value=t),
439 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
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_pack fails after target is created, directory is cleaned up."""
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_pack.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 format="text",
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 """If _init_muse_dir fails, the target directory is cleaned up."""
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 format="text",
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 """--no-checkout must not touch the working tree."""
498 t = _make_transport_mock()
499 apply_mock = MagicMock()
500 import argparse
501 from muse.cli.commands.clone import run
502 args = argparse.Namespace(
503 url="http://localhost:19999/repo",
504 directory=str(tmp_path / "cloned"),
505 branch=None,
506 dry_run=False,
507 no_checkout=True,
508 format="text",
509 )
510 with (
511 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
512 patch("muse.cli.commands.clone.make_transport", return_value=t),
513 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
514 patch("muse.cli.commands.clone.set_remote"),
515 patch("muse.cli.commands.clone.set_remote_head"),
516 patch("muse.cli.commands.clone.set_upstream"),
517 patch("muse.cli.commands.clone.apply_manifest", apply_mock),
518 ):
519 run(args)
520 apply_mock.assert_not_called()
521
522 def test_branch_refs_written_for_every_remote_branch(
523 self, tmp_path: pathlib.Path
524 ) -> None:
525 """Every branch in branch_heads must get a local ref file."""
526 branches = {"main": "a" * 64, "dev": "b" * 64, "feat/x": "c" * 64}
527 t = _make_transport_mock(branch_heads=branches)
528 import argparse
529 from muse.cli.commands.clone import run
530 target = tmp_path / "cloned"
531 args = argparse.Namespace(
532 url="http://localhost:19999/repo",
533 directory=str(target),
534 branch=None,
535 dry_run=False,
536 no_checkout=True,
537 format="text",
538 )
539 with (
540 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
541 patch("muse.cli.commands.clone.make_transport", return_value=t),
542 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
543 patch("muse.cli.commands.clone.set_remote"),
544 patch("muse.cli.commands.clone.set_remote_head"),
545 patch("muse.cli.commands.clone.set_upstream"),
546 ):
547 run(args)
548 for branch, cid in branches.items():
549 ref_file = target / ".muse" / "refs" / "heads" / branch
550 assert ref_file.exists(), f"Missing ref file for branch {branch}"
551 assert ref_file.read_text() == cid
552
553
554 # ── Security ──────────────────────────────────────────────────────────────────
555
556 class TestSecurity:
557 def test_ansi_in_url_stripped_in_stderr(
558 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
559 ) -> None:
560 evil_url = "\x1b[31mhttp://evil.com/repo\x1b[0m"
561 import argparse
562 from muse.cli.commands.clone import run
563 from muse.core.transport import TransportError
564 t = MagicMock()
565 t.fetch_remote_info.side_effect = TransportError("refused", 503)
566 args = argparse.Namespace(
567 url=evil_url,
568 directory=str(tmp_path / "cloned"),
569 branch=None,
570 dry_run=False,
571 no_checkout=False,
572 format="text",
573 )
574 with (
575 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
576 patch("muse.cli.commands.clone.make_transport", return_value=t),
577 pytest.raises(SystemExit),
578 ):
579 run(args)
580 assert "\x1b[" not in capsys.readouterr().err
581
582 def test_ansi_in_branch_name_stripped(
583 self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]
584 ) -> None:
585 evil_branch = "\x1b[32mHACKED\x1b[0m"
586 t = _make_transport_mock(branch_heads={"main": COMMIT_ID})
587 import argparse
588 from muse.cli.commands.clone import run
589 args = argparse.Namespace(
590 url="http://localhost:19999/repo",
591 directory=str(tmp_path / "cloned"),
592 branch=evil_branch,
593 dry_run=False,
594 no_checkout=True,
595 format="text",
596 )
597 with (
598 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
599 patch("muse.cli.commands.clone.make_transport", return_value=t),
600 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
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 """In text mode, stdout must be completely empty."""
618 t = _make_transport_mock()
619 import argparse
620 from muse.cli.commands.clone import run
621 args = argparse.Namespace(
622 url="http://localhost:19999/repo",
623 directory=str(tmp_path / "cloned"),
624 branch=None,
625 dry_run=False,
626 no_checkout=True,
627 format="text",
628 )
629 with (
630 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
631 patch("muse.cli.commands.clone.make_transport", return_value=t),
632 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
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 format="text",
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 extra: list[str] = []
670 if tmp_path is not None:
671 extra = ["--", str(tmp_path / "cloned")]
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_pack", 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 # Even if transport fails (no server), dry-run may exit with an error,
697 # but the directory must not have been created.
698 assert not target.exists()
699
700 def test_dry_run_json_schema(self, tmp_path: pathlib.Path) -> None:
701 with (
702 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
703 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
704 ):
705 result = runner.invoke(
706 cli,
707 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--dry-run", "--json"],
708 )
709 assert result.exit_code == 0, result.output
710 data = _json_line(result)
711 assert data["status"] == "dry_run"
712 assert data["dry_run"] is True
713 assert data["head"] == COMMIT_ID
714 assert data["domain"] == "code"
715
716 def test_json_cloned_schema(self, tmp_path: pathlib.Path) -> None:
717 with (
718 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
719 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
720 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
721 patch("muse.cli.commands.clone.set_remote"),
722 patch("muse.cli.commands.clone.set_remote_head"),
723 patch("muse.cli.commands.clone.set_upstream"),
724 patch("muse.cli.commands.clone.apply_manifest"),
725 patch("muse.cli.commands.clone.read_commit", return_value=MagicMock(snapshot_id=SNAP_ID)),
726 patch("muse.cli.commands.clone.read_snapshot", return_value=MagicMock(manifest={})),
727 ):
728 result = runner.invoke(
729 cli,
730 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--json"],
731 )
732 assert result.exit_code == 0, result.output
733 data = _json_line(result)
734 for key in ("status", "url", "directory", "branch", "commits_received", "objects_written", "head", "domain", "dry_run"):
735 assert key in data, f"Missing key: {key}"
736 assert data["status"] == "cloned"
737 assert data["dry_run"] is False
738 assert data["commits_received"] == 5
739 assert data["objects_written"] == 11
740
741 def test_format_json_equivalent_to_json_flag(self, tmp_path: pathlib.Path) -> None:
742 kwargs = dict(
743 patches=[
744 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
745 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
746 ]
747 )
748
749 def _run(flag: str) -> _CloneJson:
750 with (
751 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
752 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
753 ):
754 r = runner.invoke(cli, ["clone", "http://localhost:19999/repo", str(tmp_path / flag), "--dry-run", flag])
755 assert r.exit_code == 0
756 return _json_line(r)
757
758 assert _run("--json")["status"] == _run("--format=json")["status"]
759
760 def test_no_checkout_flag_accepted(self, tmp_path: pathlib.Path) -> None:
761 with (
762 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
763 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
764 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
765 patch("muse.cli.commands.clone.set_remote"),
766 patch("muse.cli.commands.clone.set_remote_head"),
767 patch("muse.cli.commands.clone.set_upstream"),
768 ):
769 result = runner.invoke(
770 cli,
771 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--no-checkout"],
772 )
773 assert result.exit_code == 0, result.output
774
775 def test_transport_error_exits_nonzero(self) -> None:
776 from muse.core.transport import TransportError
777 t = MagicMock()
778 t.fetch_remote_info.side_effect = TransportError("refused", 503)
779 with (
780 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
781 patch("muse.cli.commands.clone.make_transport", return_value=t),
782 ):
783 result = runner.invoke(cli, ["clone", "http://localhost:19999/repo", "/tmp/muse-test-clone-xxx"])
784 assert result.exit_code != 0
785
786 def test_empty_repository_exits_nonzero(self) -> None:
787 t = MagicMock()
788 t.fetch_remote_info.return_value = _make_remote_info(branch_heads={})
789 with (
790 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
791 patch("muse.cli.commands.clone.make_transport", return_value=t),
792 ):
793 result = runner.invoke(cli, ["clone", "http://localhost:19999/repo", "/tmp/muse-test-clone-empty"])
794 assert result.exit_code != 0
795
796 def test_json_on_stdout_parseable(self, tmp_path: pathlib.Path) -> None:
797 with (
798 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
799 patch("muse.cli.commands.clone.make_transport", return_value=_make_transport_mock()),
800 ):
801 result = runner.invoke(
802 cli,
803 ["clone", "http://localhost:19999/repo", str(tmp_path / "out"), "--dry-run", "--json"],
804 )
805 assert result.exit_code == 0
806 data = _json_line(result)
807 assert "status" in data
808
809
810 # ── Stress: concurrent clones into isolated directories ───────────────────────
811
812 class TestStressConcurrent:
813 def test_8_concurrent_clones_isolated(self, tmp_path: pathlib.Path) -> None:
814 """8 concurrent clone calls into separate directories must not interfere."""
815 import argparse
816 from muse.cli.commands.clone import run
817 errors: list[str] = []
818
819 def _do(idx: int) -> None:
820 try:
821 target = tmp_path / f"repo{idx}"
822 t = _make_transport_mock()
823 args = argparse.Namespace(
824 url=f"http://localhost:19999/repo{idx}",
825 directory=str(target),
826 branch=None,
827 dry_run=False,
828 no_checkout=True,
829 format="text",
830 )
831 with (
832 patch("muse.cli.commands.clone.get_signing_identity", return_value=None),
833 patch("muse.cli.commands.clone.make_transport", return_value=t),
834 patch("muse.cli.commands.clone.apply_pack", return_value=_make_apply_result()),
835 patch("muse.cli.commands.clone.set_remote"),
836 patch("muse.cli.commands.clone.set_remote_head"),
837 patch("muse.cli.commands.clone.set_upstream"),
838 ):
839 run(args)
840 assert (target / ".muse").is_dir()
841 assert (target / ".muse" / "tags").is_dir()
842 meta = json.loads((target / ".muse" / "repo.json").read_text())
843 assert meta["domain"] == "code"
844 except Exception as exc:
845 errors.append(f"Thread {idx}: {exc}")
846
847 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
848 for th in threads:
849 th.start()
850 for th in threads:
851 th.join()
852 assert errors == [], f"Concurrent clone failures:\n" + "\n".join(errors)
File History 5 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago
sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago
sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago