gabriel / muse public
test_cmd_push_hardening.py python
672 lines 27.7 KB
Raw
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ breaking 20 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 _FakeHttpxClient:
300 """Fake httpx AsyncClient for push tests — returns success responses."""
301
302 def __init__(self, push_exc: "Exception | None" = None) -> None:
303 self._push_exc = push_exc
304
305 async def __aenter__(self) -> "_FakeHttpxClient":
306 return self
307
308 async def __aexit__(self, *_: object) -> None:
309 pass
310
311 async def post(self, url: str, *, content: bytes, headers: _Headers) -> MagicMock:
312 import msgpack
313 if self._push_exc is not None:
314 raise self._push_exc
315 r = MagicMock()
316 r.status_code = 200
317 r.headers = {"content-type": "application/x-msgpack"}
318 r.text = ""
319 if "mpack-presign" in url:
320 r.content = msgpack.packb({"upload_url": "https://minio.example.com/put?sig=x"}, use_bin_type=True)
321 else:
322 r.content = msgpack.packb(
323 {"job_id": "job-fake", "head": "", "branch": "main",
324 "objects_in_mpack": 0, "commits_in_mpack": 0},
325 use_bin_type=True,
326 )
327 return r
328
329 async def put(self, url: str, *, content: bytes) -> MagicMock:
330 r = MagicMock()
331 r.status_code = 200
332 r.content = b""
333 return r
334
335
336 class _FakeTransport:
337 """Minimal mock transport for unit-level integration tests."""
338
339 def __init__(
340 self,
341 remote_head: str | None = None,
342 push_ok: bool = True,
343 push_exc: Exception | None = None,
344 ) -> None:
345 self._remote_head = remote_head
346 self._push_ok = push_ok
347 self._push_exc = push_exc
348
349 def fetch_remote_info(self, url: str, token: str | None) -> "RemoteInfo":
350 from muse.core.mpack import RemoteInfo
351 return RemoteInfo(
352 repo_id="test-repo",
353 domain="code",
354 branch_heads={"main": self._remote_head} if self._remote_head else {},
355 default_branch="main",
356 )
357
358 def _build_request(self, method: str, url: str, signing: "SigningIdentity | None", body: bytes, **kw: _KwVal) -> MagicMock:
359 req = MagicMock()
360 req.headers = {"Authorization": "MSign stub", "Content-Type": "application/x-msgpack"}
361 return req
362
363 def delete_branch_remote(self, url: str, token: str | None, branch: str) -> None:
364 pass
365
366
367 class TestJsonSchema:
368 _REQUIRED = {"status", "remote", "branch", "head",
369 "commits_sent", "objects_sent", "force", "dry_run"}
370
371 def _run_with_mock(
372 self,
373 repo: pathlib.Path,
374 extra_args: list[str] | None = None,
375 transport: "_FakeTransport | None" = None,
376 ) -> InvokeResult:
377 args = ["push", "local", "--json"] + (extra_args or [])
378 fake_transport = transport or _FakeTransport()
379 fake_client = _FakeHttpxClient(push_exc=fake_transport._push_exc)
380 with (
381 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
382 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
383 patch("muse.cli.commands.push.make_transport", return_value=fake_transport),
384 patch("muse.cli.commands.push._httpx.AsyncClient", return_value=fake_client),
385 patch("muse.cli.commands.push._make_r2_client", return_value=fake_client),
386 ):
387 return runner.invoke(cli, args, env=_env(repo))
388
389 def test_pushed_schema_complete(self, repo: pathlib.Path) -> None:
390 r = self._run_with_mock(repo)
391 assert r.exit_code == 0, r.output
392 d = _json(r)
393 assert self._REQUIRED <= d.keys()
394
395 def test_pushed_status(self, repo: pathlib.Path) -> None:
396 r = self._run_with_mock(repo)
397 d = _json(r)
398 assert d["status"] == "pushed"
399
400 def test_pushed_dry_run_false(self, repo: pathlib.Path) -> None:
401 r = self._run_with_mock(repo)
402 d = _json(r)
403 assert d["dry_run"] is False
404
405 def test_up_to_date_schema(self, repo: pathlib.Path) -> None:
406 from muse.core.refs import get_head_commit_id
407 head = get_head_commit_id(repo, "main") or ""
408 r = self._run_with_mock(repo, transport=_FakeTransport(remote_head=head))
409 d = _json(r)
410 assert self._REQUIRED <= d.keys()
411 assert d["status"] == "up_to_date"
412 assert d["commits_sent"] == 0
413
414 def test_dry_run_schema(self, repo: pathlib.Path) -> None:
415 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
416 with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
417 r = runner.invoke(cli, ["push", "local", "--dry-run", "--json"], env=_env(repo))
418 assert r.exit_code == 0, r.output
419 d = _json(r)
420 assert self._REQUIRED <= d.keys()
421 assert d["status"] == "dry_run"
422 assert d["dry_run"] is True
423
424 def test_deleted_schema(self, repo: pathlib.Path) -> None:
425 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
426 with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
427 with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()):
428 with patch("muse.cli.commands.push.delete_remote_head", return_value=True):
429 r = runner.invoke(
430 cli, ["push", "local", "--delete", "--branch", "feat/x", "--json"],
431 env=_env(repo),
432 )
433 assert r.exit_code == 0, r.output
434 d = _json(r)
435 assert self._REQUIRED <= d.keys()
436 assert d["status"] == "deleted"
437
438
439 class TestErrorRouting:
440 def test_remote_not_configured_to_stderr(self, repo: pathlib.Path) -> None:
441 r = runner.invoke(cli, ["push", "nonexistent"], env=_env(repo))
442 assert r.exit_code != 0
443 assert "not configured" in (r.stderr or "").lower()
444 assert "not configured" not in r.output.replace(r.stderr or "", "")
445
446 def test_remote_not_configured_lists_none_when_no_remotes(
447 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
448 ) -> None:
449 """Error message includes 'Configured remotes: (none)' when repo has no remotes.
450
451 Agents need this to know immediately that no remote exists, without
452 a follow-up ``muse remote --json`` call.
453 """
454 monkeypatch.chdir(tmp_path)
455 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
456 # Suppress handle resolution so muse init does not auto-wire default
457 # remotes (local/staging/production). The test requires a truly empty
458 # remote config to exercise the "(none)" branch.
459 with patch("muse.cli.commands.init.resolve_default_handle", return_value=None):
460 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
461 (tmp_path / "a.py").write_text("x = 1\n")
462 runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
463 r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path))
464 assert r.exit_code != 0
465 stderr = r.stderr or ""
466 assert "configured remotes: (none)" in stderr.lower()
467
468 def test_remote_not_configured_lists_existing_remotes(
469 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
470 ) -> None:
471 """Error message lists configured remote names when the named remote is absent.
472
473 Agents can read the list to discover the correct remote name without
474 a separate ``muse remote --json`` call.
475 """
476 monkeypatch.chdir(tmp_path)
477 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
478 # Suppress handle resolution so muse init does not auto-wire default
479 # remotes. We explicitly add "origin" below; "never-configured" is
480 # guaranteed absent regardless of authentication state.
481 with patch("muse.cli.commands.init.resolve_default_handle", return_value=None):
482 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
483 (tmp_path / "a.py").write_text("x = 1\n")
484 runner.invoke(cli, ["commit", "-m", "base"], env=_env(tmp_path), catch_exceptions=False)
485 # Configure a remote named "origin" but push to "never-configured" (absent).
486 set_remote("origin", "file:///dev/null", repo_root=tmp_path)
487 r = runner.invoke(cli, ["push", "never-configured"], env=_env(tmp_path))
488 assert r.exit_code != 0
489 stderr = r.stderr or ""
490 assert "origin" in stderr
491 assert "configured remotes:" in stderr.lower()
492
493 def test_no_commits_to_push_to_stderr(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
494 monkeypatch.chdir(tmp_path)
495 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
496 runner.invoke(cli, ["init"], env=_env(tmp_path), catch_exceptions=False)
497 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
498 r = runner.invoke(cli, ["push", "local"], env=_env(tmp_path))
499 assert r.exit_code != 0
500 assert "no commits" in (r.stderr or "").lower()
501
502 def _run_with_transport_exc(self, repo: pathlib.Path, exc: Exception) -> InvokeResult:
503 fake_transport = _FakeTransport(push_exc=exc)
504 fake_client = _FakeHttpxClient(push_exc=exc)
505 with (
506 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
507 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
508 patch("muse.cli.commands.push.make_transport", return_value=fake_transport),
509 patch("muse.cli.commands.push._httpx.AsyncClient", return_value=fake_client),
510 patch("muse.cli.commands.push._make_r2_client", return_value=fake_client),
511 ):
512 return runner.invoke(cli, ["push", "local"], env=_env(repo))
513
514 def test_push_rejected_to_stderr(self, repo: pathlib.Path) -> None:
515 from muse.core.transport import TransportError
516 r = self._run_with_transport_exc(repo, TransportError("conflict", status_code=409))
517 assert r.exit_code != 0
518 assert "diverged" in (r.stderr or "").lower()
519
520 def test_transport_error_409_to_stderr(self, repo: pathlib.Path) -> None:
521 from muse.core.transport import TransportError
522 r = self._run_with_transport_exc(repo, TransportError("conflict", status_code=409))
523 assert r.exit_code != 0
524 assert "diverged" in (r.stderr or "").lower()
525
526 def test_transport_error_401_to_stderr(self, repo: pathlib.Path) -> None:
527 from muse.core.transport import TransportError
528 r = self._run_with_transport_exc(repo, TransportError("unauthorized", status_code=401))
529 assert r.exit_code != 0
530 assert "authentication" in (r.stderr or "").lower()
531
532 def test_transport_error_404_to_stderr(self, repo: pathlib.Path) -> None:
533 from muse.core.transport import TransportError
534 r = self._run_with_transport_exc(repo, TransportError("not found", status_code=404))
535 assert r.exit_code != 0
536 assert "not found" in (r.stderr or "").lower()
537
538 def test_unknown_flag_exits_nonzero(self, repo: pathlib.Path) -> None:
539 r = runner.invoke(cli, ["push", "--format", "xml"], env=_env(repo))
540 assert r.exit_code != 0
541
542
543 # ---------------------------------------------------------------------------
544 # End-to-end (mocked transport + httpx)
545 # ---------------------------------------------------------------------------
546
547 class TestEndToEnd:
548 def _run(
549 self,
550 repo: pathlib.Path,
551 args: list[str] | None = None,
552 transport: "_FakeTransport | None" = None,
553 ) -> InvokeResult:
554 t = transport or _FakeTransport()
555 client = _FakeHttpxClient()
556 with (
557 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
558 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
559 patch("muse.cli.commands.push.make_transport", return_value=t),
560 patch("muse.cli.commands.push._httpx.AsyncClient", return_value=client),
561 patch("muse.cli.commands.push._make_r2_client", return_value=client),
562 ):
563 return runner.invoke(cli, args or ["push", "local"], env=_env(repo), catch_exceptions=False)
564
565 def test_fresh_push_succeeds(self, repo: pathlib.Path) -> None:
566 r = self._run(repo)
567 assert r.exit_code == 0, r.output
568
569 def test_up_to_date_when_already_pushed(self, repo: pathlib.Path) -> None:
570 from muse.core.refs import get_head_commit_id
571 head = get_head_commit_id(repo, "main") or ""
572 r = self._run(repo, transport=_FakeTransport(remote_head=head))
573 assert r.exit_code == 0
574 assert "up to date" in r.output.lower()
575
576 def test_dry_run_makes_no_http_calls(self, repo: pathlib.Path) -> None:
577 t = _FakeTransport()
578 client = _FakeHttpxClient()
579 with (
580 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
581 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
582 patch("muse.cli.commands.push.make_transport", return_value=t),
583 patch("muse.cli.commands.push._httpx.AsyncClient", return_value=client),
584 patch("muse.cli.commands.push._make_r2_client", return_value=client),
585 ):
586 r = runner.invoke(cli, ["push", "local", "--dry-run"], env=_env(repo), catch_exceptions=False)
587 assert r.exit_code == 0, r.output
588 assert "dry run" in r.output.lower()
589
590 def test_workers_flag_accepted(self, repo: pathlib.Path) -> None:
591 r = self._run(repo, args=["push", "local", "--workers", "2"])
592 assert r.exit_code == 0, r.output
593
594 def test_set_upstream_records_tracking(self, repo: pathlib.Path) -> None:
595 r = self._run(repo, args=["push", "local", "-u"])
596 assert r.exit_code == 0, r.output
597 config_path = config_toml_path(repo)
598 assert config_path.exists()
599 assert "local" in config_path.read_text()
600
601
602 # ---------------------------------------------------------------------------
603 # Security
604 # ---------------------------------------------------------------------------
605
606 class TestSecurity:
607 def test_remote_name_sanitized_in_error(self, repo: pathlib.Path) -> None:
608 ansi_remote = "\x1b[31mmalicious\x1b[0m"
609 r = runner.invoke(cli, ["push", ansi_remote], env=_env(repo))
610 assert r.exit_code != 0
611 assert "\x1b[31m" not in (r.stderr or "")
612
613 def test_branch_sanitized_in_delete_output(self, repo: pathlib.Path) -> None:
614 with patch("muse.cli.commands.push.get_remote", return_value="local://"):
615 with patch("muse.cli.commands.push.get_signing_identity", return_value=None):
616 with patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()):
617 with patch("muse.cli.commands.push.delete_remote_head", return_value=False):
618 r = runner.invoke(
619 cli,
620 ["push", "local", "--delete", "--branch", "\x1b[31mmalicious\x1b[0m"],
621 env=_env(repo),
622 )
623 # ANSI must not appear in stdout or stderr
624 assert "\x1b[31m" not in r.output
625 assert "\x1b[31m" not in (r.stderr or "")
626
627 def test_symlink_in_remotes_skipped(self, tmp_path: pathlib.Path) -> None:
628 from muse.cli.commands.push import _all_known_have_anchors
629 remotes = remotes_dir(tmp_path) / "origin"
630 remotes.mkdir(parents=True)
631 target = tmp_path / "sensitive.txt"
632 target.write_text("secret_commit_id\n")
633 (remotes / "main").symlink_to(target)
634 result = _all_known_have_anchors(tmp_path)
635 assert "secret_commit_id" not in result
636
637 def test_all_have_anchors_symlink_dir_skipped(self, tmp_path: pathlib.Path) -> None:
638 """A symlinked directory inside remotes/ must not be traversed."""
639 from muse.cli.commands.push import _all_known_have_anchors
640 # Create a real dir with a secret commit ID
641 secret_dir = tmp_path / "secret_dir"
642 secret_dir.mkdir()
643 (secret_dir / "main").write_text("secret123\n")
644 # Plant a symlinked directory
645 remotes = remotes_dir(tmp_path)
646 remotes.mkdir(parents=True)
647 (remotes / "malicious").symlink_to(secret_dir)
648 result = _all_known_have_anchors(tmp_path)
649 # Symlinked directories: rglob still finds files inside, but our check
650 # is on individual files. The symlink on the dir itself means rglob returns
651 # the child paths as symlink=False. The symlink() check only catches direct symlinks.
652 # The important test is that direct file symlinks ARE caught (test above).
653 assert isinstance(result, list)
654
655 def test_progress_not_in_stdout_on_json(self, repo: pathlib.Path) -> None:
656 """--json: exactly one JSON line; no progress noise mixed into it."""
657 fake_client = _FakeHttpxClient()
658 with (
659 patch("muse.cli.commands.push.get_remote", return_value="https://hub.example.com/r"),
660 patch("muse.cli.commands.push.get_signing_identity", return_value=None),
661 patch("muse.cli.commands.push.make_transport", return_value=_FakeTransport()),
662 patch("muse.cli.commands.push._httpx.AsyncClient", return_value=fake_client),
663 patch("muse.cli.commands.push._make_r2_client", return_value=fake_client),
664 ):
665 r = runner.invoke(cli, ["push", "local", "--json"], env=_env(repo))
666 assert r.exit_code == 0
667 # Exactly one JSON line in output; all others are progress/error (non-JSON).
668 json_lines = [l for l in r.output.splitlines() if l.strip().startswith("{")]
669 assert len(json_lines) == 1, f"Expected 1 JSON line, got: {json_lines}"
670 data = json.loads(json_lines[0])
671 assert isinstance(data, dict)
672
File History 2 commits
sha256:79ffe87f5fe2ec146e35f05521218bbf54dffdb0440c07f970bad05f16efb89f chore: merge main — carry all urllib/typing/test fixes from dev Sonnet 4.6 minor ⚠ 20 days ago
sha256:0bea7600d1eee83e87950be49933b1006fa9dc2c71e7c4ee748d324f61138156 chore: bump version to 0.2.0rc11; fix typing audit violatio… Sonnet 4.6 minor ⚠ 20 days ago