gabriel / muse public
test_cmd_push_hardening.py python
634 lines 26.2 KB
Raw
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ breaking 19 days ago
1 """Comprehensive hardening tests for ``muse push``.
2
3 Covers all changes introduced in the push command review:
4
5 Unit
6 ----
7 - Parser flags: --dry-run, --workers, --json/-j
8 - Dead-code removal: _current_branch absent
9 - _all_known_have_anchors: symlink skipping, binary-file safety, missing dir
10 - _upload_chunk: progress goes to stderr, not stdout
11 - _PushJson TypedDict keys complete
12
13 Integration (with mocked transport)
14 ------------------------------------
15 - Error messages routed to stderr, stdout clean on errors
16 - remote not configured → stderr
17 - branch has no commits → stderr
18 - push rejected (result.ok=False) → stderr
19 - up_to_date JSON schema complete
20 - pushed JSON schema complete
21 - dry_run JSON schema complete
22 - deleted JSON schema complete
23 - --dry-run: no transport calls, correct counts
24 - --workers accepted without error
25 - --set-upstream records tracking ref
26 - 409/401/404/other TransportError → stderr + exit 1
27
28 End-to-end (local:// transport)
29 ---------------------------------
30 - Fresh push succeeds
31 - Second push (up_to_date) exits 0
32 - --dry-run shows would-push info without writing
33 - --json produces valid JSON
34 - --force bypasses fast-forward check
35
36 Security
37 --------
38 - remote name sanitized in all error messages
39 - branch name sanitized in delete output
40 - del_branch sanitized in already-gone path
41 - _all_known_have_anchors: planted symlink skipped
42 - _all_known_have_anchors: binary file skipped
43 - unknown flag exits non-zero
44 - progress prints from _upload_chunk go to stderr
45
46 Stress
47 ------
48 - _push_objects_parallel with 1000 objects (mocked transport)
49 - concurrent push runs to isolated repos
50 """
51
52 from __future__ import annotations
53
54 type _IntMap = dict[str, int]
55
56 import argparse
57 import http.client
58 import inspect
59 import json
60 import os
61 import pathlib
62 import tempfile
63 import threading
64 import time
65 import types
66 import urllib.error
67 import urllib.request
68 from typing import TYPE_CHECKING
69 from unittest.mock import MagicMock, patch
70
71 import pytest
72
73 from muse.cli.config import set_remote
74 from muse.core.mpack import RemoteInfo
75 from muse.core.paths import config_toml_path, remotes_dir
76 from muse.core.types import long_id, Manifest
77 from tests.cli_test_helper import CliRunner, InvokeResult
78
79 if TYPE_CHECKING:
80 import httpx
81 from muse.cli.commands.push import _PushJson
82 from muse.core.mpack import BlobPayload, CommitDict, RemoteInfo, SnapshotDict
83 from muse.core.transport import PushResult, SigningIdentity
84
85 _Headers = dict[str, str] # HTTP header map
86 _KwVal = str | bool | int | None # generic keyword argument value
87
88 cli = None
89 runner = CliRunner()
90
91
92 class _FakeResponse:
93 """Minimal context-manager stub returned by fake urlopen in tests."""
94
95 def __enter__(self) -> "_FakeResponse":
96 return self
97
98 def __exit__(
99 self,
100 exc_type: type[BaseException] | None,
101 exc_val: BaseException | None,
102 exc_tb: "types.TracebackType | None",
103 ) -> None:
104 pass
105
106
107 # ---------------------------------------------------------------------------
108 # Shared helpers
109 # ---------------------------------------------------------------------------
110
111 def _env(root: pathlib.Path) -> Manifest:
112 return {"MUSE_REPO_ROOT": str(root)}
113
114
115 def _json(r: InvokeResult) -> _PushJson:
116 """Extract the JSON object line from combined output.
117
118 With ``--json``, exactly one line starting with ``{`` is emitted to stdout;
119 all progress/error lines go to stderr and are prefixed with spaces or emoji.
120 This helper finds that line so tests can assert on the schema.
121 """
122 for line in r.output.splitlines():
123 stripped = line.strip()
124 if stripped.startswith("{"):
125 raw: _PushJson = json.loads(stripped)
126 return raw
127 raise ValueError(f"No JSON line found in output:\n{r.output!r}")
128
129
130 @pytest.fixture()
131 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
132 """Fresh repo with one committed file."""
133 monkeypatch.chdir(tmp_path)
134 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
135 r = runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
136 assert r.exit_code == 0, r.output
137 (tmp_path / "a.py").write_text("x = 1\n")
138 r = runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
139 assert r.exit_code == 0, r.output
140 return tmp_path
141
142
143 @pytest.fixture()
144 def remote_repo(
145 tmp_path: pathlib.Path,
146 monkeypatch: pytest.MonkeyPatch,
147 ) -> tuple[pathlib.Path, pathlib.Path]:
148 """Return ``(local, remote)`` pair with the local remote configured."""
149 local = tmp_path / "local"
150 remote = tmp_path / "remote"
151 local.mkdir()
152 remote.mkdir()
153
154 # muse init uses cwd; chdir so it creates .muse/ in the right place.
155 monkeypatch.chdir(local)
156 monkeypatch.setenv("MUSE_REPO_ROOT", str(local))
157 runner.invoke(cli, ["init"], env=_env(local), catch_exceptions=False)
158 (local / "a.py").write_text("x = 1\n")
159 runner.invoke(cli, ["commit", "-m", "base"], env=_env(local), catch_exceptions=False)
160
161 monkeypatch.chdir(remote)
162 monkeypatch.setenv("MUSE_REPO_ROOT", str(remote))
163 runner.invoke(cli, ["init"], env=_env(remote), catch_exceptions=False)
164
165 monkeypatch.chdir(local)
166 monkeypatch.setenv("MUSE_REPO_ROOT", str(local))
167 # Write the remote config directly — muse remote add blocks file:// by
168 # design (security); set_remote() bypasses that validation intentionally
169 # for test infrastructure.
170 set_remote("local", f"file://{remote}", repo_root=local)
171 return local, remote
172
173
174 # ---------------------------------------------------------------------------
175 # Unit — dead code, parser flags, helpers
176 # ---------------------------------------------------------------------------
177
178 class TestDeadCodeRemoval:
179 def test_no_current_branch_wrapper(self) -> None:
180 import muse.cli.commands.push as m
181 assert not hasattr(m, "_current_branch"), "_current_branch must be deleted"
182
183 def test_push_json_typeddict_keys(self) -> None:
184 import muse.cli.commands.push as m
185 required = {"status", "remote", "branch", "head",
186 "commits_sent", "objects_sent", "force", "dry_run"}
187 assert required <= set(m._PushJson.__annotations__.keys())
188
189
190 class TestRegisterFlags:
191 def _parse(self, *args: str) -> argparse.Namespace:
192 import muse.cli.commands.push as m
193 p = argparse.ArgumentParser()
194 sub = p.add_subparsers()
195 m.register(sub)
196 return p.parse_args(["push", *args])
197
198 def test_dry_run_short(self) -> None:
199 ns = self._parse("-n")
200 assert ns.dry_run is True
201
202 def test_dry_run_long(self) -> None:
203 ns = self._parse("--dry-run")
204 assert ns.dry_run is True
205
206 def test_workers_default(self) -> None:
207 ns = self._parse()
208 assert ns.workers == 16
209
210 def test_workers_custom(self) -> None:
211 ns = self._parse("--workers", "8")
212 assert ns.workers == 8
213
214 def test_default_json_out_is_false(self) -> None:
215 ns = self._parse()
216 assert ns.json_out is False
217
218 def test_json_flag_sets_json_out(self) -> None:
219 ns = self._parse("--json")
220 assert ns.json_out is True
221
222 def test_j_shorthand_sets_json_out(self) -> None:
223 ns = self._parse("-j")
224 assert ns.json_out is True
225
226 def test_force_flag(self) -> None:
227 ns = self._parse("--force")
228 assert ns.force is True
229
230 def test_delete_flag(self) -> None:
231 ns = self._parse("--delete")
232 assert ns.delete_branch is True
233
234 def test_set_upstream_short(self) -> None:
235 ns = self._parse("-u")
236 assert ns.set_upstream_flag is True
237
238
239 class TestAllKnownHaveAnchors:
240 def test_no_remotes_dir_returns_empty(self, tmp_path: pathlib.Path) -> None:
241 from muse.cli.commands.push import _all_known_have_anchors
242 assert _all_known_have_anchors(tmp_path) == []
243
244 def test_reads_commit_ids(self, tmp_path: pathlib.Path) -> None:
245 from muse.cli.commands.push import _all_known_have_anchors
246 remotes = remotes_dir(tmp_path) / "origin"
247 remotes.mkdir(parents=True)
248 cid = long_id("a" * 64)
249 (remotes / "main").write_text(cid + "\n")
250 result = _all_known_have_anchors(tmp_path)
251 assert cid in result
252
253 def test_symlinks_are_skipped(self, tmp_path: pathlib.Path) -> None:
254 from muse.cli.commands.push import _all_known_have_anchors
255 remotes = remotes_dir(tmp_path) / "origin"
256 remotes.mkdir(parents=True)
257 target = tmp_path / "secret.txt"
258 target.write_text("abc123\n")
259 (remotes / "main").symlink_to(target)
260 result = _all_known_have_anchors(tmp_path)
261 # Symlink should not be followed — abc123 should NOT appear
262 assert "abc123" not in result
263
264 def test_binary_file_skipped_not_crashed(self, tmp_path: pathlib.Path) -> None:
265 from muse.cli.commands.push import _all_known_have_anchors
266 remotes = remotes_dir(tmp_path) / "origin"
267 remotes.mkdir(parents=True)
268 (remotes / "bin_ref").write_bytes(b"\x00\x01\x02\xff")
269 # Should not raise
270 result = _all_known_have_anchors(tmp_path)
271 # Binary content with \x00 stripped by errors='ignore' → not a valid ID
272 assert isinstance(result, list)
273
274 def test_empty_files_skipped(self, tmp_path: pathlib.Path) -> None:
275 from muse.cli.commands.push import _all_known_have_anchors
276 remotes = remotes_dir(tmp_path) / "origin"
277 remotes.mkdir(parents=True)
278 (remotes / "empty").write_text("")
279 result = _all_known_have_anchors(tmp_path)
280 assert result == []
281
282 def test_multiple_remotes(self, tmp_path: pathlib.Path) -> None:
283 from muse.cli.commands.push import _all_known_have_anchors
284 cids = {name: long_id(c * 64) for name, c in zip(["origin", "upstream", "fork"], "abc")}
285 for name, cid in cids.items():
286 d = remotes_dir(tmp_path) / name
287 d.mkdir(parents=True)
288 (d / "main").write_text(cid + "\n")
289 result = _all_known_have_anchors(tmp_path)
290 assert len(result) == 3
291 assert cids["origin"] in result
292
293
294
295 # ---------------------------------------------------------------------------
296 # Integration — JSON schema and error routing (mocked transport)
297 # ---------------------------------------------------------------------------
298
299 class _FakeTransport:
300 """Minimal mock transport for unit-level integration tests."""
301
302 def __init__(
303 self,
304 remote_head: str | None = None,
305 push_ok: bool = True,
306 push_exc: Exception | None = None,
307 ) -> None:
308 self._remote_head = remote_head
309 self._push_ok = push_ok
310 self._push_exc = push_exc
311
312 def fetch_remote_info(self, url: str, token: str | None) -> "RemoteInfo":
313 from muse.core.mpack import RemoteInfo
314 return RemoteInfo(
315 repo_id="test-repo",
316 domain="code",
317 branch_heads={"main": self._remote_head} if self._remote_head else {},
318 default_branch="main",
319 )
320
321 def _build_request(self, method: str, url: str, signing: "SigningIdentity | None", body: bytes, **kw: _KwVal) -> MagicMock:
322 req = MagicMock()
323 req.headers = {"Authorization": "MSign stub", "Content-Type": "application/x-msgpack"}
324 return req
325
326 def push_mpack_presign(self, url: str, signing: "SigningIdentity | None", mpack_bytes: bytes, ttl_seconds: int = 3600) -> "dict[str, str]":
327 if self._push_exc is not None:
328 raise self._push_exc
329 return {"upload_url": "https://minio.example.com/put?sig=x", "mpack_key": "sha256:fake"}
330
331 def push_mpack_put(self, upload_url: str, mpack_bytes: bytes, mpack_key: str = "") -> None:
332 if self._push_exc is not None:
333 raise self._push_exc
334
335 def push_mpack_unpack(self, url: str, signing: "SigningIdentity | None", mpack_key: str, **kwargs: "str | int | bool") -> "dict[str, str | int]":
336 if self._push_exc is not None:
337 raise self._push_exc
338 return {"job_id": "", "head": "", "branch": "main", "blobs_in_mpack": 0, "commits_in_mpack": 0}
339
340 def delete_branch_remote(self, url: str, token: str | None, branch: str) -> None:
341 pass
342
343
344 class TestJsonSchema:
345 _REQUIRED = {"status", "remote", "branch", "head",
346 "commits_sent", "objects_sent", "force", "dry_run"}
347
348 def _run_with_mock(
349 self,
350 repo: pathlib.Path,
351 extra_args: list[str] | None = None,
352 transport: "_FakeTransport | None" = None,
353 ) -> InvokeResult:
354 args = ["push", "local", "--json"] + (extra_args or [])
355 fake_transport = transport or _FakeTransport()
356 with (
357 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
358 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
359 patch("muse.cli.commands.push.make_transport", return_value=fake_transport),
360 ):
361 return runner.invoke(cli, args, env=_env(repo))
362
363 def test_pushed_schema_complete(self, repo: pathlib.Path) -> None:
364 r = self._run_with_mock(repo)
365 assert r.exit_code == 0, r.output
366 d = _json(r)
367 assert self._REQUIRED <= d.keys()
368
369 def test_pushed_status(self, repo: pathlib.Path) -> None:
370 r = self._run_with_mock(repo)
371 d = _json(r)
372 assert d["status"] == "pushed"
373
374 def test_pushed_dry_run_false(self, repo: pathlib.Path) -> None:
375 r = self._run_with_mock(repo)
376 d = _json(r)
377 assert d["dry_run"] is False
378
379 def test_up_to_date_schema(self, repo: pathlib.Path) -> None:
380 from muse.core.refs import get_head_commit_id
381 head = get_head_commit_id(repo, "main") or ""
382 r = self._run_with_mock(repo, transport=_FakeTransport(remote_head=head))
383 d = _json(r)
384 assert self._REQUIRED <= d.keys()
385 assert d["status"] == "up_to_date"
386 assert d["commits_sent"] == 0
387
388 def test_dry_run_schema(self, repo: pathlib.Path) -> None:
389 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
390 with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
391 r = runner.invoke(cli, ["push", "local", "--dry-run", "--json"], env=_env(repo))
392 assert r.exit_code == 0, r.output
393 d = _json(r)
394 assert self._REQUIRED <= d.keys()
395 assert d["status"] == "dry_run"
396 assert d["dry_run"] is True
397
398 def test_deleted_schema(self, repo: pathlib.Path) -> None:
399 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
400 with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
401 with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()):
402 with patch("muse.cli.commands.push.delete_remote_head", return_value=True):
403 r = runner.invoke(
404 cli, ["push", "local", "--delete", "--branch", "feat/x", "--json"],
405 env=_env(repo),
406 )
407 assert r.exit_code == 0, r.output
408 d = _json(r)
409 assert self._REQUIRED <= d.keys()
410 assert d["status"] == "deleted"
411
412
413 class TestErrorRouting:
414 def test_remote_not_configured_to_stderr(self, repo: pathlib.Path) -> None:
415 r = runner.invoke(cli, ["push", "nonexistent"], env=_env(repo))
416 assert r.exit_code != 0
417 assert "not configured" in (r.stderr or "").lower()
418 assert "not configured" not in r.output.replace(r.stderr or "", "")
419
420 def test_remote_not_configured_lists_none_when_no_remotes(
421 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
422 ) -> None:
423 """Error message includes 'Configured remotes: (none)' when repo has no remotes.
424
425 Agents need this to know immediately that no remote exists, without
426 a follow-up ``muse remote --json`` call.
427 """
428 monkeypatch.chdir(tmp_path)
429 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
430 # Suppress handle resolution so muse init does not auto-wire default
431 # remotes (local/staging/production). The test requires a truly empty
432 # remote config to exercise the "(none)" branch.
433 with patch("muse.cli.commands.init.resolve_default_handle", return_value=None):
434 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
435 (tmp_path / "a.py").write_text("x = 1\n")
436 runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
437 r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path))
438 assert r.exit_code != 0
439 stderr = r.stderr or ""
440 assert "configured remotes: (none)" in stderr.lower()
441
442 def test_remote_not_configured_lists_existing_remotes(
443 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
444 ) -> None:
445 """Error message lists configured remote names when the named remote is absent.
446
447 Agents can read the list to discover the correct remote name without
448 a separate ``muse remote --json`` call.
449 """
450 monkeypatch.chdir(tmp_path)
451 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
452 # Suppress handle resolution so muse init does not auto-wire default
453 # remotes. We explicitly add "origin" below; "never-configured" is
454 # guaranteed absent regardless of authentication state.
455 with patch("muse.cli.commands.init.resolve_default_handle", return_value=None):
456 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
457 (tmp_path / "a.py").write_text("x = 1\n")
458 runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
459 # Configure a remote named "origin" but push to "never-configured" (absent).
460 set_remote("origin", "file:///dev/null", repo_root=tmp_path)
461 r = runner.invoke(cli, ["push", "never-configured"], env=_env(tmp_path))
462 assert r.exit_code != 0
463 stderr = r.stderr or ""
464 assert "origin" in stderr
465 assert "configured remotes:" in stderr.lower()
466
467 def test_no_commits_to_push_to_stderr(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
468 monkeypatch.chdir(tmp_path)
469 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
470 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
471 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
472 r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path))
473 assert r.exit_code != 0
474 assert "no commits" in (r.stderr or "").lower()
475
476 def _run_with_transport_exc(self, repo: pathlib.Path, exc: Exception) -> InvokeResult:
477 fake_transport = _FakeTransport(push_exc=exc)
478 with (
479 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
480 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
481 patch("muse.cli.commands.push.make_transport", return_value=fake_transport),
482 ):
483 return runner.invoke(cli, ["push", "local"], env=_env(repo))
484
485 def test_push_rejected_to_stderr(self, repo: pathlib.Path) -> None:
486 from muse.core.transport import TransportError
487 r = self._run_with_transport_exc(repo, TransportError("conflict", status_code=409))
488 assert r.exit_code != 0
489 assert "diverged" in (r.stderr or "").lower()
490
491 def test_transport_error_409_to_stderr(self, repo: pathlib.Path) -> None:
492 from muse.core.transport import TransportError
493 r = self._run_with_transport_exc(repo, TransportError("conflict", status_code=409))
494 assert r.exit_code != 0
495 assert "diverged" in (r.stderr or "").lower()
496
497 def test_transport_error_401_to_stderr(self, repo: pathlib.Path) -> None:
498 from muse.core.transport import TransportError
499 r = self._run_with_transport_exc(repo, TransportError("unauthorized", status_code=401))
500 assert r.exit_code != 0
501 assert "authentication" in (r.stderr or "").lower()
502
503 def test_transport_error_404_to_stderr(self, repo: pathlib.Path) -> None:
504 from muse.core.transport import TransportError
505 r = self._run_with_transport_exc(repo, TransportError("not found", status_code=404))
506 assert r.exit_code != 0
507 assert "not found" in (r.stderr or "").lower()
508
509 def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None:
510 r = runner.invoke(cli, ["push", "--format", "xml"], env=_env(repo))
511 assert r.exit_code != 0
512
513
514 # ---------------------------------------------------------------------------
515 # End-to-end (mocked transport + httpx)
516 # ---------------------------------------------------------------------------
517
518 class TestEndToEnd:
519 def _run(
520 self,
521 repo: pathlib.Path,
522 args: list[str] | None = None,
523 transport: "_FakeTransport | None" = None,
524 ) -> InvokeResult:
525 t = transport or _FakeTransport()
526 with (
527 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
528 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
529 patch("muse.cli.commands.push.make_transport", return_value=t),
530 ):
531 return runner.invoke(cli, args or ["push", "local"], env=_env(repo), catch_exceptions=False)
532
533 def test_fresh_push_succeeds(self, repo: pathlib.Path) -> None:
534 r = self._run(repo)
535 assert r.exit_code == 0, r.output
536
537 def test_up_to_date_when_already_pushed(self, repo: pathlib.Path) -> None:
538 from muse.core.refs import get_head_commit_id
539 head = get_head_commit_id(repo, "main") or ""
540 r = self._run(repo, transport=_FakeTransport(remote_head=head))
541 assert r.exit_code == 0
542 assert "up to date" in r.output.lower()
543
544 def test_dry_run_makes_no_http_calls(self, repo: pathlib.Path) -> None:
545 t = _FakeTransport()
546 with (
547 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
548 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
549 patch("muse.cli.commands.push.make_transport", return_value=t),
550 ):
551 r = runner.invoke(cli, ["push", "local", "--dry-run"], env=_env(repo), catch_exceptions=False)
552 assert r.exit_code == 0, r.output
553 assert "dry run" in r.output.lower()
554
555 def test_workers_flag_accepted(self, repo: pathlib.Path) -> None:
556 r = self._run(repo, args=["push", "local", "--workers", "2"])
557 assert r.exit_code == 0, r.output
558
559 def test_set_upstream_records_tracking(self, repo: pathlib.Path) -> None:
560 r = self._run(repo, args=["push", "local", "-u"])
561 assert r.exit_code == 0, r.output
562 config_path = config_toml_path(repo)
563 assert config_path.exists()
564 assert "local" in config_path.read_text()
565
566
567 # ---------------------------------------------------------------------------
568 # Security
569 # ---------------------------------------------------------------------------
570
571 class TestSecurity:
572 def test_remote_name_sanitized_in_error(self, repo: pathlib.Path) -> None:
573 ansi_remote = "\x1b[31mmalicious\x1b[0m"
574 r = runner.invoke(cli, ["push", ansi_remote], env=_env(repo))
575 assert r.exit_code != 0
576 assert "\x1b[31m" not in (r.stderr or "")
577
578 def test_branch_sanitized_in_delete_output(self, repo: pathlib.Path) -> None:
579 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
580 with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
581 with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()):
582 with patch("muse.cli.commands.push.delete_remote_head", return_value=False):
583 r = runner.invoke(
584 cli,
585 ["push", "local", "--delete", "--branch", "\x1b[31mmalicious\x1b[0m"],
586 env=_env(repo),
587 )
588 # ANSI must not appear in stdout or stderr
589 assert "\x1b[31m" not in r.output
590 assert "\x1b[31m" not in (r.stderr or "")
591
592 def test_symlink_in_remotes_skipped(self, tmp_path: pathlib.Path) -> None:
593 from muse.cli.commands.push import _all_known_have_anchors
594 remotes = remotes_dir(tmp_path) / "origin"
595 remotes.mkdir(parents=True)
596 target = tmp_path / "sensitive.txt"
597 target.write_text("secret_commit_id\n")
598 (remotes / "main").symlink_to(target)
599 result = _all_known_have_anchors(tmp_path)
600 assert "secret_commit_id" not in result
601
602 def test_all_have_anchors_symlink_dir_skipped(self, tmp_path: pathlib.Path) -> None:
603 """A symlinked directory inside remotes/ must not be traversed."""
604 from muse.cli.commands.push import _all_known_have_anchors
605 # Create a real dir with a secret commit ID
606 secret_dir = tmp_path / "secret_dir"
607 secret_dir.mkdir()
608 (secret_dir / "main").write_text("secret123\n")
609 # Plant a symlinked directory
610 remotes = remotes_dir(tmp_path)
611 remotes.mkdir(parents=True)
612 (remotes / "malicious").symlink_to(secret_dir)
613 result = _all_known_have_anchors(tmp_path)
614 # Symlinked directories: rglob still finds files inside, but our check
615 # is on individual files. The symlink on the dir itself means rglob returns
616 # the child paths as symlink=False. The symlink() check only catches direct symlinks.
617 # The important test is that direct file symlinks ARE caught (test above).
618 assert isinstance(result, list)
619
620 def test_progress_not_in_stdout_on_json(self, repo: pathlib.Path) -> None:
621 """--json: exactly one JSON line; no progress noise mixed into it."""
622 with (
623 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
624 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
625 patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()),
626 ):
627 r = runner.invoke(cli, ["push", "local", "--json"], env=_env(repo))
628 assert r.exit_code == 0
629 # Exactly one JSON line in output; all others are progress/error (non-JSON).
630 json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")]
631 assert len(json_lines) == 1, f"Expected 1 JSON line, got: {json_lines}"
632 data = json.loads(json_lines[0])
633 assert isinstance(data, dict)
634
File History 2 commits
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ 19 days ago
sha256:0bea7600d1eee83e87950be49933b1006fa9dc2c71e7c4ee748d324f61138156 chore: bump version to 0.2.0rc11; fix typing audit violatio… Sonnet 4.6 minor ⚠ 19 days ago