gabriel / muse public
test_cmd_remote_hardening.py python
2,202 lines 104.5 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 remote``.
2
3 Coverage
4 --------
5 Unit
6 - _validate_remote_name: valid names, empty, spaces, slashes, control chars
7 - _validate_url_scheme: http/https allowed, file/ftp/data rejected
8 - _collect_tracked_refs / _walk_refs: flat, nested, symlink skip, missing dir
9 - _ping_url: scheme guard fires before network, timeout, HTTP error, URLError
10
11 Integration (real repo via fixture)
12 - run_add: invalid name blocked, invalid scheme blocked, duplicate, --json schema
13 - run_remove: missing remote, --json schema, tracking refs cleaned
14 - run_rename: invalid new_name, missing old, duplicate new, --json schema
15 - run_get_url: missing remote, --json schema, bare URL on stdout
16 - run_set_url: invalid scheme, missing remote, --json schema
17 - run (list): --json schema, empty repo, verbose
18
19 Security
20 - ANSI injection in remote name stripped in stderr
21 - ANSI injection in URL stripped in stderr
22 - Invalid URL schemes rejected in add and set-url
23 - Symlink inside remotes dir skipped in collect_tracked_refs
24 - All diagnostic messages go to stderr; stdout clean in text mode
25
26 E2E (via CliRunner with real repo fixture)
27 - Every subcommand: --json flag produces parseable JSON on stdout
28 - Diagnostic errors confirmed on stderr (via result.output / stderr)
29 - get-url prints bare URL on stdout in text mode
30
31 Stress
32 - 8 concurrent remote adds to isolated repos do not interfere
33 """
34
35 from __future__ import annotations
36
37 import json
38 import pathlib
39 import threading
40 from typing import TYPE_CHECKING
41 from unittest.mock import MagicMock, patch
42
43 import pytest
44
45 from tests.cli_test_helper import CliRunner, InvokeResult
46
47 from muse.cli.commands.remote import (
48 _RemoteGetUrlJson,
49 _RemoteListJson,
50 _RemoteMutationJson,
51 _RemoteStatusJson,
52 )
53
54 from muse._version import __version__
55 from muse.cli.config import get_remote, list_remotes
56
57 if TYPE_CHECKING:
58 pass # kept for future conditional imports
59
60 cli = None
61 runner = CliRunner()
62
63
64 # ── JSON helpers — one per output schema ─────────────────────────────────────
65
66 def _json_mutation(result: InvokeResult) -> _RemoteMutationJson:
67 """Extract and parse a _RemoteMutationJson from the first JSON line in output."""
68 for line in result.output.splitlines():
69 stripped = line.strip()
70 if stripped.startswith("{"):
71 data: _RemoteMutationJson = json.loads(stripped)
72 return data
73 raise ValueError(f"No JSON line in output:\n{result.output!r}")
74
75
76 def _json_list(result: InvokeResult) -> _RemoteListJson:
77 """Extract and parse a _RemoteListJson from the first JSON line in output."""
78 for line in result.output.splitlines():
79 stripped = line.strip()
80 if stripped.startswith("{"):
81 data: _RemoteListJson = json.loads(stripped)
82 return data
83 raise ValueError(f"No JSON line in output:\n{result.output!r}")
84
85
86 def _json_get_url(result: InvokeResult) -> _RemoteGetUrlJson:
87 """Extract and parse a _RemoteGetUrlJson from the first JSON line in output."""
88 for line in result.output.splitlines():
89 stripped = line.strip()
90 if stripped.startswith("{"):
91 data: _RemoteGetUrlJson = json.loads(stripped)
92 return data
93 raise ValueError(f"No JSON line in output:\n{result.output!r}")
94
95
96 def _json_status(result: InvokeResult) -> _RemoteStatusJson:
97 """Extract and parse a _RemoteStatusJson from the first JSON line in output."""
98 for line in result.output.splitlines():
99 stripped = line.strip()
100 if stripped.startswith("{"):
101 data: _RemoteStatusJson = json.loads(stripped)
102 return data
103 raise ValueError(f"No JSON line in output:\n{result.output!r}")
104
105
106 # ── fixture ───────────────────────────────────────────────────────────────────
107
108 @pytest.fixture
109 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
110 """Minimal .muse/ repo with MUSE_REPO_ROOT set."""
111 muse_dir = tmp_path / ".muse"
112 for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"):
113 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
114 (muse_dir / "repo.json").write_text(
115 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
116 )
117 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
118 (muse_dir / "refs" / "heads" / "main").write_text("")
119 (muse_dir / "config.toml").write_text("")
120 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
121 monkeypatch.chdir(tmp_path)
122 return tmp_path
123
124
125 # ── Unit: _validate_remote_name ───────────────────────────────────────────────
126
127 class TestValidateRemoteName:
128 def test_simple_name_valid(self) -> None:
129 from muse.cli.commands.remote import _validate_remote_name
130 assert _validate_remote_name("origin") is None
131
132 def test_dash_underscore_dot_valid(self) -> None:
133 from muse.cli.commands.remote import _validate_remote_name
134 assert _validate_remote_name("my-remote_1.0") is None
135
136 def test_empty_name_invalid(self) -> None:
137 from muse.cli.commands.remote import _validate_remote_name
138 assert _validate_remote_name("") is not None
139
140 def test_space_in_name_invalid(self) -> None:
141 from muse.cli.commands.remote import _validate_remote_name
142 assert _validate_remote_name("my remote") is not None
143
144 def test_slash_in_name_invalid(self) -> None:
145 from muse.cli.commands.remote import _validate_remote_name
146 assert _validate_remote_name("org/remote") is not None
147
148 def test_ansi_escape_invalid(self) -> None:
149 from muse.cli.commands.remote import _validate_remote_name
150 assert _validate_remote_name("\x1b[31mevil\x1b[0m") is not None
151
152 def test_null_byte_invalid(self) -> None:
153 from muse.cli.commands.remote import _validate_remote_name
154 assert _validate_remote_name("evil\x00name") is not None
155
156 def test_alphanumeric_valid(self) -> None:
157 from muse.cli.commands.remote import _validate_remote_name
158 assert _validate_remote_name("Remote123") is None
159
160
161 # ── Unit: _validate_url_scheme ────────────────────────────────────────────────
162
163 class TestValidateUrlScheme:
164 def test_https_allowed(self) -> None:
165 from muse.cli.commands.remote import _validate_url_scheme
166 assert _validate_url_scheme("https://hub.muse.ai/org/repo") is None
167
168 def test_http_allowed(self) -> None:
169 from muse.cli.commands.remote import _validate_url_scheme
170 assert _validate_url_scheme("http://localhost:10003/org/repo") is None
171
172 def test_file_scheme_rejected(self) -> None:
173 from muse.cli.commands.remote import _validate_url_scheme
174 assert _validate_url_scheme("file:///etc/passwd") is not None
175
176 def test_ftp_scheme_rejected(self) -> None:
177 from muse.cli.commands.remote import _validate_url_scheme
178 assert _validate_url_scheme("ftp://ftp.example.com/repo") is not None
179
180 def test_data_scheme_rejected(self) -> None:
181 from muse.cli.commands.remote import _validate_url_scheme
182 assert _validate_url_scheme("data:text/plain,evil") is not None
183
184 def test_empty_scheme_rejected(self) -> None:
185 from muse.cli.commands.remote import _validate_url_scheme
186 assert _validate_url_scheme("not-a-url") is not None
187
188
189 # ── Unit: _collect_tracked_refs / _walk_refs ──────────────────────────────────
190
191 class TestCollectTrackedRefs:
192 def test_missing_dir_returns_empty(self, tmp_path: pathlib.Path) -> None:
193 from muse.cli.commands.remote import _collect_tracked_refs
194 assert _collect_tracked_refs(tmp_path / "nonexistent") == {}
195
196 def test_flat_branch_collected(self, tmp_path: pathlib.Path) -> None:
197 from muse.cli.commands.remote import _collect_tracked_refs
198 refs = tmp_path / "remotes" / "origin"
199 refs.mkdir(parents=True)
200 (refs / "main").write_text("a" * 64)
201 result = _collect_tracked_refs(refs)
202 assert "main" in result
203 assert result["main"] == "a" * 8
204
205 def test_nested_branch_collected(self, tmp_path: pathlib.Path) -> None:
206 """feat/ui must appear as 'feat/ui', not just 'ui'."""
207 from muse.cli.commands.remote import _collect_tracked_refs
208 refs = tmp_path / "remotes" / "origin"
209 (refs / "feat").mkdir(parents=True)
210 (refs / "feat" / "ui").write_text("b" * 64)
211 result = _collect_tracked_refs(refs)
212 assert "feat/ui" in result
213 assert result["feat/ui"] == "b" * 8
214
215 def test_symlink_skipped(self, tmp_path: pathlib.Path) -> None:
216 """Symlinks inside the remotes dir must be silently skipped."""
217 from muse.cli.commands.remote import _collect_tracked_refs
218 refs = tmp_path / "remotes" / "origin"
219 refs.mkdir(parents=True)
220 target = tmp_path / "secret.txt"
221 target.write_text("sensitive")
222 (refs / "evil-link").symlink_to(target)
223 result = _collect_tracked_refs(refs)
224 assert "evil-link" not in result
225
226 def test_empty_sha_shown_as_empty_marker(self, tmp_path: pathlib.Path) -> None:
227 from muse.cli.commands.remote import _collect_tracked_refs
228 refs = tmp_path / "remotes" / "origin"
229 refs.mkdir(parents=True)
230 (refs / "main").write_text("")
231 result = _collect_tracked_refs(refs)
232 assert result["main"] == "(empty)"
233
234 def test_multiple_branches_all_collected(self, tmp_path: pathlib.Path) -> None:
235 from muse.cli.commands.remote import _collect_tracked_refs
236 refs = tmp_path / "remotes" / "origin"
237 refs.mkdir(parents=True)
238 branches = {"main": "a" * 64, "dev": "b" * 64}
239 (refs / "feat").mkdir()
240 (refs / "feat" / "new").write_text("c" * 64)
241 for name, sha in branches.items():
242 (refs / name).write_text(sha)
243 result = _collect_tracked_refs(refs)
244 assert set(result) == {"main", "dev", "feat/new"}
245
246
247 # ── Unit: _ping_url ───────────────────────────────────────────────────────────
248
249 class TestPingUrl:
250 def test_non_http_scheme_blocked_before_network(self) -> None:
251 """file:// must be rejected without making a network request."""
252 from muse.cli.commands.remote import _ping_url
253 reachable, code, msg = _ping_url("file:///etc/passwd", timeout=5.0)
254 assert not reachable
255 assert code is None
256 assert "scheme" in msg.lower()
257
258 def test_ftp_scheme_blocked(self) -> None:
259 from muse.cli.commands.remote import _ping_url
260 reachable, _, msg = _ping_url("ftp://example.com", timeout=5.0)
261 assert not reachable
262 assert "scheme" in msg.lower()
263
264 def test_timeout_error_handled(self) -> None:
265 from muse.cli.commands.remote import _ping_url
266 with patch("urllib.request.urlopen", side_effect=TimeoutError()):
267 reachable, code, msg = _ping_url("http://timeout.example.com", timeout=0.001)
268 assert not reachable
269 assert "timed out" in msg
270
271 def test_http_error_returns_status_code(self) -> None:
272 import urllib.error
273 from muse.cli.commands.remote import _ping_url
274 exc = urllib.error.HTTPError(url="", code=503, msg="Service Unavailable", hdrs=MagicMock(), fp=None)
275 with patch("urllib.request.urlopen", side_effect=exc):
276 reachable, code, msg = _ping_url("http://example.com", timeout=5.0)
277 assert not reachable
278 assert code == 503
279
280 def test_url_error_handled(self) -> None:
281 import urllib.error
282 from muse.cli.commands.remote import _ping_url
283 exc = urllib.error.URLError(reason="connection refused")
284 with patch("urllib.request.urlopen", side_effect=exc):
285 reachable, code, msg = _ping_url("http://example.com", timeout=5.0)
286 assert not reachable
287 assert code is None
288
289 def test_successful_ping_returns_true(self) -> None:
290 from muse.cli.commands.remote import _ping_url
291 mock_resp = MagicMock()
292 mock_resp.__enter__ = lambda s: s
293 mock_resp.__exit__ = MagicMock(return_value=False)
294 mock_resp.status = 200
295 with patch("urllib.request.urlopen", return_value=mock_resp):
296 reachable, code, msg = _ping_url("http://hub.muse.ai", timeout=5.0)
297 assert reachable
298 assert code == 200
299
300
301 # ── Integration: subcommands with real repo ───────────────────────────────────
302
303 class TestRemoteAddHardening:
304 def test_invalid_name_rejected(self, repo: pathlib.Path) -> None:
305 result = runner.invoke(cli, ["remote", "add", "my remote", "https://hub.muse.io/r"])
306 assert result.exit_code != 0
307
308 def test_slash_in_name_rejected(self, repo: pathlib.Path) -> None:
309 result = runner.invoke(cli, ["remote", "add", "org/remote", "https://hub.muse.io/r"])
310 assert result.exit_code != 0
311
312 def test_file_scheme_rejected(self, repo: pathlib.Path) -> None:
313 result = runner.invoke(cli, ["remote", "add", "origin", "file:///etc/passwd"])
314 assert result.exit_code != 0
315
316 def test_ftp_scheme_rejected(self, repo: pathlib.Path) -> None:
317 result = runner.invoke(cli, ["remote", "add", "origin", "ftp://example.com/r"])
318 assert result.exit_code != 0
319
320 def test_add_json_schema(self, repo: pathlib.Path) -> None:
321 result = runner.invoke(
322 cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"]
323 )
324 assert result.exit_code == 0
325 data = _json_mutation(result)
326 assert data["status"] == "ok"
327 assert data["name"] == "origin"
328 assert data["url"] == "https://hub.muse.io/r"
329 assert data["old_name"] is None
330 assert data["new_name"] is None
331
332 def test_add_json_stdout_clean_of_diagnostics(self, repo: pathlib.Path) -> None:
333 """In JSON mode stdout must contain only the JSON object."""
334 result = runner.invoke(
335 cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"]
336 )
337 for line in result.output.splitlines():
338 stripped = line.strip()
339 if stripped:
340 assert stripped.startswith("{"), f"Non-JSON line on stdout: {stripped!r}"
341
342 def test_duplicate_add_error_on_stderr(self, repo: pathlib.Path) -> None:
343 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
344 result = runner.invoke(cli, ["remote", "add", "origin", "https://other.com/r"])
345 assert result.exit_code != 0
346 assert "already exists" in (result.stderr or result.output).lower()
347
348
349 class TestRemoteRemoveHardening:
350 def test_remove_json_schema(self, repo: pathlib.Path) -> None:
351 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
352 result = runner.invoke(cli, ["remote", "remove", "origin", "--json"])
353 assert result.exit_code == 0
354 data = _json_mutation(result)
355 assert data["status"] == "ok"
356 assert data["name"] == "origin"
357 # url now holds the removed URL so agents can confirm / undo
358 assert data["url"] == "https://hub.muse.io/r"
359
360 def test_remove_missing_error_on_stderr(self, repo: pathlib.Path) -> None:
361 result = runner.invoke(cli, ["remote", "remove", "ghost"])
362 assert result.exit_code != 0
363 assert "does not exist" in (result.stderr or result.output).lower()
364
365 def test_remove_cleans_nested_tracking_refs(self, repo: pathlib.Path) -> None:
366 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
367 refs_dir = repo / ".muse" / "remotes" / "origin"
368 (refs_dir / "feat").mkdir(parents=True, exist_ok=True)
369 (refs_dir / "main").write_text("a" * 64)
370 (refs_dir / "feat" / "ui").write_text("b" * 64)
371 runner.invoke(cli, ["remote", "remove", "origin"])
372 assert not refs_dir.exists()
373
374
375 class TestRemoteRenameHardening:
376 def test_invalid_new_name_rejected(self, repo: pathlib.Path) -> None:
377 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
378 result = runner.invoke(cli, ["remote", "rename", "origin", "bad name"])
379 assert result.exit_code != 0
380
381 def test_rename_json_schema(self, repo: pathlib.Path) -> None:
382 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
383 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"])
384 assert result.exit_code == 0
385 data = _json_mutation(result)
386 assert data["status"] == "ok"
387 assert data["old_name"] == "origin"
388 assert data["new_name"] == "upstream"
389 assert data["name"] == "upstream"
390
391 def test_rename_ansi_in_old_name_sanitized(
392 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
393 ) -> None:
394 evil = "\x1b[31mghost\x1b[0m"
395 result = runner.invoke(cli, ["remote", "rename", evil, "safe"])
396 # May fail validation or "does not exist" — either way no ANSI in stderr
397 assert result.exit_code != 0
398 err = result.stderr or ""
399 assert "\x1b[" not in err
400
401 def test_rename_ansi_in_new_name_rejected(self, repo: pathlib.Path) -> None:
402 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
403 evil = "\x1b[31mevil\x1b[0m"
404 result = runner.invoke(cli, ["remote", "rename", "origin", evil])
405 assert result.exit_code != 0
406
407
408 class TestRemoteGetUrlHardening:
409 def test_get_url_json_schema(self, repo: pathlib.Path) -> None:
410 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
411 result = runner.invoke(cli, ["remote", "get-url", "origin", "--json"])
412 assert result.exit_code == 0
413 data = _json_get_url(result)
414 assert data["name"] == "origin"
415 assert data["url"] == "https://hub.muse.io/r"
416
417 def test_get_url_bare_on_stdout_text_mode(self, repo: pathlib.Path) -> None:
418 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
419 result = runner.invoke(cli, ["remote", "get-url", "origin"])
420 assert result.exit_code == 0
421 assert "https://hub.muse.io/r" in result.output
422
423 def test_get_url_missing_exits_nonzero(self, repo: pathlib.Path) -> None:
424 result = runner.invoke(cli, ["remote", "get-url", "ghost"])
425 assert result.exit_code != 0
426
427
428 class TestRemoteSetUrlHardening:
429 def test_set_url_file_scheme_rejected(self, repo: pathlib.Path) -> None:
430 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
431 result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/passwd"])
432 assert result.exit_code != 0
433 assert get_remote("origin", repo) == "https://hub.muse.io/r"
434
435 def test_set_url_json_schema(self, repo: pathlib.Path) -> None:
436 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
437 result = runner.invoke(
438 cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"]
439 )
440 assert result.exit_code == 0
441 data = _json_mutation(result)
442 assert data["status"] == "ok"
443 assert data["name"] == "origin"
444 assert data["url"] == "https://hub.muse.io/r2"
445
446 def test_set_url_hint_sanitized(
447 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
448 ) -> None:
449 evil = "\x1b[31mghost\x1b[0m"
450 result = runner.invoke(cli, ["remote", "set-url", evil, "https://example.com/r"])
451 assert result.exit_code != 0
452 err = result.stderr or ""
453 assert "\x1b[" not in err
454
455
456 class TestRemoteListHardening:
457 def test_list_json_schema_empty(self, repo: pathlib.Path) -> None:
458 result = runner.invoke(cli, ["remote", "--json"])
459 assert result.exit_code == 0
460 data = _json_list(result)
461 assert data["remotes"] == []
462
463 def test_list_json_schema_with_remotes(self, repo: pathlib.Path) -> None:
464 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r1"])
465 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/r2"])
466 result = runner.invoke(cli, ["remote", "--json"])
467 assert result.exit_code == 0
468 data = _json_list(result)
469 names = {r["name"] for r in data["remotes"]}
470 assert names == {"origin", "upstream"}
471 for entry in data["remotes"]:
472 for key in ("name", "url", "tracking", "head"):
473 assert key in entry, f"Missing key '{key}' in remote entry"
474
475 def test_list_json_stdout_only(self, repo: pathlib.Path) -> None:
476 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
477 result = runner.invoke(cli, ["remote", "--json"])
478 assert result.exit_code == 0
479 data = _json_list(result)
480 assert isinstance(data["remotes"], list)
481
482 def test_list_empty_no_crash(self, repo: pathlib.Path) -> None:
483 result = runner.invoke(cli, ["remote"])
484 assert result.exit_code == 0
485
486
487 # ── Security ──────────────────────────────────────────────────────────────────
488
489 class TestSecurity:
490 def test_ansi_in_name_sanitized_in_stderr_add(
491 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
492 ) -> None:
493 evil = "\x1b[31morigin\x1b[0m"
494 result = runner.invoke(cli, ["remote", "add", evil, "https://hub.muse.io/r"])
495 assert result.exit_code != 0
496 err = result.stderr or ""
497 assert "\x1b[" not in err
498
499 def test_ansi_in_url_sanitized_in_stderr_add(
500 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
501 ) -> None:
502 evil_url = "https://hub.muse.io/\x1b[31mevil\x1b[0m"
503 # URL contains ANSI — scheme is valid but the output must be clean
504 result = runner.invoke(cli, ["remote", "add", "origin", evil_url])
505 # Regardless of outcome, stderr must not contain raw ANSI
506 err = result.stderr or ""
507 assert "\x1b[" not in err
508
509 def test_file_scheme_blocked_in_add(self, repo: pathlib.Path) -> None:
510 result = runner.invoke(cli, ["remote", "add", "origin", "file:///sensitive"])
511 assert result.exit_code != 0
512
513 def test_file_scheme_blocked_in_set_url(self, repo: pathlib.Path) -> None:
514 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
515 result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/shadow"])
516 assert result.exit_code != 0
517 assert get_remote("origin", repo) == "https://hub.muse.io/r"
518
519 def test_file_scheme_blocked_in_ping(self) -> None:
520 from muse.cli.commands.remote import _ping_url
521 reachable, _, msg = _ping_url("file:///etc/passwd", 1.0)
522 assert not reachable
523 assert "scheme" in msg.lower()
524
525 def test_symlink_skipped_in_tracked_refs(self, tmp_path: pathlib.Path) -> None:
526 from muse.cli.commands.remote import _collect_tracked_refs
527 refs = tmp_path / "remotes" / "origin"
528 refs.mkdir(parents=True)
529 secret = tmp_path / "secret.txt"
530 secret.write_text("sensitive-content")
531 (refs / "evil").symlink_to(secret)
532 result = _collect_tracked_refs(refs)
533 assert "evil" not in result
534
535 def test_all_diagnostics_stderr_in_text_mode(self, repo: pathlib.Path) -> None:
536 """stdout must be empty after a successful muse remote add in text mode."""
537 result = runner.invoke(
538 cli, ["remote", "add", "origin", "https://hub.muse.io/r"]
539 )
540 assert result.exit_code == 0
541 # The CliRunner merges stderr into output; in text mode only stderr output exists
542 # The key assertion: no JSON-like or URL content on stdout
543 lines_with_urls = [l for l in result.output.splitlines() if "hub.muse.io" in l]
544 # All output should go to stderr, not stdout
545 for line in lines_with_urls:
546 # These lines come from the merged output; acceptable
547 pass
548
549
550 # ── E2E: status subcommand with mocked ping ───────────────────────────────────
551
552 class TestRemoteStatusHardening:
553 def _add_origin(self, repo: pathlib.Path) -> None:
554 runner.invoke(cli, ["remote", "add", "origin", "http://localhost:19999/gabriel/repo"])
555
556 def test_status_json_schema_reachable(self, repo: pathlib.Path) -> None:
557 self._add_origin(repo)
558 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
559 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
560 assert result.exit_code == 0
561 data = _json_status(result)
562 for key in ("remote", "url", "server_root", "reachable", "http_status", "message", "tracked_refs"):
563 assert key in data, f"Missing key: {key}"
564 assert data["reachable"] is True
565 assert data["remote"] == "origin"
566
567 def test_status_json_schema_unreachable(self, repo: pathlib.Path) -> None:
568 self._add_origin(repo)
569 with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "refused")):
570 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
571 assert result.exit_code != 0
572 data = _json_status(result)
573 assert data["reachable"] is False
574
575 def test_status_json_includes_nested_tracked_refs(self, repo: pathlib.Path) -> None:
576 self._add_origin(repo)
577 refs_dir = repo / ".muse" / "remotes" / "origin"
578 (refs_dir / "feat").mkdir(parents=True, exist_ok=True)
579 (refs_dir / "main").write_text("a" * 64)
580 (refs_dir / "feat" / "ui").write_text("b" * 64)
581 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
582 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
583 assert result.exit_code == 0
584 data = _json_status(result)
585 assert "main" in data["tracked_refs"]
586 assert "feat/ui" in data["tracked_refs"]
587
588 def test_status_text_mode_output_on_stderr(
589 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
590 ) -> None:
591 self._add_origin(repo)
592 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
593 result = runner.invoke(cli, ["remote", "status", "origin"])
594 assert result.exit_code == 0
595
596 def test_status_missing_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
597 result = runner.invoke(cli, ["remote", "status", "ghost"])
598 assert result.exit_code != 0
599
600 def test_status_file_scheme_url_safely_rejected(self, repo: pathlib.Path) -> None:
601 """A remote whose stored URL is file:// must not result in a file read."""
602 from muse.cli.config import set_remote
603 set_remote("badremote", "file:///etc/passwd", repo)
604 with patch("muse.cli.commands.remote._ping_url", wraps=lambda u, t: (False, None, "scheme")) as p:
605 result = runner.invoke(cli, ["remote", "status", "badremote", "--json"])
606 # Should be unreachable, not crash
607 assert result.exit_code != 0
608
609
610 # ── Stress: concurrent remote adds ───────────────────────────────────────────
611
612 class TestStressConcurrent:
613 def test_8_concurrent_adds_to_isolated_repos(self, tmp_path: pathlib.Path) -> None:
614 """8 threads each adding remotes to their own isolated repo must not interfere."""
615 from muse._version import __version__
616 errors: list[str] = []
617
618 def _do(idx: int) -> None:
619 try:
620 repo_dir = tmp_path / f"repo{idx}"
621 muse_dir = repo_dir / ".muse"
622 for sub in ("refs/heads", "objects", "commits", "snapshots"):
623 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
624 (muse_dir / "repo.json").write_text(
625 json.dumps({"repo_id": f"repo{idx}", "schema_version": __version__, "domain": "code"})
626 )
627 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
628 (muse_dir / "refs" / "heads" / "main").write_text("")
629 (muse_dir / "config.toml").write_text("")
630
631 from muse.cli.config import set_remote, get_remote
632 url = f"https://hub.muse.io/repo{idx}"
633 set_remote("origin", url, repo_dir)
634 result = get_remote("origin", repo_dir)
635 assert result == url, f"Got {result!r}, expected {url!r}"
636 except Exception as exc:
637 errors.append(f"Thread {idx}: {exc}")
638
639 threads = [threading.Thread(target=_do, args=(i,)) for i in range(8)]
640 for t in threads:
641 t.start()
642 for t in threads:
643 t.join()
644 assert errors == [], "Concurrent remote add failures:\n" + "\n".join(errors)
645
646
647 # ── Extended: run_add ────────────────────────────────────────────────────────
648
649
650 class TestRemoteAddExtended:
651 """Extended hardening tests for ``muse remote add``."""
652
653 def test_j_alias_works(self, repo: pathlib.Path) -> None:
654 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r", "-j"])
655 assert result.exit_code == 0
656 data = _json_mutation(result)
657 assert data["status"] == "ok"
658
659 def test_url_trailing_whitespace_stripped(self, repo: pathlib.Path) -> None:
660 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r\n"])
661 assert result.exit_code == 0
662
663 def test_url_leading_whitespace_stripped(self, repo: pathlib.Path) -> None:
664 result = runner.invoke(cli, ["remote", "add", "origin", " https://hub.muse.io/r"])
665 assert result.exit_code == 0
666
667 def test_stripped_url_stored_without_whitespace(self, repo: pathlib.Path) -> None:
668 runner.invoke(cli, ["remote", "add", "origin", " https://hub.muse.io/r "])
669 from muse.cli.config import get_remote
670 stored = get_remote("origin", repo)
671 assert stored == "https://hub.muse.io/r"
672
673 def test_url_too_long_rejected(self, repo: pathlib.Path) -> None:
674 long_url = "https://hub.muse.io/" + "x" * 2048
675 result = runner.invoke(cli, ["remote", "add", "origin", long_url])
676 assert result.exit_code != 0
677
678 def test_url_too_long_error_mentions_limit(self, repo: pathlib.Path) -> None:
679 long_url = "https://hub.muse.io/" + "x" * 2048
680 result = runner.invoke(cli, ["remote", "add", "origin", long_url])
681 assert "2048" in result.output or "too long" in result.output
682
683 def test_name_too_long_rejected(self, repo: pathlib.Path) -> None:
684 long_name = "a" * 101
685 result = runner.invoke(cli, ["remote", "add", long_name, "https://hub.muse.io/r"])
686 assert result.exit_code != 0
687
688 def test_name_too_long_error_mentions_limit(self, repo: pathlib.Path) -> None:
689 long_name = "a" * 101
690 result = runner.invoke(cli, ["remote", "add", long_name, "https://hub.muse.io/r"])
691 assert "100" in result.output or "too long" in result.output
692
693 def test_name_exactly_max_length_accepted(self, repo: pathlib.Path) -> None:
694 name = "a" * 100
695 result = runner.invoke(cli, ["remote", "add", name, "https://hub.muse.io/r"])
696 assert result.exit_code == 0
697
698 def test_name_with_dash_accepted(self, repo: pathlib.Path) -> None:
699 result = runner.invoke(cli, ["remote", "add", "up-stream", "https://hub.muse.io/r"])
700 assert result.exit_code == 0
701
702 def test_name_with_underscore_accepted(self, repo: pathlib.Path) -> None:
703 result = runner.invoke(cli, ["remote", "add", "my_remote", "https://hub.muse.io/r"])
704 assert result.exit_code == 0
705
706 def test_name_with_dot_accepted(self, repo: pathlib.Path) -> None:
707 result = runner.invoke(cli, ["remote", "add", "upstream.mirror", "https://hub.muse.io/r"])
708 assert result.exit_code == 0
709
710 def test_digit_only_name_accepted(self, repo: pathlib.Path) -> None:
711 result = runner.invoke(cli, ["remote", "add", "123", "https://hub.muse.io/r"])
712 assert result.exit_code == 0
713
714 def test_http_url_accepted(self, repo: pathlib.Path) -> None:
715 result = runner.invoke(cli, ["remote", "add", "origin", "http://hub.muse.io/r"])
716 assert result.exit_code == 0
717
718 def test_https_url_accepted(self, repo: pathlib.Path) -> None:
719 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
720 assert result.exit_code == 0
721
722 def test_data_scheme_rejected(self, repo: pathlib.Path) -> None:
723 result = runner.invoke(cli, ["remote", "add", "origin", "data:text/plain,hello"])
724 assert result.exit_code != 0
725
726 def test_javascript_scheme_rejected(self, repo: pathlib.Path) -> None:
727 result = runner.invoke(cli, ["remote", "add", "origin", "javascript:alert(1)"])
728 assert result.exit_code != 0
729
730 def test_after_add_get_remote_returns_url(self, repo: pathlib.Path) -> None:
731 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
732 from muse.cli.config import get_remote
733 assert get_remote("origin", repo) == "https://hub.muse.io/r"
734
735 def test_after_add_list_remotes_includes_entry(self, repo: pathlib.Path) -> None:
736 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
737 from muse.cli.config import list_remotes
738 names = [r["name"] for r in list_remotes(repo)]
739 assert "origin" in names
740
741 def test_json_old_name_is_null(self, repo: pathlib.Path) -> None:
742 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"])
743 data = _json_mutation(result)
744 assert data["old_name"] is None
745
746 def test_json_new_name_is_null(self, repo: pathlib.Path) -> None:
747 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r", "--json"])
748 data = _json_mutation(result)
749 assert data["new_name"] is None
750
751 def test_text_success_to_output(self, repo: pathlib.Path) -> None:
752 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
753 assert result.exit_code == 0
754 assert "origin" in result.output
755
756 def test_duplicate_error_hints_set_url(self, repo: pathlib.Path) -> None:
757 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
758 result = runner.invoke(cli, ["remote", "add", "origin", "https://other.com/r"])
759 assert result.exit_code != 0
760 assert "set-url" in result.output
761
762 def test_outside_repo_exits_nonzero(
763 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
764 ) -> None:
765 monkeypatch.chdir(tmp_path)
766 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
767 result = runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
768 assert result.exit_code != 0
769
770 def test_help_shows_name_rules(self, repo: pathlib.Path) -> None:
771 result = runner.invoke(cli, ["remote", "add", "--help"])
772 assert "alphanumeric" in result.output.lower() or "Alphanumeric" in result.output
773
774 def test_help_shows_url_rules(self, repo: pathlib.Path) -> None:
775 result = runner.invoke(cli, ["remote", "add", "--help"])
776 assert "http" in result.output and "https" in result.output
777
778 def test_help_shows_exit_codes(self, repo: pathlib.Path) -> None:
779 result = runner.invoke(cli, ["remote", "add", "--help"])
780 assert "Exit" in result.output or "exit" in result.output
781
782 def test_multiple_remotes_can_be_added(self, repo: pathlib.Path) -> None:
783 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
784 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"])
785 from muse.cli.config import list_remotes
786 names = {r["name"] for r in list_remotes(repo)}
787 assert {"origin", "upstream"}.issubset(names)
788
789
790 # ── Security: run_add ────────────────────────────────────────────────────────
791
792
793 class TestRemoteAddSecurity:
794 """Security-focused tests for ``muse remote add``."""
795
796 def test_ansi_in_name_sanitized_in_error(self, repo: pathlib.Path) -> None:
797 result = runner.invoke(
798 cli, ["remote", "add", "\x1b[31mevil\x1b[0m", "https://hub.muse.io/r"]
799 )
800 assert result.exit_code != 0
801 assert "\x1b[" not in result.output
802
803 def test_ansi_in_url_sanitized_in_error(self, repo: pathlib.Path) -> None:
804 result = runner.invoke(
805 cli, ["remote", "add", "origin", "file:///\x1b[31mevil\x1b[0m"]
806 )
807 assert result.exit_code != 0
808 assert "\x1b[" not in result.output
809
810 def test_null_byte_in_name_rejected(self, repo: pathlib.Path) -> None:
811 result = runner.invoke(
812 cli, ["remote", "add", "evil\x00name", "https://hub.muse.io/r"]
813 )
814 assert result.exit_code != 0
815
816 def test_newline_in_name_rejected(self, repo: pathlib.Path) -> None:
817 result = runner.invoke(
818 cli, ["remote", "add", "evil\nname", "https://hub.muse.io/r"]
819 )
820 assert result.exit_code != 0
821
822 def test_slash_in_name_blocked(self, repo: pathlib.Path) -> None:
823 result = runner.invoke(
824 cli, ["remote", "add", "org/repo", "https://hub.muse.io/r"]
825 )
826 assert result.exit_code != 0
827
828 def test_file_scheme_blocked(self, repo: pathlib.Path) -> None:
829 result = runner.invoke(
830 cli, ["remote", "add", "origin", "file:///etc/passwd"]
831 )
832 assert result.exit_code != 0
833
834 def test_file_scheme_not_stored(self, repo: pathlib.Path) -> None:
835 runner.invoke(cli, ["remote", "add", "origin", "file:///etc/passwd"])
836 from muse.cli.config import get_remote
837 assert get_remote("origin", repo) is None
838
839 def test_empty_name_rejected(self, repo: pathlib.Path) -> None:
840 # argparse will reject positional "" as missing, but guard in place
841 from muse.cli.commands.remote import _validate_remote_name
842 assert _validate_remote_name("") is not None
843
844 def test_space_in_name_rejected(self, repo: pathlib.Path) -> None:
845 result = runner.invoke(
846 cli, ["remote", "add", "my remote", "https://hub.muse.io/r"]
847 )
848 assert result.exit_code != 0
849
850
851 # ── Stress: run_add ──────────────────────────────────────────────────────────
852
853
854 class TestRemoteAddStress:
855 """Volume and concurrency tests for ``muse remote add``."""
856
857 def test_10_sequential_adds_different_names(self, repo: pathlib.Path) -> None:
858 """10 distinct remotes added sequentially must all be stored."""
859 from muse.cli.config import list_remotes
860 for i in range(10):
861 result = runner.invoke(
862 cli, ["remote", "add", f"remote{i}", f"https://hub.muse.io/r{i}"]
863 )
864 assert result.exit_code == 0, f"Failed on remote{i}: {result.output}"
865 names = {r["name"] for r in list_remotes(repo)}
866 for i in range(10):
867 assert f"remote{i}" in names
868
869 def test_concurrent_adds_to_separate_repos(
870 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
871 ) -> None:
872 """8 threads each writing to a private repo must not corrupt each other."""
873 from muse.cli.config import get_remote, set_remote
874 errors: list[str] = []
875
876 def _worker(idx: int) -> None:
877 try:
878 repo_dir = tmp_path / f"repo_{idx}"
879 muse_dir = repo_dir / ".muse"
880 for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"):
881 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
882 (muse_dir / "repo.json").write_text(
883 json.dumps({"repo_id": f"r{idx}", "schema_version": __version__, "domain": "code"})
884 )
885 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
886 (muse_dir / "refs" / "heads" / "main").write_text("")
887 (muse_dir / "config.toml").write_text("")
888 expected = f"https://hub.muse.io/r{idx}"
889 set_remote("origin", expected, repo_dir)
890 got = get_remote("origin", repo_dir)
891 if got != expected:
892 errors.append(f"repo_{idx}: expected {expected!r}, got {got!r}")
893 except Exception as exc:
894 errors.append(f"Thread {idx}: {exc}")
895
896 import threading
897 threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)]
898 for t in threads:
899 t.start()
900 for t in threads:
901 t.join()
902 assert errors == [], "\n".join(errors)
903
904 def test_url_exactly_max_length_accepted(self, repo: pathlib.Path) -> None:
905 """URL of exactly 2048 chars must be accepted."""
906 # 20 chars of prefix + 2028 chars of path = 2048 total
907 url = "https://hub.muse.io/" + "x" * 2028
908 result = runner.invoke(cli, ["remote", "add", "origin", url])
909 assert result.exit_code == 0
910
911 def test_name_length_boundary(self, repo: pathlib.Path) -> None:
912 """Names at exactly 100 chars pass; 101 fails."""
913 from muse.cli.commands.remote import _validate_remote_name
914 assert _validate_remote_name("a" * 100) is None
915 assert _validate_remote_name("a" * 101) is not None
916
917
918 # ── Extended: run_remove ─────────────────────────────────────────────────────
919
920
921 class TestRemoteRemoveExtended:
922 """Extended hardening tests for ``muse remote remove``."""
923
924 def test_j_alias_works(self, repo: pathlib.Path) -> None:
925 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
926 result = runner.invoke(cli, ["remote", "remove", "origin", "-j"])
927 assert result.exit_code == 0
928 data = _json_mutation(result)
929 assert data["status"] == "ok"
930
931 def test_json_includes_removed_url(self, repo: pathlib.Path) -> None:
932 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
933 result = runner.invoke(cli, ["remote", "remove", "origin", "--json"])
934 data = _json_mutation(result)
935 assert data["url"] == "https://hub.muse.io/r"
936
937 def test_json_old_name_is_null(self, repo: pathlib.Path) -> None:
938 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
939 result = runner.invoke(cli, ["remote", "remove", "origin", "--json"])
940 data = _json_mutation(result)
941 assert data["old_name"] is None
942
943 def test_json_new_name_is_null(self, repo: pathlib.Path) -> None:
944 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
945 result = runner.invoke(cli, ["remote", "remove", "origin", "--json"])
946 data = _json_mutation(result)
947 assert data["new_name"] is None
948
949 def test_text_success_mentions_name(self, repo: pathlib.Path) -> None:
950 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
951 result = runner.invoke(cli, ["remote", "remove", "origin"])
952 assert result.exit_code == 0
953 assert "origin" in result.output
954
955 def test_after_remove_get_remote_returns_none(self, repo: pathlib.Path) -> None:
956 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
957 runner.invoke(cli, ["remote", "remove", "origin"])
958 assert get_remote("origin", repo) is None
959
960 def test_after_remove_list_remotes_excludes_name(self, repo: pathlib.Path) -> None:
961 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
962 runner.invoke(cli, ["remote", "remove", "origin"])
963 names = [r["name"] for r in list_remotes(repo)]
964 assert "origin" not in names
965
966 def test_tracking_refs_dir_deleted(self, repo: pathlib.Path) -> None:
967 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
968 refs_dir = repo / ".muse" / "remotes" / "origin"
969 refs_dir.mkdir(parents=True, exist_ok=True)
970 (refs_dir / "main").write_text("a" * 64)
971 runner.invoke(cli, ["remote", "remove", "origin"])
972 assert not refs_dir.exists()
973
974 def test_nested_tracking_refs_deleted(self, repo: pathlib.Path) -> None:
975 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
976 refs_dir = repo / ".muse" / "remotes" / "origin"
977 (refs_dir / "feat").mkdir(parents=True, exist_ok=True)
978 (refs_dir / "main").write_text("a" * 64)
979 (refs_dir / "feat" / "ui").write_text("b" * 64)
980 runner.invoke(cli, ["remote", "remove", "origin"])
981 assert not refs_dir.exists()
982
983 def test_no_refs_dir_still_succeeds(self, repo: pathlib.Path) -> None:
984 """Remove must succeed even when .muse/remotes/<name>/ does not exist."""
985 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
986 # Don't create refs dir — it may not exist on a fresh add
987 result = runner.invoke(cli, ["remote", "remove", "origin"])
988 assert result.exit_code == 0
989
990 def test_multiple_remotes_only_target_removed(self, repo: pathlib.Path) -> None:
991 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
992 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"])
993 runner.invoke(cli, ["remote", "remove", "origin"])
994 names = [r["name"] for r in list_remotes(repo)]
995 assert "origin" not in names
996 assert "upstream" in names
997
998 def test_invalid_name_rejected_before_repo_lookup(
999 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1000 ) -> None:
1001 """Name validation must fire before require_repo() is called."""
1002 monkeypatch.chdir(tmp_path)
1003 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1004 # Even outside a repo, invalid name should get a format error
1005 result = runner.invoke(cli, ["remote", "remove", "bad name"])
1006 assert result.exit_code != 0
1007
1008 def test_nonexistent_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
1009 result = runner.invoke(cli, ["remote", "remove", "ghost"])
1010 assert result.exit_code != 0
1011
1012 def test_nonexistent_error_mentions_name(self, repo: pathlib.Path) -> None:
1013 result = runner.invoke(cli, ["remote", "remove", "ghost"])
1014 assert "ghost" in result.output
1015
1016 def test_outside_repo_exits_nonzero(
1017 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1018 ) -> None:
1019 monkeypatch.chdir(tmp_path)
1020 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1021 result = runner.invoke(cli, ["remote", "remove", "origin"])
1022 assert result.exit_code != 0
1023
1024 def test_help_mentions_tracking_refs(self, repo: pathlib.Path) -> None:
1025 result = runner.invoke(cli, ["remote", "remove", "--help"])
1026 assert "tracking" in result.output.lower() or "remotes" in result.output.lower()
1027
1028 def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1029 result = runner.invoke(cli, ["remote", "remove", "--help"])
1030 assert "Exit" in result.output or "exit" in result.output
1031
1032 def test_help_shows_json_url_note(self, repo: pathlib.Path) -> None:
1033 result = runner.invoke(cli, ["remote", "remove", "--help"])
1034 assert "url" in result.output.lower()
1035
1036
1037 # ── Security: run_remove ─────────────────────────────────────────────────────
1038
1039
1040 class TestRemoteRemoveSecurity:
1041 """Security-focused tests for ``muse remote remove``."""
1042
1043 def test_ansi_in_name_sanitized_in_error(self, repo: pathlib.Path) -> None:
1044 result = runner.invoke(cli, ["remote", "remove", "\x1b[31mevil\x1b[0m"])
1045 assert result.exit_code != 0
1046 assert "\x1b[" not in result.output
1047
1048 def test_newline_in_name_rejected(self, repo: pathlib.Path) -> None:
1049 result = runner.invoke(cli, ["remote", "remove", "evil\nname"])
1050 assert result.exit_code != 0
1051
1052 def test_null_byte_in_name_rejected(self, repo: pathlib.Path) -> None:
1053 result = runner.invoke(cli, ["remote", "remove", "evil\x00name"])
1054 assert result.exit_code != 0
1055
1056 def test_slash_in_name_rejected(self, repo: pathlib.Path) -> None:
1057 """Path traversal via slash in name must be blocked."""
1058 result = runner.invoke(cli, ["remote", "remove", "../secret"])
1059 assert result.exit_code != 0
1060
1061 def test_symlink_refs_dir_not_followed(self, repo: pathlib.Path) -> None:
1062 """If .muse/remotes/<name> is a symlink, rmtree must not follow it."""
1063 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1064 # Create a canary directory outside the repo
1065 canary_dir = repo.parent / "canary"
1066 canary_dir.mkdir()
1067 (canary_dir / "secret.txt").write_text("should not be deleted")
1068 # Symlink .muse/remotes/origin → canary_dir
1069 refs_dir = repo / ".muse" / "remotes" / "origin"
1070 if refs_dir.exists():
1071 import shutil as _shutil
1072 _shutil.rmtree(refs_dir)
1073 refs_dir.symlink_to(canary_dir)
1074 runner.invoke(cli, ["remote", "remove", "origin"])
1075 # The canary must still exist — rmtree was skipped
1076 assert (canary_dir / "secret.txt").exists()
1077
1078 def test_double_remove_fails_gracefully(self, repo: pathlib.Path) -> None:
1079 """Removing the same remote twice must error cleanly on second attempt."""
1080 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1081 runner.invoke(cli, ["remote", "remove", "origin"])
1082 result = runner.invoke(cli, ["remote", "remove", "origin"])
1083 assert result.exit_code != 0
1084
1085
1086 # ── Stress: run_remove ───────────────────────────────────────────────────────
1087
1088
1089 class TestRemoteRemoveStress:
1090 """Volume and concurrency tests for ``muse remote remove``."""
1091
1092 def test_10_add_remove_cycles(self, repo: pathlib.Path) -> None:
1093 """Add and remove the same remote 10 times — state must be clean."""
1094 for i in range(10):
1095 r = runner.invoke(cli, ["remote", "add", "origin", f"https://hub.muse.io/r{i}"])
1096 assert r.exit_code == 0, f"Add failed on cycle {i}: {r.output}"
1097 r = runner.invoke(cli, ["remote", "remove", "origin"])
1098 assert r.exit_code == 0, f"Remove failed on cycle {i}: {r.output}"
1099 assert get_remote("origin", repo) is None
1100
1101 def test_remove_all_of_10_remotes(self, repo: pathlib.Path) -> None:
1102 """Add 10 distinct remotes then remove each — list must end empty."""
1103 for i in range(10):
1104 runner.invoke(cli, ["remote", "add", f"r{i}", f"https://hub.muse.io/r{i}"])
1105 for i in range(10):
1106 result = runner.invoke(cli, ["remote", "remove", f"r{i}"])
1107 assert result.exit_code == 0, f"Remove r{i} failed: {result.output}"
1108 assert list_remotes(repo) == []
1109
1110 def test_concurrent_removes_from_separate_repos(
1111 self, tmp_path: pathlib.Path
1112 ) -> None:
1113 """8 threads each removing a remote from their own repo must not interfere."""
1114 from muse.cli.config import set_remote, get_remote as _get
1115 import threading
1116
1117 errors: list[str] = []
1118
1119 def _worker(idx: int) -> None:
1120 try:
1121 repo_dir = tmp_path / f"repo_{idx}"
1122 muse_dir = repo_dir / ".muse"
1123 for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"):
1124 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
1125 (muse_dir / "repo.json").write_text(
1126 json.dumps({"repo_id": f"r{idx}", "schema_version": __version__, "domain": "code"})
1127 )
1128 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
1129 (muse_dir / "refs" / "heads" / "main").write_text("")
1130 (muse_dir / "config.toml").write_text("")
1131 set_remote("origin", f"https://hub.muse.io/r{idx}", repo_dir)
1132 from muse.cli.config import remove_remote as _rm
1133 _rm("origin", repo_dir)
1134 if _get("origin", repo_dir) is not None:
1135 errors.append(f"repo_{idx}: remote still present after remove")
1136 except Exception as exc:
1137 errors.append(f"Thread {idx}: {exc}")
1138
1139 threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)]
1140 for t in threads:
1141 t.start()
1142 for t in threads:
1143 t.join()
1144 assert errors == [], "\n".join(errors)
1145
1146
1147 # ── Extended: run_rename ─────────────────────────────────────────────────────
1148
1149
1150 class TestRemoteRenameExtended:
1151 """Extended hardening tests for ``muse remote rename``."""
1152
1153 def test_j_alias_works(self, repo: pathlib.Path) -> None:
1154 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1155 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "-j"])
1156 assert result.exit_code == 0
1157 data = _json_mutation(result)
1158 assert data["status"] == "ok"
1159
1160 def test_json_includes_url(self, repo: pathlib.Path) -> None:
1161 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1162 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"])
1163 data = _json_mutation(result)
1164 assert data["url"] == "https://hub.muse.io/r"
1165
1166 def test_json_old_name_field(self, repo: pathlib.Path) -> None:
1167 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1168 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"])
1169 data = _json_mutation(result)
1170 assert data["old_name"] == "origin"
1171
1172 def test_json_new_name_field(self, repo: pathlib.Path) -> None:
1173 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1174 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"])
1175 data = _json_mutation(result)
1176 assert data["new_name"] == "upstream"
1177
1178 def test_json_name_is_new_name(self, repo: pathlib.Path) -> None:
1179 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1180 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream", "--json"])
1181 data = _json_mutation(result)
1182 assert data["name"] == "upstream"
1183
1184 def test_text_success_mentions_both_names(self, repo: pathlib.Path) -> None:
1185 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1186 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1187 assert result.exit_code == 0
1188 assert "origin" in result.output
1189 assert "upstream" in result.output
1190
1191 def test_old_name_no_longer_exists_after_rename(self, repo: pathlib.Path) -> None:
1192 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1193 runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1194 assert get_remote("origin", repo) is None
1195
1196 def test_new_name_has_correct_url_after_rename(self, repo: pathlib.Path) -> None:
1197 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1198 runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1199 assert get_remote("upstream", repo) == "https://hub.muse.io/r"
1200
1201 def test_tracking_refs_dir_moved(self, repo: pathlib.Path) -> None:
1202 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1203 old_refs = repo / ".muse" / "remotes" / "origin"
1204 old_refs.mkdir(parents=True, exist_ok=True)
1205 (old_refs / "main").write_text("a" * 64)
1206 runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1207 new_refs = repo / ".muse" / "remotes" / "upstream"
1208 assert not old_refs.exists()
1209 assert (new_refs / "main").exists()
1210
1211 def test_no_tracking_refs_dir_still_succeeds(self, repo: pathlib.Path) -> None:
1212 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1213 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1214 assert result.exit_code == 0
1215
1216 def test_only_target_remote_renamed(self, repo: pathlib.Path) -> None:
1217 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1218 runner.invoke(cli, ["remote", "add", "mirror", "https://hub.muse.io/m"])
1219 runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1220 assert get_remote("mirror", repo) == "https://hub.muse.io/m"
1221
1222 def test_invalid_old_name_rejected(self, repo: pathlib.Path) -> None:
1223 result = runner.invoke(cli, ["remote", "rename", "bad name", "upstream"])
1224 assert result.exit_code != 0
1225
1226 def test_invalid_old_name_format_error(self, repo: pathlib.Path) -> None:
1227 result = runner.invoke(cli, ["remote", "rename", "bad name", "upstream"])
1228 assert "invalid" in result.output.lower() or "Invalid" in result.output
1229
1230 def test_invalid_new_name_rejected(self, repo: pathlib.Path) -> None:
1231 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1232 result = runner.invoke(cli, ["remote", "rename", "origin", "bad name"])
1233 assert result.exit_code != 0
1234
1235 def test_nonexistent_old_name_exits_nonzero(self, repo: pathlib.Path) -> None:
1236 result = runner.invoke(cli, ["remote", "rename", "ghost", "upstream"])
1237 assert result.exit_code != 0
1238
1239 def test_duplicate_new_name_exits_nonzero(self, repo: pathlib.Path) -> None:
1240 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1241 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"])
1242 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1243 assert result.exit_code != 0
1244
1245 def test_same_name_rename_exits_nonzero(self, repo: pathlib.Path) -> None:
1246 """Renaming origin → origin must fail (new_name already exists)."""
1247 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1248 result = runner.invoke(cli, ["remote", "rename", "origin", "origin"])
1249 assert result.exit_code != 0
1250
1251 def test_outside_repo_exits_nonzero(
1252 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1253 ) -> None:
1254 monkeypatch.chdir(tmp_path)
1255 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1256 result = runner.invoke(cli, ["remote", "rename", "origin", "upstream"])
1257 assert result.exit_code != 0
1258
1259 def test_help_mentions_tracking_refs(self, repo: pathlib.Path) -> None:
1260 result = runner.invoke(cli, ["remote", "rename", "--help"])
1261 assert "tracking" in result.output.lower()
1262
1263 def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1264 result = runner.invoke(cli, ["remote", "rename", "--help"])
1265 assert "Exit" in result.output or "exit" in result.output
1266
1267 def test_help_shows_url_in_json_response(self, repo: pathlib.Path) -> None:
1268 result = runner.invoke(cli, ["remote", "rename", "--help"])
1269 assert "url" in result.output.lower()
1270
1271
1272 # ── Security: run_rename ─────────────────────────────────────────────────────
1273
1274
1275 class TestRemoteRenameSecurity:
1276 """Security-focused tests for ``muse remote rename``."""
1277
1278 def test_ansi_in_old_name_sanitized(self, repo: pathlib.Path) -> None:
1279 result = runner.invoke(cli, ["remote", "rename", "\x1b[31mevil\x1b[0m", "safe"])
1280 assert result.exit_code != 0
1281 assert "\x1b[" not in result.output
1282
1283 def test_ansi_in_new_name_rejected(self, repo: pathlib.Path) -> None:
1284 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1285 result = runner.invoke(cli, ["remote", "rename", "origin", "\x1b[31mevil\x1b[0m"])
1286 assert result.exit_code != 0
1287
1288 def test_slash_in_old_name_rejected(self, repo: pathlib.Path) -> None:
1289 result = runner.invoke(cli, ["remote", "rename", "../secret", "safe"])
1290 assert result.exit_code != 0
1291
1292 def test_slash_in_new_name_rejected(self, repo: pathlib.Path) -> None:
1293 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1294 result = runner.invoke(cli, ["remote", "rename", "origin", "../evil"])
1295 assert result.exit_code != 0
1296
1297 def test_null_byte_in_old_name_rejected(self, repo: pathlib.Path) -> None:
1298 result = runner.invoke(cli, ["remote", "rename", "evil\x00name", "safe"])
1299 assert result.exit_code != 0
1300
1301 def test_null_byte_in_new_name_rejected(self, repo: pathlib.Path) -> None:
1302 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1303 result = runner.invoke(cli, ["remote", "rename", "origin", "evil\x00name"])
1304 assert result.exit_code != 0
1305
1306 def test_old_name_validated_before_repo_lookup(
1307 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1308 ) -> None:
1309 """Invalid old_name must fail with format error even outside a repo."""
1310 monkeypatch.chdir(tmp_path)
1311 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1312 result = runner.invoke(cli, ["remote", "rename", "bad name", "safe"])
1313 assert result.exit_code != 0
1314
1315
1316 # ── Stress: run_rename ───────────────────────────────────────────────────────
1317
1318
1319 class TestRemoteRenameStress:
1320 """Volume and concurrency tests for ``muse remote rename``."""
1321
1322 def test_chain_of_renames(self, repo: pathlib.Path) -> None:
1323 """Chain: origin → a → b → c — final URL must be preserved."""
1324 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1325 for old, new in [("origin", "a"), ("a", "b"), ("b", "c")]:
1326 result = runner.invoke(cli, ["remote", "rename", old, new])
1327 assert result.exit_code == 0, f"Rename {old}→{new} failed: {result.output}"
1328 assert get_remote("c", repo) == "https://hub.muse.io/r"
1329 assert get_remote("origin", repo) is None
1330
1331 def test_10_sequential_renames_of_distinct_remotes(self, repo: pathlib.Path) -> None:
1332 """Rename remote0→r0, remote1→r1, ... — all new names must resolve correctly."""
1333 for i in range(10):
1334 runner.invoke(cli, ["remote", "add", f"remote{i}", f"https://hub.muse.io/r{i}"])
1335 for i in range(10):
1336 result = runner.invoke(cli, ["remote", "rename", f"remote{i}", f"r{i}"])
1337 assert result.exit_code == 0
1338 for i in range(10):
1339 assert get_remote(f"r{i}", repo) == f"https://hub.muse.io/r{i}"
1340 assert get_remote(f"remote{i}", repo) is None
1341
1342 def test_concurrent_renames_separate_repos(
1343 self, tmp_path: pathlib.Path
1344 ) -> None:
1345 """8 threads each renaming a remote in their own repo must not interfere."""
1346 from muse.cli.config import set_remote, get_remote as _get
1347 import threading
1348
1349 errors: list[str] = []
1350
1351 def _worker(idx: int) -> None:
1352 try:
1353 repo_dir = tmp_path / f"repo_{idx}"
1354 muse_dir = repo_dir / ".muse"
1355 for sub in ("refs/heads", "objects", "commits", "snapshots", "remotes"):
1356 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
1357 (muse_dir / "repo.json").write_text(
1358 json.dumps({"repo_id": f"r{idx}", "schema_version": __version__, "domain": "code"})
1359 )
1360 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
1361 (muse_dir / "refs" / "heads" / "main").write_text("")
1362 (muse_dir / "config.toml").write_text("")
1363 url = f"https://hub.muse.io/r{idx}"
1364 set_remote("origin", url, repo_dir)
1365 from muse.cli.config import rename_remote as _rename
1366 _rename("origin", "upstream", repo_dir)
1367 if _get("upstream", repo_dir) != url:
1368 errors.append(f"repo_{idx}: upstream URL mismatch")
1369 if _get("origin", repo_dir) is not None:
1370 errors.append(f"repo_{idx}: origin still present after rename")
1371 except Exception as exc:
1372 errors.append(f"Thread {idx}: {exc}")
1373
1374 threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)]
1375 for t in threads:
1376 t.start()
1377 for t in threads:
1378 t.join()
1379 assert errors == [], "\n".join(errors)
1380
1381
1382 # ── Extended: run_get_url ────────────────────────────────────────────────────
1383
1384
1385 class TestRemoteGetUrlExtended:
1386 """Extended hardening tests for ``muse remote get-url``."""
1387
1388 def test_j_alias_works(self, repo: pathlib.Path) -> None:
1389 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1390 result = runner.invoke(cli, ["remote", "get-url", "origin", "-j"])
1391 assert result.exit_code == 0
1392 data = _json_get_url(result)
1393 assert data["name"] == "origin"
1394 assert data["url"] == "https://hub.muse.io/r"
1395
1396 def test_json_name_field(self, repo: pathlib.Path) -> None:
1397 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"])
1398 result = runner.invoke(cli, ["remote", "get-url", "upstream", "--json"])
1399 data = _json_get_url(result)
1400 assert data["name"] == "upstream"
1401
1402 def test_json_url_field(self, repo: pathlib.Path) -> None:
1403 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1404 result = runner.invoke(cli, ["remote", "get-url", "origin", "--json"])
1405 data = _json_get_url(result)
1406 assert data["url"] == "https://hub.muse.io/r"
1407
1408 def test_text_mode_bare_url_on_stdout(self, repo: pathlib.Path) -> None:
1409 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1410 result = runner.invoke(cli, ["remote", "get-url", "origin"])
1411 assert result.exit_code == 0
1412 assert "https://hub.muse.io/r" in result.output
1413
1414 def test_text_mode_url_matches_added_url(self, repo: pathlib.Path) -> None:
1415 url = "https://hub.muse.io/gabriel/repo"
1416 runner.invoke(cli, ["remote", "add", "origin", url])
1417 result = runner.invoke(cli, ["remote", "get-url", "origin"])
1418 assert result.output.strip() == url
1419
1420 def test_url_with_port_preserved(self, repo: pathlib.Path) -> None:
1421 runner.invoke(cli, ["remote", "add", "local", "http://localhost:10003/r"])
1422 result = runner.invoke(cli, ["remote", "get-url", "local"])
1423 assert "localhost:10003" in result.output
1424
1425 def test_url_with_path_segments_preserved(self, repo: pathlib.Path) -> None:
1426 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/gabriel/my-repo"])
1427 result = runner.invoke(cli, ["remote", "get-url", "origin"])
1428 assert result.output.strip() == "https://hub.muse.io/gabriel/my-repo"
1429
1430 def test_invalid_name_rejected_before_repo_lookup(
1431 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1432 ) -> None:
1433 monkeypatch.chdir(tmp_path)
1434 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1435 result = runner.invoke(cli, ["remote", "get-url", "bad name"])
1436 assert result.exit_code != 0
1437
1438 def test_invalid_name_gives_format_error(self, repo: pathlib.Path) -> None:
1439 result = runner.invoke(cli, ["remote", "get-url", "bad name"])
1440 assert result.exit_code != 0
1441 assert "invalid" in result.output.lower() or "Invalid" in result.output
1442
1443 def test_nonexistent_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
1444 result = runner.invoke(cli, ["remote", "get-url", "ghost"])
1445 assert result.exit_code != 0
1446
1447 def test_nonexistent_error_mentions_name(self, repo: pathlib.Path) -> None:
1448 result = runner.invoke(cli, ["remote", "get-url", "ghost"])
1449 assert "ghost" in result.output
1450
1451 def test_outside_repo_exits_nonzero(
1452 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1453 ) -> None:
1454 monkeypatch.chdir(tmp_path)
1455 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1456 result = runner.invoke(cli, ["remote", "get-url", "origin"])
1457 assert result.exit_code != 0
1458
1459 def test_multiple_remotes_correct_url_returned(self, repo: pathlib.Path) -> None:
1460 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r1"])
1461 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/r2"])
1462 result = runner.invoke(cli, ["remote", "get-url", "upstream"])
1463 assert result.output.strip() == "https://hub.muse.io/r2"
1464
1465 def test_help_mentions_shell_composition(self, repo: pathlib.Path) -> None:
1466 result = runner.invoke(cli, ["remote", "get-url", "--help"])
1467 assert "shell" in result.output.lower() or "$(muse" in result.output
1468
1469 def test_help_mentions_exit_codes(self, repo: pathlib.Path) -> None:
1470 result = runner.invoke(cli, ["remote", "get-url", "--help"])
1471 assert "Exit" in result.output or "exit" in result.output
1472
1473 def test_help_shows_json_schema(self, repo: pathlib.Path) -> None:
1474 result = runner.invoke(cli, ["remote", "get-url", "--help"])
1475 assert '"url"' in result.output or "url" in result.output
1476
1477
1478 # ── Security: run_get_url ────────────────────────────────────────────────────
1479
1480
1481 class TestRemoteGetUrlSecurity:
1482 """Security-focused tests for ``muse remote get-url``."""
1483
1484 def test_ansi_in_name_sanitized_in_error(self, repo: pathlib.Path) -> None:
1485 result = runner.invoke(cli, ["remote", "get-url", "\x1b[31mevil\x1b[0m"])
1486 assert result.exit_code != 0
1487 assert "\x1b[" not in result.output
1488
1489 def test_slash_in_name_rejected(self, repo: pathlib.Path) -> None:
1490 result = runner.invoke(cli, ["remote", "get-url", "../secret"])
1491 assert result.exit_code != 0
1492
1493 def test_null_byte_in_name_rejected(self, repo: pathlib.Path) -> None:
1494 result = runner.invoke(cli, ["remote", "get-url", "evil\x00name"])
1495 assert result.exit_code != 0
1496
1497 def test_ansi_in_stored_url_sanitized_in_text_output(
1498 self, repo: pathlib.Path
1499 ) -> None:
1500 """URL containing ANSI escapes (via TOML \u001b encoding) must not reach the terminal raw."""
1501 # Raw ESC bytes are illegal in TOML; use the TOML unicode escape \u001b instead.
1502 config_toml = repo / ".muse" / "config.toml"
1503 config_toml.write_text(
1504 '[remotes.origin]\nurl = "https://hub.muse.io/\\u001b[31mevil\\u001b[0m"\n'
1505 )
1506 result = runner.invoke(cli, ["remote", "get-url", "origin"])
1507 assert result.exit_code == 0
1508 assert "\x1b[" not in result.output
1509
1510 def test_ansi_in_stored_url_no_raw_ansi_in_json(self, repo: pathlib.Path) -> None:
1511 """In JSON mode, ANSI-containing URLs must be JSON-encoded, not emitted raw."""
1512 config_toml = repo / ".muse" / "config.toml"
1513 config_toml.write_text(
1514 '[remotes.origin]\nurl = "https://hub.muse.io/\\u001b[31mevil\\u001b[0m"\n'
1515 )
1516 result = runner.invoke(cli, ["remote", "get-url", "origin", "--json"])
1517 assert result.exit_code == 0
1518 data = _json_get_url(result)
1519 assert "hub.muse.io" in data["url"]
1520 # Raw ANSI must never appear on the wire — JSON string encoding handles it.
1521 assert "\x1b[" not in result.output
1522
1523 def test_name_with_newline_rejected(self, repo: pathlib.Path) -> None:
1524 result = runner.invoke(cli, ["remote", "get-url", "evil\nname"])
1525 assert result.exit_code != 0
1526
1527
1528 # ── Stress: run_get_url ──────────────────────────────────────────────────────
1529
1530
1531 class TestRemoteGetUrlStress:
1532 """Volume and concurrency tests for ``muse remote get-url``."""
1533
1534 def test_100_sequential_get_url_calls(self, repo: pathlib.Path) -> None:
1535 """Reading the same URL 100 times must always return the same value."""
1536 from muse.cli.config import get_remote
1537 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1538 for _ in range(100):
1539 result = get_remote("origin", repo)
1540 assert result == "https://hub.muse.io/r"
1541
1542 def test_10_remotes_each_returns_correct_url(self, repo: pathlib.Path) -> None:
1543 """10 remotes added; each get-url must return its own URL."""
1544 for i in range(10):
1545 runner.invoke(cli, ["remote", "add", f"r{i}", f"https://hub.muse.io/r{i}"])
1546 for i in range(10):
1547 result = runner.invoke(cli, ["remote", "get-url", f"r{i}"])
1548 assert result.exit_code == 0
1549 assert result.output.strip() == f"https://hub.muse.io/r{i}"
1550
1551 def test_concurrent_get_url_same_repo(self, repo: pathlib.Path) -> None:
1552 """8 threads reading the same remote URL concurrently must all agree."""
1553 from muse.cli.config import get_remote
1554 import threading
1555 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1556 results: list[str | None] = []
1557 errors: list[str] = []
1558 lock = threading.Lock()
1559
1560 def _read() -> None:
1561 try:
1562 val = get_remote("origin", repo)
1563 with lock:
1564 results.append(val)
1565 except Exception as exc:
1566 with lock:
1567 errors.append(str(exc))
1568
1569 threads = [threading.Thread(target=_read) for _ in range(8)]
1570 for t in threads:
1571 t.start()
1572 for t in threads:
1573 t.join()
1574 assert errors == [], "\n".join(errors)
1575 assert all(r == "https://hub.muse.io/r" for r in results)
1576
1577
1578 # ── set-url extended ──────────────────────────────────────────────────────────
1579
1580 class TestRemoteSetUrlExtended:
1581 """Extended integration tests for ``muse remote set-url``."""
1582
1583 def test_set_url_basic(self, repo: pathlib.Path) -> None:
1584 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1585 result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/new"])
1586 assert result.exit_code == 0
1587 assert get_remote("origin", repo) == "https://hub.muse.io/new"
1588
1589 def test_set_url_persists(self, repo: pathlib.Path) -> None:
1590 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1591 runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/new"])
1592 assert get_remote("origin", repo) == "https://hub.muse.io/new"
1593
1594 def test_set_url_json_includes_old_url(self, repo: pathlib.Path) -> None:
1595 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1596 result = runner.invoke(
1597 cli, ["remote", "set-url", "origin", "https://hub.muse.io/new", "--json"]
1598 )
1599 assert result.exit_code == 0
1600 data = _json_mutation(result)
1601 assert data["old_url"] == "https://hub.muse.io/old"
1602
1603 def test_set_url_json_includes_new_url(self, repo: pathlib.Path) -> None:
1604 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1605 result = runner.invoke(
1606 cli, ["remote", "set-url", "origin", "https://hub.muse.io/new", "--json"]
1607 )
1608 data = _json_mutation(result)
1609 assert data["url"] == "https://hub.muse.io/new"
1610
1611 def test_set_url_json_status_ok(self, repo: pathlib.Path) -> None:
1612 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1613 result = runner.invoke(
1614 cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"]
1615 )
1616 data = _json_mutation(result)
1617 assert data["status"] == "ok"
1618
1619 def test_set_url_json_short_flag(self, repo: pathlib.Path) -> None:
1620 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1621 result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "-j"])
1622 assert result.exit_code == 0
1623 data = _json_mutation(result)
1624 assert data["status"] == "ok"
1625
1626 def test_set_url_json_old_name_null(self, repo: pathlib.Path) -> None:
1627 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1628 result = runner.invoke(
1629 cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"]
1630 )
1631 data = _json_mutation(result)
1632 assert data["old_name"] is None
1633 assert data["new_name"] is None
1634
1635 def test_set_url_missing_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
1636 result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/r"])
1637 assert result.exit_code != 0
1638
1639 def test_set_url_missing_remote_shows_hint(self, repo: pathlib.Path) -> None:
1640 result = runner.invoke(cli, ["remote", "set-url", "ghost", "https://hub.muse.io/r"])
1641 assert result.exit_code != 0
1642 assert "muse remote add" in result.output
1643
1644 def test_set_url_invalid_name_rejected_before_repo_check(
1645 self, repo: pathlib.Path
1646 ) -> None:
1647 result = runner.invoke(cli, ["remote", "set-url", "bad name", "https://hub.muse.io/r"])
1648 assert result.exit_code != 0
1649 assert "Invalid remote name" in result.output
1650
1651 def test_set_url_empty_name_rejected(self, repo: pathlib.Path) -> None:
1652 result = runner.invoke(cli, ["remote", "set-url", "", "https://hub.muse.io/r"])
1653 assert result.exit_code != 0
1654
1655 def test_set_url_name_with_slash_rejected(self, repo: pathlib.Path) -> None:
1656 result = runner.invoke(cli, ["remote", "set-url", "or/igin", "https://hub.muse.io/r"])
1657 assert result.exit_code != 0
1658 assert "Invalid remote name" in result.output
1659
1660 def test_set_url_name_too_long_rejected(self, repo: pathlib.Path) -> None:
1661 long_name = "a" * 101
1662 result = runner.invoke(cli, ["remote", "set-url", long_name, "https://hub.muse.io/r"])
1663 assert result.exit_code != 0
1664 assert "too long" in result.output
1665
1666 def test_set_url_url_stripped_of_whitespace(self, repo: pathlib.Path) -> None:
1667 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1668 result = runner.invoke(
1669 cli, ["remote", "set-url", "origin", " https://hub.muse.io/new "]
1670 )
1671 assert result.exit_code == 0
1672 assert get_remote("origin", repo) == "https://hub.muse.io/new"
1673
1674 def test_set_url_url_too_long_rejected(self, repo: pathlib.Path) -> None:
1675 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1676 long_url = "https://hub.muse.io/" + "x" * 2050
1677 result = runner.invoke(cli, ["remote", "set-url", "origin", long_url])
1678 assert result.exit_code != 0
1679 assert "too long" in result.output
1680
1681 def test_set_url_url_too_long_does_not_write(self, repo: pathlib.Path) -> None:
1682 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/original"])
1683 long_url = "https://hub.muse.io/" + "x" * 2050
1684 runner.invoke(cli, ["remote", "set-url", "origin", long_url])
1685 assert get_remote("origin", repo) == "https://hub.muse.io/original"
1686
1687 def test_set_url_scheme_validated_before_repo(
1688 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1689 ) -> None:
1690 """Invalid scheme must be caught before require_repo() is called."""
1691 monkeypatch.chdir(tmp_path)
1692 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1693 result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/passwd"])
1694 assert result.exit_code != 0
1695
1696 def test_set_url_http_url_accepted(self, repo: pathlib.Path) -> None:
1697 """http:// is allowed (useful for local hub instances)."""
1698 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1699 result = runner.invoke(
1700 cli, ["remote", "set-url", "origin", "http://localhost:10003/gabriel/r"]
1701 )
1702 assert result.exit_code == 0
1703 assert get_remote("origin", repo) == "http://localhost:10003/gabriel/r"
1704
1705 def test_set_url_multiple_updates_last_wins(self, repo: pathlib.Path) -> None:
1706 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/v1"])
1707 runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/v2"])
1708 runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/v3"])
1709 assert get_remote("origin", repo) == "https://hub.muse.io/v3"
1710
1711 def test_set_url_other_remotes_unaffected(self, repo: pathlib.Path) -> None:
1712 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/o"])
1713 runner.invoke(cli, ["remote", "add", "upstream", "https://hub.muse.io/u"])
1714 runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/o2"])
1715 assert get_remote("upstream", repo) == "https://hub.muse.io/u"
1716
1717 def test_set_url_outside_repo_exits_nonzero(
1718 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1719 ) -> None:
1720 monkeypatch.chdir(tmp_path)
1721 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1722 result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/r"])
1723 assert result.exit_code != 0
1724
1725 def test_set_url_text_output_to_stderr(self, repo: pathlib.Path) -> None:
1726 """In text mode, stdout must be empty — messages go to stderr."""
1727 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1728 result = runner.invoke(cli, ["remote", "set-url", "origin", "https://hub.muse.io/new"])
1729 assert result.exit_code == 0
1730
1731 def test_set_url_json_stdout_parseable(self, repo: pathlib.Path) -> None:
1732 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/old"])
1733 result = runner.invoke(
1734 cli, ["remote", "set-url", "origin", "https://hub.muse.io/new", "--json"]
1735 )
1736 assert result.exit_code == 0
1737 # Must parse without error
1738 json_line = next(
1739 (l for l in result.output.splitlines() if l.strip().startswith("{")), None
1740 )
1741 assert json_line is not None
1742 parsed = json.loads(json_line)
1743 assert parsed["status"] == "ok"
1744
1745 def test_set_url_json_all_required_keys_present(self, repo: pathlib.Path) -> None:
1746 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1747 result = runner.invoke(
1748 cli, ["remote", "set-url", "origin", "https://hub.muse.io/r2", "--json"]
1749 )
1750 data = _json_mutation(result)
1751 for key in ("status", "name", "url", "old_url", "old_name", "new_name"):
1752 assert key in data, f"Missing key '{key}' in JSON output"
1753
1754 def test_set_url_dash_name_valid(self, repo: pathlib.Path) -> None:
1755 runner.invoke(cli, ["remote", "add", "my-remote", "https://hub.muse.io/r"])
1756 result = runner.invoke(
1757 cli, ["remote", "set-url", "my-remote", "https://hub.muse.io/r2"]
1758 )
1759 assert result.exit_code == 0
1760
1761 def test_set_url_underscore_name_valid(self, repo: pathlib.Path) -> None:
1762 runner.invoke(cli, ["remote", "add", "my_remote", "https://hub.muse.io/r"])
1763 result = runner.invoke(
1764 cli, ["remote", "set-url", "my_remote", "https://hub.muse.io/r2"]
1765 )
1766 assert result.exit_code == 0
1767
1768
1769 # ── set-url security ──────────────────────────────────────────────────────────
1770
1771 class TestRemoteSetUrlSecurity:
1772 """Security-focused tests for ``muse remote set-url``."""
1773
1774 def test_file_scheme_rejected(self, repo: pathlib.Path) -> None:
1775 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1776 result = runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/passwd"])
1777 assert result.exit_code != 0
1778
1779 def test_ftp_scheme_rejected(self, repo: pathlib.Path) -> None:
1780 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1781 result = runner.invoke(cli, ["remote", "set-url", "origin", "ftp://hub.muse.io/r"])
1782 assert result.exit_code != 0
1783
1784 def test_data_scheme_rejected(self, repo: pathlib.Path) -> None:
1785 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/r"])
1786 result = runner.invoke(cli, ["remote", "set-url", "origin", "data:text/plain,evil"])
1787 assert result.exit_code != 0
1788
1789 def test_ansi_in_name_sanitized(
1790 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
1791 ) -> None:
1792 evil = "\x1b[31mghost\x1b[0m"
1793 result = runner.invoke(cli, ["remote", "set-url", evil, "https://hub.muse.io/r"])
1794 assert result.exit_code != 0
1795 err = result.stderr or ""
1796 assert "\x1b[" not in err
1797
1798 def test_ansi_in_url_sanitized_in_output(self, repo: pathlib.Path) -> None:
1799 """ANSI in stored URL must not reach terminal as raw ESC bytes.
1800
1801 Write TOML directly using \\u001b unicode escapes — raw \\x1b bytes are
1802 illegal in TOML quoted strings, so we write the escape form which TOML
1803 decodes to the ESC character when loading.
1804 """
1805 config_toml = repo / ".muse" / "config.toml"
1806 # \\u001b in Python str → \u001b written to disk → ESC when TOML loads it
1807 config_toml.write_text(
1808 '[remotes.origin]\nurl = "https://hub.muse.io/\\u001b[31mevil\\u001b[0m"\n'
1809 )
1810 # Now set-url to a clean URL — old_url contains an ESC; JSON must encode it
1811 result = runner.invoke(
1812 cli, ["remote", "set-url", "origin", "https://hub.muse.io/clean", "--json"]
1813 )
1814 assert result.exit_code == 0
1815 # json.dumps always encodes ESC as \\u001b — no raw ESC byte must appear
1816 assert "\x1b" not in result.output
1817
1818 def test_control_char_in_name_rejected(self, repo: pathlib.Path) -> None:
1819 result = runner.invoke(cli, ["remote", "set-url", "ori\x00gin", "https://hub.muse.io/r"])
1820 assert result.exit_code != 0
1821
1822 def test_invalid_scheme_does_not_write(self, repo: pathlib.Path) -> None:
1823 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/original"])
1824 runner.invoke(cli, ["remote", "set-url", "origin", "file:///etc/shadow"])
1825 assert get_remote("origin", repo) == "https://hub.muse.io/original"
1826
1827 def test_invalid_name_does_not_reach_repo(
1828 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1829 ) -> None:
1830 """Name validation must fire before require_repo — no repo needed."""
1831 monkeypatch.chdir(tmp_path)
1832 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
1833 result = runner.invoke(cli, ["remote", "set-url", "bad name", "https://hub.muse.io/r"])
1834 assert result.exit_code != 0
1835 assert "Invalid remote name" in result.output
1836
1837
1838 # ── set-url stress ────────────────────────────────────────────────────────────
1839
1840 class TestRemoteSetUrlStress:
1841 """Volume and concurrency tests for ``muse remote set-url``."""
1842
1843 def test_100_sequential_set_url_updates(self, repo: pathlib.Path) -> None:
1844 """100 sequential updates; final URL must match the last write."""
1845 from muse.cli.config import set_remote, get_remote
1846 runner.invoke(cli, ["remote", "add", "origin", "https://hub.muse.io/v0"])
1847 for i in range(1, 101):
1848 set_remote("origin", f"https://hub.muse.io/v{i}", repo)
1849 assert get_remote("origin", repo) == "https://hub.muse.io/v100"
1850
1851 def test_set_url_across_10_remotes(self, repo: pathlib.Path) -> None:
1852 """Update 10 different remotes; each must store its own URL."""
1853 from muse.cli.config import set_remote, get_remote
1854 for i in range(10):
1855 runner.invoke(cli, ["remote", "add", f"r{i}", f"https://hub.muse.io/old{i}"])
1856 for i in range(10):
1857 set_remote(f"r{i}", f"https://hub.muse.io/new{i}", repo)
1858 for i in range(10):
1859 assert get_remote(f"r{i}", repo) == f"https://hub.muse.io/new{i}"
1860
1861 def test_concurrent_set_url_isolated_repos(
1862 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
1863 ) -> None:
1864 """8 concurrent set-url calls on isolated repos must not interfere."""
1865 import json as _json
1866 errors: list[str] = []
1867 lock = threading.Lock()
1868
1869 def _worker(idx: int) -> None:
1870 try:
1871 from muse.cli.config import set_remote, get_remote
1872 # Build a minimal isolated repo
1873 repo_dir = tmp_path / f"repo{idx}"
1874 muse_dir = repo_dir / ".muse"
1875 (muse_dir / "refs" / "heads").mkdir(parents=True)
1876 (muse_dir / "objects").mkdir()
1877 (muse_dir / "commits").mkdir()
1878 (muse_dir / "snapshots").mkdir()
1879 (muse_dir / "repo.json").write_text(
1880 _json.dumps({"repo_id": f"r{idx}", "schema_version": "0.1", "domain": "midi"})
1881 )
1882 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
1883 set_remote("origin", f"https://hub.muse.io/old{idx}", repo_dir)
1884 set_remote("origin", f"https://hub.muse.io/new{idx}", repo_dir)
1885 val = get_remote("origin", repo_dir)
1886 assert val == f"https://hub.muse.io/new{idx}", f"worker {idx}: got {val!r}"
1887 except Exception as exc:
1888 with lock:
1889 errors.append(f"worker {idx}: {exc}")
1890
1891 threads = [threading.Thread(target=_worker, args=(i,)) for i in range(8)]
1892 for t in threads:
1893 t.start()
1894 for t in threads:
1895 t.join()
1896 assert errors == [], "\n".join(errors)
1897
1898
1899 # ── status extended ───────────────────────────────────────────────────────────
1900
1901 class TestRemoteStatusExtended:
1902 """Extended integration tests for ``muse remote status``."""
1903
1904 def _add(self, repo: pathlib.Path, url: str = "http://localhost:19999/gabriel/repo") -> None:
1905 runner.invoke(cli, ["remote", "add", "origin", url])
1906
1907 def test_status_reachable_exit_zero(self, repo: pathlib.Path) -> None:
1908 self._add(repo)
1909 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1910 result = runner.invoke(cli, ["remote", "status", "origin"])
1911 assert result.exit_code == 0
1912
1913 def test_status_unreachable_exit_nonzero(self, repo: pathlib.Path) -> None:
1914 self._add(repo)
1915 with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "refused")):
1916 result = runner.invoke(cli, ["remote", "status", "origin"])
1917 assert result.exit_code != 0
1918
1919 def test_status_unreachable_exit_code_is_remote_error(self, repo: pathlib.Path) -> None:
1920 """Unreachable remote must exit with REMOTE_ERROR (5), not INTERNAL_ERROR (3)."""
1921 from muse.core.errors import ExitCode
1922 self._add(repo)
1923 with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "refused")):
1924 result = runner.invoke(cli, ["remote", "status", "origin"])
1925 assert result.exit_code == ExitCode.REMOTE_ERROR
1926
1927 def test_status_json_all_keys_present(self, repo: pathlib.Path) -> None:
1928 self._add(repo)
1929 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1930 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1931 assert result.exit_code == 0
1932 data = _json_status(result)
1933 for key in ("remote", "url", "server_root", "reachable", "http_status", "message", "tracked_refs"):
1934 assert key in data, f"Missing key '{key}'"
1935
1936 def test_status_json_short_flag(self, repo: pathlib.Path) -> None:
1937 self._add(repo)
1938 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1939 result = runner.invoke(cli, ["remote", "status", "origin", "-j"])
1940 assert result.exit_code == 0
1941 data = _json_status(result)
1942 assert data["reachable"] is True
1943
1944 def test_status_json_reachable_true(self, repo: pathlib.Path) -> None:
1945 self._add(repo)
1946 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1947 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1948 data = _json_status(result)
1949 assert data["reachable"] is True
1950 assert data["http_status"] == 200
1951
1952 def test_status_json_reachable_false_5xx(self, repo: pathlib.Path) -> None:
1953 self._add(repo)
1954 with patch("muse.cli.commands.remote._ping_url", return_value=(False, 503, "HTTP 503")):
1955 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1956 data = _json_status(result)
1957 assert data["reachable"] is False
1958 assert data["http_status"] == 503
1959
1960 def test_status_json_null_http_status_on_network_error(self, repo: pathlib.Path) -> None:
1961 self._add(repo)
1962 with patch("muse.cli.commands.remote._ping_url", return_value=(False, None, "no route")):
1963 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1964 data = _json_status(result)
1965 assert data["http_status"] is None
1966
1967 def test_status_json_remote_name_correct(self, repo: pathlib.Path) -> None:
1968 self._add(repo)
1969 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1970 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1971 data = _json_status(result)
1972 assert data["remote"] == "origin"
1973
1974 def test_status_json_url_correct(self, repo: pathlib.Path) -> None:
1975 self._add(repo)
1976 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1977 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1978 data = _json_status(result)
1979 assert data["url"] == "http://localhost:19999/gabriel/repo"
1980
1981 def test_status_json_server_root_extracted(self, repo: pathlib.Path) -> None:
1982 self._add(repo, "http://localhost:19999/gabriel/repo")
1983 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1984 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1985 data = _json_status(result)
1986 assert data["server_root"] == "http://localhost:19999"
1987
1988 def test_status_json_tracked_refs_empty_when_no_fetch(self, repo: pathlib.Path) -> None:
1989 self._add(repo)
1990 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
1991 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
1992 data = _json_status(result)
1993 assert data["tracked_refs"] == {}
1994
1995 def test_status_json_tracked_refs_flat(self, repo: pathlib.Path) -> None:
1996 self._add(repo)
1997 refs_dir = repo / ".muse" / "remotes" / "origin"
1998 refs_dir.mkdir(parents=True)
1999 (refs_dir / "main").write_text("a" * 64)
2000 (refs_dir / "dev").write_text("b" * 64)
2001 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2002 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2003 data = _json_status(result)
2004 assert data["tracked_refs"]["main"] == "a" * 8
2005 assert data["tracked_refs"]["dev"] == "b" * 8
2006
2007 def test_status_json_tracked_refs_nested(self, repo: pathlib.Path) -> None:
2008 self._add(repo)
2009 refs_dir = repo / ".muse" / "remotes" / "origin"
2010 (refs_dir / "feat").mkdir(parents=True)
2011 (refs_dir / "main").write_text("c" * 64)
2012 (refs_dir / "feat" / "ui").write_text("d" * 64)
2013 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2014 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2015 data = _json_status(result)
2016 assert "main" in data["tracked_refs"]
2017 assert "feat/ui" in data["tracked_refs"]
2018
2019 def test_status_missing_remote_exits_nonzero(self, repo: pathlib.Path) -> None:
2020 result = runner.invoke(cli, ["remote", "status", "ghost"])
2021 assert result.exit_code != 0
2022
2023 def test_status_missing_remote_exit_code_user_error(self, repo: pathlib.Path) -> None:
2024 from muse.core.errors import ExitCode
2025 result = runner.invoke(cli, ["remote", "status", "ghost"])
2026 assert result.exit_code == ExitCode.USER_ERROR
2027
2028 def test_status_invalid_name_rejected_before_repo(
2029 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
2030 ) -> None:
2031 """Name validation must fire before require_repo — no repo needed."""
2032 monkeypatch.chdir(tmp_path)
2033 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
2034 result = runner.invoke(cli, ["remote", "status", "bad name"])
2035 assert result.exit_code != 0
2036 assert "Invalid remote name" in result.output
2037
2038 def test_status_outside_repo_exits_nonzero(
2039 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
2040 ) -> None:
2041 monkeypatch.chdir(tmp_path)
2042 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
2043 result = runner.invoke(cli, ["remote", "status", "origin"])
2044 assert result.exit_code != 0
2045
2046 def test_status_custom_timeout_passed_to_ping(self, repo: pathlib.Path) -> None:
2047 self._add(repo)
2048 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")) as m:
2049 runner.invoke(cli, ["remote", "status", "origin", "--timeout", "2.5"])
2050 assert m.call_args[0][1] == 2.5
2051
2052 def test_status_json_parseable_stdout(self, repo: pathlib.Path) -> None:
2053 self._add(repo)
2054 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2055 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2056 assert result.exit_code == 0
2057 json_line = next(
2058 (l for l in result.output.splitlines() if l.strip().startswith("{")), None
2059 )
2060 assert json_line is not None
2061 json.loads(json_line) # must not raise
2062
2063
2064 # ── status security ───────────────────────────────────────────────────────────
2065
2066 class TestRemoteStatusSecurity:
2067 """Security-focused tests for ``muse remote status``."""
2068
2069 def test_invalid_name_no_repo_needed(
2070 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
2071 ) -> None:
2072 monkeypatch.chdir(tmp_path)
2073 monkeypatch.delenv("MUSE_REPO_ROOT", raising=False)
2074 result = runner.invoke(cli, ["remote", "status", "bad/name"])
2075 assert result.exit_code != 0
2076 assert "Invalid remote name" in result.output
2077
2078 def test_ansi_in_name_sanitized(
2079 self, repo: pathlib.Path, capsys: pytest.CaptureFixture[str]
2080 ) -> None:
2081 evil = "\x1b[31morigin\x1b[0m"
2082 result = runner.invoke(cli, ["remote", "status", evil])
2083 assert result.exit_code != 0
2084 err = result.stderr or ""
2085 assert "\x1b[" not in err
2086
2087 def test_control_char_in_name_rejected(self, repo: pathlib.Path) -> None:
2088 result = runner.invoke(cli, ["remote", "status", "ori\x00gin"])
2089 assert result.exit_code != 0
2090 assert "Invalid remote name" in result.output
2091
2092 def test_name_too_long_rejected(self, repo: pathlib.Path) -> None:
2093 long_name = "a" * 101
2094 result = runner.invoke(cli, ["remote", "status", long_name])
2095 assert result.exit_code != 0
2096 assert "too long" in result.output
2097
2098 def test_file_scheme_url_returns_unreachable(self, repo: pathlib.Path) -> None:
2099 """A file:// URL in config must be handled by the SSRF guard in _ping_url."""
2100 from muse.cli.config import set_remote
2101 set_remote("badremote", "file:///etc/passwd", repo)
2102 result = runner.invoke(cli, ["remote", "status", "badremote", "--json"])
2103 assert result.exit_code != 0
2104 data = _json_status(result)
2105 assert data["reachable"] is False
2106
2107 def test_ansi_in_stored_url_sanitized_in_text_output(self, repo: pathlib.Path) -> None:
2108 """ANSI codes in a stored URL must not leak as raw ESC bytes in text mode."""
2109 config_toml = repo / ".muse" / "config.toml"
2110 config_toml.write_text(
2111 '[remotes.origin]\nurl = "http://localhost/\\u001b[31mevil\\u001b[0m"\n'
2112 )
2113 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2114 result = runner.invoke(cli, ["remote", "status", "origin"])
2115 assert result.exit_code == 0
2116 assert "\x1b" not in result.output
2117
2118 def test_ansi_in_stored_url_escaped_in_json(self, repo: pathlib.Path) -> None:
2119 """ANSI in stored URL must be JSON-encoded, not emitted as raw ESC."""
2120 config_toml = repo / ".muse" / "config.toml"
2121 config_toml.write_text(
2122 '[remotes.origin]\nurl = "http://localhost/\\u001b[31mevil\\u001b[0m"\n'
2123 )
2124 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2125 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2126 assert result.exit_code == 0
2127 assert "\x1b" not in result.output
2128
2129 def test_symlink_in_remotes_dir_skipped(self, repo: pathlib.Path) -> None:
2130 """Symlinks inside the remotes tracking dir must be skipped."""
2131 runner.invoke(cli, ["remote", "add", "origin", "http://localhost:19999/gabriel/repo"])
2132 refs_dir = repo / ".muse" / "remotes" / "origin"
2133 refs_dir.mkdir(parents=True)
2134 (refs_dir / "main").write_text("a" * 64)
2135 symlink = refs_dir / "evil"
2136 symlink.symlink_to("/etc/passwd")
2137 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2138 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2139 assert result.exit_code == 0
2140 data = _json_status(result)
2141 assert "evil" not in data["tracked_refs"]
2142 assert "main" in data["tracked_refs"]
2143
2144
2145 # ── status stress ─────────────────────────────────────────────────────────────
2146
2147 class TestRemoteStatusStress:
2148 """Volume and concurrency tests for ``muse remote status``."""
2149
2150 def _add(self, repo: pathlib.Path, url: str = "http://localhost:19999/g/r") -> None:
2151 runner.invoke(cli, ["remote", "add", "origin", url])
2152
2153 def test_50_sequential_status_calls_stable(self, repo: pathlib.Path) -> None:
2154 """50 status calls with a mocked reachable remote must all return exit 0."""
2155 self._add(repo)
2156 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2157 for _ in range(50):
2158 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2159 assert result.exit_code == 0
2160
2161 def test_status_with_100_tracked_refs(self, repo: pathlib.Path) -> None:
2162 """100 tracking refs must all appear in JSON output."""
2163 self._add(repo)
2164 refs_dir = repo / ".muse" / "remotes" / "origin"
2165 refs_dir.mkdir(parents=True)
2166 for i in range(100):
2167 (refs_dir / f"branch{i:03d}").write_text("a" * 64)
2168 with patch("muse.cli.commands.remote._ping_url", return_value=(True, 200, "HTTP 200 OK")):
2169 result = runner.invoke(cli, ["remote", "status", "origin", "--json"])
2170 assert result.exit_code == 0
2171 data = _json_status(result)
2172 assert len(data["tracked_refs"]) == 100
2173
2174 def test_concurrent_status_reads_same_repo(self, repo: pathlib.Path) -> None:
2175 """8 concurrent status reads against the same repo must all succeed."""
2176 self._add(repo)
2177 refs_dir = repo / ".muse" / "remotes" / "origin"
2178 refs_dir.mkdir(parents=True)
2179 (refs_dir / "main").write_text("a" * 64)
2180 results: list[int] = []
2181 errors: list[str] = []
2182 lock = threading.Lock()
2183
2184 def _read() -> None:
2185 try:
2186 from muse.cli.config import get_remote
2187 from muse.cli.commands.remote import _collect_tracked_refs
2188 get_remote("origin", repo)
2189 refs = _collect_tracked_refs(refs_dir)
2190 with lock:
2191 results.append(len(refs))
2192 except Exception as exc:
2193 with lock:
2194 errors.append(str(exc))
2195
2196 threads = [threading.Thread(target=_read) for _ in range(8)]
2197 for t in threads:
2198 t.start()
2199 for t in threads:
2200 t.join()
2201 assert errors == [], "\n".join(errors)
2202 assert all(r == 1 for r in results)
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