gabriel / muse public
test_plumbing_ls_remote.py python
324 lines 12.4 KB
Raw
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 28 days ago
1 """Comprehensive tests for muse plumbing 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.cli.commands.plumbing.ls_remote import _FORMAT_CHOICES
21 from muse.core.errors import ExitCode
22 from muse.core.pack import RemoteInfo
23 from muse.core.transport import TransportError
24 from muse.core._types import Manifest
25 from tests.cli_test_helper import CliRunner, InvokeResult
26
27 runner = CliRunner()
28
29 # ---------------------------------------------------------------------------
30 # Helpers
31 # ---------------------------------------------------------------------------
32
33 _FAKE_OID = "a" * 64
34 _FAKE_URL = "http://localhost:10003/gabriel/muse"
35
36
37 def _init_repo(path: pathlib.Path) -> pathlib.Path:
38 muse = path / ".muse"
39 (muse / "commits").mkdir(parents=True, exist_ok=True)
40 (muse / "snapshots").mkdir(parents=True, exist_ok=True)
41 (muse / "objects").mkdir(parents=True, exist_ok=True)
42 (muse / "refs" / "heads").mkdir(parents=True, exist_ok=True)
43 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
44 (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 (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.plumbing.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_format_choices(self) -> None:
97 assert "json" in _FORMAT_CHOICES
98 assert "text" in _FORMAT_CHOICES
99
100 def test_remote_info_fields(self) -> None:
101 r = _make_remote_info()
102 assert "repo_id" in r
103 assert "domain" in r
104 assert "branch_heads" in r
105 assert "default_branch" in r
106
107
108 # ---------------------------------------------------------------------------
109 # Integration: JSON output
110 # ---------------------------------------------------------------------------
111
112 class TestJsonOutput:
113 def test_single_branch_json(self, tmp_path: pathlib.Path) -> None:
114 r = _lr(tmp_path, "local")
115 assert r.exit_code == 0
116 d = json.loads(r.output)
117 assert d["repo_id"] == "test-repo"
118 assert d["domain"] == "generic"
119 assert "main" in d["branches"]
120 assert d["branches"]["main"] == _FAKE_OID
121 assert d["default_branch"] == "main"
122
123 def test_json_shorthand(self, tmp_path: pathlib.Path) -> None:
124 r = _lr(tmp_path, "local", "--json")
125 assert r.exit_code == 0
126 d = json.loads(r.output)
127 assert "branches" in d
128
129 def test_multiple_branches(self, tmp_path: pathlib.Path) -> None:
130 info = _make_remote_info(
131 branches={"main": _FAKE_OID, "dev": "b" * 64, "feat/x": "c" * 64},
132 default="main",
133 )
134 r = _lr(tmp_path, "local", remote_info=info)
135 assert r.exit_code == 0
136 d = json.loads(r.output)
137 assert len(d["branches"]) == 3
138 assert "feat/x" in d["branches"]
139
140 def test_empty_branches(self, tmp_path: pathlib.Path) -> None:
141 info = _make_remote_info(branches={})
142 r = _lr(tmp_path, "local", remote_info=info)
143 assert r.exit_code == 0
144 d = json.loads(r.output)
145 assert d["branches"] == {}
146
147 def test_non_default_branch_flag(self, tmp_path: pathlib.Path) -> None:
148 info = _make_remote_info(
149 branches={"main": _FAKE_OID, "dev": "b" * 64}, default="main"
150 )
151 r = _lr(tmp_path, "local", remote_info=info)
152 assert r.exit_code == 0
153 d = json.loads(r.output)
154 assert d["default_branch"] == "main"
155
156
157 # ---------------------------------------------------------------------------
158 # Integration: text output
159 # ---------------------------------------------------------------------------
160
161 class TestTextOutput:
162 def test_text_format_shows_commit_and_branch(self, tmp_path: pathlib.Path) -> None:
163 r = _lr(tmp_path, "local", "--format", "text")
164 assert r.exit_code == 0
165 assert _FAKE_OID in r.output
166 assert "main" in r.output
167
168 def test_text_format_empty_repo(self, tmp_path: pathlib.Path) -> None:
169 info = _make_remote_info(branches={})
170 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
171 assert r.exit_code == 0
172 assert "(no branches)" in r.output
173
174 def test_text_format_default_branch_marker(self, tmp_path: pathlib.Path) -> None:
175 info = _make_remote_info(
176 branches={"main": _FAKE_OID, "dev": "b" * 64}, default="main"
177 )
178 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
179 assert r.exit_code == 0
180 # Default branch should have a marker (*) in text output
181 lines = r.output.strip().split("\n")
182 default_line = next(l for l in lines if "main" in l)
183 assert "*" in default_line
184
185 def test_text_format_non_default_no_marker(self, tmp_path: pathlib.Path) -> None:
186 info = _make_remote_info(
187 branches={"main": _FAKE_OID, "dev": "b" * 64}, default="main"
188 )
189 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
190 lines = r.output.strip().split("\n")
191 dev_line = next(l for l in lines if "dev" in l)
192 assert "*" not in dev_line
193
194 def test_text_format_sorted_output(self, tmp_path: pathlib.Path) -> None:
195 info = _make_remote_info(
196 branches={"zeta": _FAKE_OID, "alpha": "b" * 64, "main": "c" * 64},
197 )
198 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
199 lines = [l for l in r.output.strip().split("\n") if l]
200 # Branch names should be sorted
201 branch_names = [l.split("\t")[1].strip().rstrip(" *") for l in lines]
202 assert branch_names == sorted(branch_names)
203
204 def test_url_direct_bypass_remote_config(self, tmp_path: pathlib.Path) -> None:
205 """Passing a URL directly instead of a remote name should work."""
206 r = _lr(tmp_path, _FAKE_URL)
207 assert r.exit_code == 0
208
209
210 # ---------------------------------------------------------------------------
211 # Integration: error paths
212 # ---------------------------------------------------------------------------
213
214 class TestErrors:
215 def test_transport_error_exits_nonzero(self, tmp_path: pathlib.Path) -> None:
216 r = _lr(
217 tmp_path,
218 "local",
219 transport_error=TransportError("Connection refused", 0),
220 )
221 assert r.exit_code != 0
222
223 def test_transport_error_goes_to_stderr(self, tmp_path: pathlib.Path) -> None:
224 r = _lr(
225 tmp_path,
226 "local",
227 transport_error=TransportError("404 Not Found", 404),
228 )
229 assert r.exit_code != 0
230 # Error message goes to stderr; output must be empty
231 assert r.stdout_bytes == b""
232 assert "cannot reach remote" in r.stderr.lower() or r.exit_code != 0
233
234 def test_unknown_remote_name_errors(self, tmp_path: pathlib.Path) -> None:
235 from muse.cli.app import main as cli
236
237 repo = _init_repo(tmp_path)
238 with patch("muse.cli.commands.plumbing.ls_remote.HttpTransport"):
239 result = runner.invoke(
240 cli,
241 ["ls-remote", "nonexistent-remote"],
242 env={"MUSE_REPO_ROOT": str(repo)},
243 )
244 assert result.exit_code != 0
245
246 def test_format_error_to_stderr(self, tmp_path: pathlib.Path) -> None:
247 r = _lr(tmp_path, "local", "--format", "xml")
248 assert r.exit_code != 0
249 assert r.stdout_bytes == b""
250 assert "error" in r.stderr.lower()
251
252 def test_no_traceback_on_transport_failure(self, tmp_path: pathlib.Path) -> None:
253 r = _lr(
254 tmp_path,
255 "local",
256 transport_error=TransportError("timed out", 0),
257 )
258 assert "Traceback" not in r.output
259 assert "Traceback" not in r.stderr
260
261 def test_no_traceback_on_bad_format(self, tmp_path: pathlib.Path) -> None:
262 r = _lr(tmp_path, "local", "--format", "bad")
263 assert "Traceback" not in r.output
264 assert "Traceback" not in r.stderr
265
266
267 # ---------------------------------------------------------------------------
268 # Security
269 # ---------------------------------------------------------------------------
270
271 class TestSecurity:
272 def test_ansi_in_branch_name_stripped_text(self, tmp_path: pathlib.Path) -> None:
273 """ANSI in remote-provided branch name must not leak to text output."""
274 ansi_branch = "\x1b[31mmalicious\x1b[0m"
275 info = _make_remote_info(branches={ansi_branch: _FAKE_OID})
276 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
277 assert "\x1b" not in r.output
278
279 def test_ansi_in_commit_id_stripped_text(self, tmp_path: pathlib.Path) -> None:
280 """ANSI in remote-provided commit ID must not leak to text output."""
281 ansi_oid = "\x1b[31m" + "a" * 58 + "\x1b[0m"
282 info = _make_remote_info(branches={"main": ansi_oid})
283 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
284 assert "\x1b" not in r.output
285
286 def test_ansi_encoded_in_json(self, tmp_path: pathlib.Path) -> None:
287 """ANSI in remote data is JSON-encoded (\\u001b), not emitted raw."""
288 ansi_branch = "\x1b[31mred\x1b[0m"
289 info = _make_remote_info(branches={ansi_branch: _FAKE_OID})
290 r = _lr(tmp_path, "local", remote_info=info)
291 assert r.exit_code == 0
292 # json.dumps encodes \x1b as \u001b — raw ESC must not appear in output
293 assert "\x1b" not in r.output
294 # Even after JSON decode, the branch key is recoverable as-is
295 d = json.loads(r.output)
296 assert ansi_branch in d["branches"]
297
298
299 # ---------------------------------------------------------------------------
300 # Stress
301 # ---------------------------------------------------------------------------
302
303 class TestStress:
304 def test_200_branches(self, tmp_path: pathlib.Path) -> None:
305 branches = {f"branch-{i:04d}": format(i, "064x") for i in range(200)}
306 info = _make_remote_info(branches=branches, default="branch-0000")
307 r = _lr(tmp_path, "local", remote_info=info)
308 assert r.exit_code == 0
309 d = json.loads(r.output)
310 assert len(d["branches"]) == 200
311
312 def test_200_sequential_calls(self, tmp_path: pathlib.Path) -> None:
313 for i in range(200):
314 r = _lr(tmp_path, "local")
315 assert r.exit_code == 0, f"failed at iteration {i}"
316
317 def test_large_branch_text_output(self, tmp_path: pathlib.Path) -> None:
318 """200 branches in text format must not crash."""
319 branches = {f"br-{i:04d}": format(i, "064x") for i in range(200)}
320 info = _make_remote_info(branches=branches, default="br-0000")
321 r = _lr(tmp_path, "local", "--format", "text", remote_info=info)
322 assert r.exit_code == 0
323 lines = [l for l in r.output.strip().split("\n") if l]
324 assert len(lines) == 200
File History 3 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 28 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 28 days ago
sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 102 days ago