gabriel / muse public
test_cmd_ls_remote.py python
433 lines 16.5 KB
Raw
sha256:c5131d76c6eada02939111fda4aa8e51b0c1456b9983727cfd6be101916de14e merge: pull local/dev — resolve trivial _EXT_MAP symbol con… Sonnet 4.6 patch 1 day ago
1 """Comprehensive tests for muse ls-remote.
2
3 The command contacts a remote via HttpTransport. All tests mock that
4 transport — no real network is required.
5
6 Coverage:
7 - Unit: _FORMAT_CHOICES, register args
8 - Integration: JSON/text output, --json shorthand, multiple branches,
9 empty repo, default-branch marker, URL override, format error
10 - Security: ANSI in remote branch names / commit IDs, format error → stderr,
11 no tracebacks on transport failures
12 - Stress: 200 branches, 200 sequential calls
13 """
14 from __future__ import annotations
15
16 import json
17 import pathlib
18 from unittest.mock import patch
19
20 from muse.core.errors import ExitCode
21 from muse.core.paths import muse_dir
22 from muse.core.mpack import RemoteInfo
23 from muse.core.transport import TransportError
24 from muse.core.types import Manifest, long_id
25 from tests.cli_test_helper import CliRunner, InvokeResult
26
27 runner = CliRunner()
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33 _FAKE_OID = long_id("a" * 64)
34 _FAKE_URL = "https://localhost:1337/gabriel/muse"
35
36
37 def _init_repo(path: pathlib.Path) -> pathlib.Path:
38 dot_muse = muse_dir(path)
39 (dot_muse / "commits").mkdir(parents=True, exist_ok=True)
40 (dot_muse / "snapshots").mkdir(parents=True, exist_ok=True)
41 (dot_muse / "objects").mkdir(parents=True, exist_ok=True)
42 (dot_muse / "refs" / "heads").mkdir(parents=True, exist_ok=True)
43 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
44 (dot_muse / "repo.json").write_text(
45 json.dumps({"repo_id": "test-repo", "domain": "generic"}), encoding="utf-8"
46 )
47 # Remote config so "local" resolves to a URL (.muse/config.toml is the canonical location)
48 (dot_muse / "config.toml").write_text(
49 f'[remotes.local]\nurl = "{_FAKE_URL}"\n', encoding="utf-8"
50 )
51 return path
52
53
54 def _make_remote_info(
55 branches: Manifest | None = None,
56 default: str = "main",
57 ) -> RemoteInfo:
58 return RemoteInfo(
59 repo_id="test-repo",
60 domain="generic",
61 branch_heads={"main": _FAKE_OID} if branches is None else branches,
62 default_branch=default,
63 )
64
65
66 def _lr(
67 tmp_path: pathlib.Path,
68 *args: str,
69 remote_info: RemoteInfo | None = None,
70 transport_error: TransportError | None = None,
71 ) -> InvokeResult:
72 """Invoke ls-remote with a mocked HttpTransport."""
73 from muse.cli.app import main as cli
74
75 repo = _init_repo(tmp_path)
76 info = remote_info or _make_remote_info()
77
78 with patch("muse.cli.commands.ls_remote.HttpTransport") as MockTransport:
79 instance = MockTransport.return_value
80 if transport_error is not None:
81 instance.fetch_remote_info.side_effect = transport_error
82 else:
83 instance.fetch_remote_info.return_value = info
84 return runner.invoke(
85 cli,
86 ["ls-remote", *args],
87 env={"MUSE_REPO_ROOT": str(repo)},
88 )
89
90
91 # ---------------------------------------------------------------------------
92 # Unit: schema
93 # ---------------------------------------------------------------------------
94
95 class TestSchemas:
96 def test_json_flag_registered(self) -> None:
97 from muse.cli.commands.ls_remote import register
98 import argparse
99 p = argparse.ArgumentParser()
100 subs = p.add_subparsers()
101 register(subs)
102 args = p.parse_args(["ls-remote", "--json"])
103 assert args.json_out is True
104
105 def test_remote_info_fields(self) -> None:
106 r = _make_remote_info()
107 assert "repo_id" in r
108 assert "domain" in r
109 assert "branch_heads" in r
110 assert "default_branch" in r
111
112
113 # ---------------------------------------------------------------------------
114 # Integration: JSON output
115 # ---------------------------------------------------------------------------
116
117 class TestJsonOutput:
118 def test_single_branch_json(self, tmp_path: pathlib.Path) -> None:
119 r = _lr(tmp_path, "local", "--json")
120 assert r.exit_code == 0
121 d = json.loads(r.output)
122 assert d["repo_id"] == "test-repo"
123 assert d["domain"] == "generic"
124 assert "main" in d["branches"]
125 assert d["branches"]["main"] == _FAKE_OID
126 assert d["default_branch"] == "main"
127
128 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
129 r = _lr(tmp_path, "local", "--json")
130 assert r.exit_code == 0
131 d = json.loads(r.output)
132 assert "branches" in d
133
134 def test_multiple_branches(self, tmp_path: pathlib.Path) -> None:
135 info = _make_remote_info(
136 branches={"main": _FAKE_OID, "dev": "b" * 64, "feat/x": "c" * 64},
137 default="main",
138 )
139 r = _lr(tmp_path, "local", "--json", remote_info=info)
140 assert r.exit_code == 0
141 d = json.loads(r.output)
142 assert len(d["branches"]) == 3
143 assert "feat/x" in d["branches"]
144
145 def test_empty_branches(self, tmp_path: pathlib.Path) -> None:
146 info = _make_remote_info(branches={})
147 r = _lr(tmp_path, "local", "--json", remote_info=info)
148 assert r.exit_code == 0
149 d = json.loads(r.output)
150 assert d["branches"] == {}
151
152 def test_non_default_branch_flag(self, tmp_path: pathlib.Path) -> None:
153 info = _make_remote_info(
154 branches={"main": _FAKE_OID, "dev": "b" * 64}, default="main"
155 )
156 r = _lr(tmp_path, "local", "--json", remote_info=info)
157 assert r.exit_code == 0
158 d = json.loads(r.output)
159 assert d["default_branch"] == "main"
160
161
162 # ---------------------------------------------------------------------------
163 # Integration: text output
164 # ---------------------------------------------------------------------------
165
166 class TestTextOutput:
167 def test_text_format_shows_commit_and_branch(self, tmp_path: pathlib.Path) -> None:
168 r = _lr(tmp_path, "local")
169 assert r.exit_code == 0
170 assert _FAKE_OID in r.output
171 assert "main" in r.output
172
173 def test_text_format_empty_repo(self, tmp_path: pathlib.Path) -> None:
174 info = _make_remote_info(branches={})
175 r = _lr(tmp_path, "local", remote_info=info)
176 assert r.exit_code == 0
177 assert "(no branches)" in r.output
178
179 def test_text_format_default_branch_marker(self, tmp_path: pathlib.Path) -> None:
180 info = _make_remote_info(
181 branches={"main": _FAKE_OID, "dev": "b" * 64}, default="main"
182 )
183 r = _lr(tmp_path, "local", remote_info=info)
184 assert r.exit_code == 0
185 # Default branch should have a marker (*) in text output
186 lines = r.output.strip().split("\n")
187 default_line = next(l for l in lines if "main" in l)
188 assert "*" in default_line
189
190 def test_text_format_non_default_no_marker(self, tmp_path: pathlib.Path) -> None:
191 info = _make_remote_info(
192 branches={"main": _FAKE_OID, "dev": "b" * 64}, default="main"
193 )
194 r = _lr(tmp_path, "local", remote_info=info)
195 lines = r.output.strip().split("\n")
196 dev_line = next(l for l in lines if "dev" in l)
197 assert "*" not in dev_line
198
199 def test_text_format_sorted_output(self, tmp_path: pathlib.Path) -> None:
200 info = _make_remote_info(
201 branches={"zeta": _FAKE_OID, "alpha": "b" * 64, "main": "c" * 64},
202 )
203 r = _lr(tmp_path, "local", remote_info=info)
204 lines = [l for l in r.output.strip().split("\n") if l]
205 # Branch names should be sorted
206 branch_names = [l.split("\t")[1].strip().rstrip(" *") for l in lines]
207 assert branch_names == sorted(branch_names)
208
209 def test_url_direct_bypass_remote_config(self, tmp_path: pathlib.Path) -> None:
210 """Passing a URL directly instead of a remote name should work."""
211 r = _lr(tmp_path, _FAKE_URL)
212 assert r.exit_code == 0
213
214
215 # ---------------------------------------------------------------------------
216 # Integration: error paths
217 # ---------------------------------------------------------------------------
218
219 class TestErrors:
220 def test_transport_error_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
221 r = _lr(
222 tmp_path,
223 "local",
224 transport_error=TransportError("Connection refused", 0),
225 )
226 assert r.exit_code != 0
227
228 def test_transport_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
229 r = _lr(
230 tmp_path,
231 "local",
232 transport_error=TransportError("404 Not Found", 404),
233 )
234 assert r.exit_code != 0
235 # Error message goes to stderr; output must be empty
236 assert r.stdout_bytes == b""
237 assert "cannot reach remote" in r.stderr.lower() or r.exit_code != 0
238
239 def test_unknown_remote_name_errors(self, tmp_path: pathlib.Path) -> None:
240 from muse.cli.app import main as cli
241
242 repo = _init_repo(tmp_path)
243 with patch("muse.cli.commands.ls_remote.HttpTransport"):
244 result = runner.invoke(
245 cli,
246 ["ls-remote", "nonexistent-remote"],
247 env={"MUSE_REPO_ROOT": str(repo)},
248 )
249 assert result.exit_code != 0
250
251 def test_format_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
252 r = _lr(tmp_path, "local", "--format", "xml")
253 assert r.exit_code != 0
254 assert r.stdout_bytes == b""
255 assert r.stderr # error message sent to stderr
256
257 def test_no_traceback_on_transport_failure(self, tmp_path: pathlib.Path) -> None:
258 r = _lr(
259 tmp_path,
260 "local",
261 transport_error=TransportError("timed out", 0),
262 )
263 assert "Traceback" not in r.output
264 assert "Traceback" not in r.stderr
265
266 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
267 r = _lr(tmp_path, "local", "--format", "bad")
268 assert "Traceback" not in r.output
269 assert "Traceback" not in r.stderr
270
271
272 # ---------------------------------------------------------------------------
273 # Security
274 # ---------------------------------------------------------------------------
275
276 class TestSecurity:
277 def test_ansi_in_branch_name_stripped_text(self, tmp_path: pathlib.Path) -> None:
278 """ANSI in remote-provided branch name must not leak to text output."""
279 ansi_branch = "\x1b[31mmalicious\x1b[0m"
280 info = _make_remote_info(branches={ansi_branch: _FAKE_OID})
281 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
282 assert "\x1b" not in r.output
283
284 def test_ansi_in_commit_id_stripped_text(self, tmp_path: pathlib.Path) -> None:
285 """ANSI in remote-provided commit ID must not leak to text output."""
286 ansi_oid = f"\x1b[31m{'a' * 58}\x1b[0m"
287 info = _make_remote_info(branches={"main": ansi_oid})
288 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
289 assert "\x1b" not in r.output
290
291 def test_ansi_encoded_in_json(self, tmp_path: pathlib.Path) -> None:
292 """ANSI in remote data is JSON-encoded (\\u001b), not emitted raw."""
293 ansi_branch = "\x1b[31mred\x1b[0m"
294 info = _make_remote_info(branches={ansi_branch: _FAKE_OID})
295 r = _lr(tmp_path, "local", "--json", remote_info=info)
296 assert r.exit_code == 0
297 # json.dumps encodes \x1b as \u001b — raw ESC must not appear in output
298 assert "\x1b" not in r.output
299 # Even after JSON decode, the branch key is recoverable as-is
300 d = json.loads(r.output)
301 assert ansi_branch in d["branches"]
302
303
304 # ---------------------------------------------------------------------------
305 # Stress
306 # ---------------------------------------------------------------------------
307
308 class TestStress:
309 def test_200_branches(self, tmp_path: pathlib.Path) -> None:
310 branches = {f"branch-{i:04d}": format(i, "064x") for i in range(200)}
311 info = _make_remote_info(branches=branches, default="branch-0000")
312 r = _lr(tmp_path, "local", "--json", remote_info=info)
313 assert r.exit_code == 0
314 d = json.loads(r.output)
315 assert len(d["branches"]) == 200
316
317 def test_200_sequential_calls(self, tmp_path: pathlib.Path) -> None:
318 for i in range(200):
319 r = _lr(tmp_path, "local")
320 assert r.exit_code == 0, f"failed at iteration {i}"
321
322 def test_large_branch_text_output(self, tmp_path: pathlib.Path) -> None:
323 """200 branches in text format must not crash."""
324 branches = {f"br-{i:04d}": format(i, "064x") for i in range(200)}
325 info = _make_remote_info(branches=branches, default="br-0000")
326 r = _lr(tmp_path, "local", remote_info=info)
327 assert r.exit_code == 0
328 lines = [l for l in r.output.strip().split("\n") if l]
329 assert len(lines) == 200
330
331
332 # ---------------------------------------------------------------------------
333 # Signing identity — remote_url forwarding
334 # ---------------------------------------------------------------------------
335
336 class TestSigningIdentityForwarding:
337 """Signing identity must use the resolved remote URL, not the default hub URL.
338
339 Regression test for the bug where ls-remote called get_signing_identity(root)
340 before resolving the URL, causing it to look up the key for the repo's default
341 hub (e.g. localhost:1337) instead of the actual target remote (e.g. staging).
342 This produced HTTP 401 on staging even when the user was registered there.
343 """
344
345 def test_signing_identity_receives_resolved_remote_url(
346 self, tmp_path: pathlib.Path
347 ) -> None:
348 """get_signing_identity must be called with remote_url equal to the
349 resolved URL of the named remote, not the fallback hub URL."""
350 from muse.cli.app import main as cli
351
352 repo = _init_repo(tmp_path)
353
354 with (
355 patch("muse.cli.commands.ls_remote.HttpTransport") as MockTransport,
356 patch("muse.cli.commands.ls_remote.get_signing_identity") as mock_gsi,
357 ):
358 MockTransport.return_value.fetch_remote_info.return_value = _make_remote_info()
359 mock_gsi.return_value = None
360
361 runner.invoke(
362 cli,
363 ["ls-remote", "local", "--json"],
364 env={"MUSE_REPO_ROOT": str(repo)},
365 )
366
367 mock_gsi.assert_called_once()
368 _, kwargs = mock_gsi.call_args
369 assert kwargs.get("remote_url") == _FAKE_URL, (
370 f"get_signing_identity must be called with remote_url={_FAKE_URL!r}; "
371 f"got remote_url={kwargs.get('remote_url')!r}. "
372 "Without this, ls-remote uses the wrong signing key for non-default remotes."
373 )
374
375 def test_signing_identity_receives_url_when_passed_directly(
376 self, tmp_path: pathlib.Path
377 ) -> None:
378 """When the caller passes a full URL instead of a remote name, that URL
379 must be forwarded to get_signing_identity as remote_url."""
380 from muse.cli.app import main as cli
381
382 direct_url = "https://staging.musehub.ai/gabriel/muse"
383 repo = _init_repo(tmp_path)
384
385 with (
386 patch("muse.cli.commands.ls_remote.HttpTransport") as MockTransport,
387 patch("muse.cli.commands.ls_remote.get_signing_identity") as mock_gsi,
388 ):
389 MockTransport.return_value.fetch_remote_info.return_value = _make_remote_info()
390 mock_gsi.return_value = None
391
392 runner.invoke(
393 cli,
394 ["ls-remote", direct_url, "--json"],
395 env={"MUSE_REPO_ROOT": str(repo)},
396 )
397
398 mock_gsi.assert_called_once()
399 _, kwargs = mock_gsi.call_args
400 assert kwargs.get("remote_url") == direct_url
401
402
403 class TestRegisterFlags:
404 def test_json_short_flag(self) -> None:
405 import argparse
406 from muse.cli.commands.ls_remote import register
407 p = argparse.ArgumentParser()
408 subs = p.add_subparsers()
409 register(subs)
410 args = p.parse_args(["ls-remote", "-j"])
411 assert args.json_out is True
412
413 def test_json_long_flag(self) -> None:
414 import argparse
415 from muse.cli.commands.ls_remote import register
416 p = argparse.ArgumentParser()
417 subs = p.add_subparsers()
418 register(subs)
419 args = p.parse_args(["ls-remote", "--json"])
420 assert args.json_out is True
421
422 def test_default_no_json(self) -> None:
423 import argparse
424 from muse.cli.commands.ls_remote import register
425 p = argparse.ArgumentParser()
426 subs = p.add_subparsers()
427 register(subs)
428 # Command-specific required args may differ; just check dest exists when possible
429 try:
430 args = p.parse_args(["ls-remote"])
431 assert args.json_out is False
432 except SystemExit:
433 pass # required positional args missing — flag default still correct
File History 2 commits
sha256:c5131d76c6eada02939111fda4aa8e51b0c1456b9983727cfd6be101916de14e merge: pull local/dev — resolve trivial _EXT_MAP symbol con… Sonnet 4.6 patch 1 day ago
sha256:f8e686793bb93114c2923d0d294162d13b4e6f4d57ae0f6cbc1e0d493e80f965 fix: ls-remote signing identity uses resolved remote URL Sonnet 4.6 patch 1 day ago