gabriel / muse public
test_ls_remote_supercharge.py python
388 lines 15.3 KB
Raw
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 20 days ago
1 """Supercharge tests for ``muse ls-remote``.
2
3 Coverage tiers
4 --------------
5 - JSON envelope schema: all required keys always present
6 - Error payload shape: exactly {status, error, exit_code} — no prose in --json mode
7 - Remote/URL fields: remote name resolved, URL echoed
8 - Duration: duration_ms is a non-negative float
9 - TypedDicts: stable wire-format types exist and are annotated
10 - Docstring: module docstring covers all envelope fields and error schema
11 - No-prose pollution: no emoji/traceback in JSON mode
12 - Data integrity: sha256: OIDs even when remote sends bare hex (defense in depth)
13 """
14 from __future__ import annotations
15
16 import json
17 import pathlib
18 from typing import get_type_hints
19 from unittest.mock import patch
20
21 from muse.core.errors import ExitCode
22 from muse.core.mpack import RemoteInfo
23 from muse.core.transport import TransportError
24 from tests.cli_test_helper import CliRunner, InvokeResult
25 from muse.core.types import long_id
26 from muse.core.paths import muse_dir
27
28 runner = CliRunner()
29
30 # ---------------------------------------------------------------------------
31 # Shared fixtures
32 # ---------------------------------------------------------------------------
33
34 _FAKE_BARE_OID = "a" * 64 # bare hex — simulates non-compliant remote
35 _FAKE_OID = long_id("a" * 64)# canonical form
36 _FAKE_URL = "https://localhost:1337/gabriel/muse"
37 _REMOTE_NAME = "local"
38
39
40 def _init_repo(path: pathlib.Path) -> pathlib.Path:
41 dot_muse = muse_dir(path)
42 (dot_muse / "commits").mkdir(parents=True, exist_ok=True)
43 (dot_muse / "snapshots").mkdir(parents=True, exist_ok=True)
44 (dot_muse / "objects").mkdir(parents=True, exist_ok=True)
45 (dot_muse / "refs" / "heads").mkdir(parents=True, exist_ok=True)
46 (dot_muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
47 (dot_muse / "repo.json").write_text(
48 json.dumps({"repo_id": "test-repo", "domain": "generic"}), encoding="utf-8"
49 )
50 (dot_muse / "config.toml").write_text(
51 f'[remotes.{_REMOTE_NAME}]\nurl = "{_FAKE_URL}"\n', encoding="utf-8"
52 )
53 return path
54
55
56 def _make_remote_info(
57 branches: dict[str, str] | None = None,
58 default: str = "main",
59 ) -> RemoteInfo:
60 # Intentionally use bare hex OIDs to test the defense-in-depth normalization.
61 return RemoteInfo(
62 repo_id="test-repo",
63 domain="generic",
64 branch_heads={"main": _FAKE_BARE_OID} if branches is None else branches,
65 default_branch=default,
66 )
67
68
69 def _lr(
70 tmp_path: pathlib.Path,
71 *args: str,
72 remote_info: RemoteInfo | None = None,
73 transport_error: TransportError | None = None,
74 ) -> InvokeResult:
75 from muse.cli.app import main as cli
76
77 repo = _init_repo(tmp_path)
78 info = remote_info or _make_remote_info()
79
80 with patch("muse.cli.commands.ls_remote.HttpTransport") as MockTransport:
81 instance = MockTransport.return_value
82 if transport_error is not None:
83 instance.fetch_remote_info.side_effect = transport_error
84 else:
85 instance.fetch_remote_info.return_value = info
86 extra = [] if "--json" in args or "-j" in args else ["--json"]
87 return runner.invoke(
88 cli,
89 ["ls-remote", *extra, *args],
90 env={"MUSE_REPO_ROOT": str(repo)},
91 )
92
93
94 # ---------------------------------------------------------------------------
95 # JSON envelope schema
96 # ---------------------------------------------------------------------------
97
98 class TestJsonEnvelopeSchema:
99 """Every required key is present in the success envelope."""
100
101 _REQUIRED_KEYS = {
102 "status", "error", "repo_id", "domain", "default_branch",
103 "branches", "remote", "url", "duration_ms", "exit_code",
104 }
105
106 def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None:
107 r = _lr(tmp_path, _REMOTE_NAME)
108 assert r.exit_code == 0
109 d = json.loads(r.output)
110 missing = self._REQUIRED_KEYS - d.keys()
111 assert not missing, f"Missing keys: {missing}"
112
113 def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None:
114 r = _lr(tmp_path, _REMOTE_NAME)
115 assert json.loads(r.output)["status"] == "ok"
116
117 def test_error_empty_on_success(self, tmp_path: pathlib.Path) -> None:
118 r = _lr(tmp_path, _REMOTE_NAME)
119 assert json.loads(r.output)["error"] == ""
120
121 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
122 r = _lr(tmp_path, _REMOTE_NAME)
123 assert json.loads(r.output)["exit_code"] == 0
124
125 def test_duration_ms_is_nonneg_float(self, tmp_path: pathlib.Path) -> None:
126 r = _lr(tmp_path, _REMOTE_NAME)
127 d = json.loads(r.output)
128 assert isinstance(d["duration_ms"], float)
129 assert d["duration_ms"] >= 0.0
130
131 def test_remote_field_reflects_name(self, tmp_path: pathlib.Path) -> None:
132 r = _lr(tmp_path, _REMOTE_NAME)
133 d = json.loads(r.output)
134 assert d["remote"] == _REMOTE_NAME
135
136 def test_url_field_reflects_resolved_url(self, tmp_path: pathlib.Path) -> None:
137 r = _lr(tmp_path, _REMOTE_NAME)
138 d = json.loads(r.output)
139 assert d["url"] == _FAKE_URL
140
141 def test_remote_null_when_url_passed_directly(self, tmp_path: pathlib.Path) -> None:
142 """When the caller passes a full URL, no remote name was resolved — remote=null."""
143 r = _lr(tmp_path, _FAKE_URL)
144 d = json.loads(r.output)
145 assert d["remote"] is None
146
147 def test_url_present_when_url_passed_directly(self, tmp_path: pathlib.Path) -> None:
148 r = _lr(tmp_path, _FAKE_URL)
149 d = json.loads(r.output)
150 assert d["url"] == _FAKE_URL
151
152 def test_repo_id_matches_remote(self, tmp_path: pathlib.Path) -> None:
153 r = _lr(tmp_path, _REMOTE_NAME)
154 d = json.loads(r.output)
155 assert d["repo_id"] == "test-repo"
156
157 def test_domain_matches_remote(self, tmp_path: pathlib.Path) -> None:
158 r = _lr(tmp_path, _REMOTE_NAME)
159 d = json.loads(r.output)
160 assert d["domain"] == "generic"
161
162 def test_branches_is_dict(self, tmp_path: pathlib.Path) -> None:
163 r = _lr(tmp_path, _REMOTE_NAME)
164 d = json.loads(r.output)
165 assert isinstance(d["branches"], dict)
166
167
168 # ---------------------------------------------------------------------------
169 # Error payload shape
170 # ---------------------------------------------------------------------------
171
172 class TestErrorPayloadShape:
173 """In --json mode, errors go to stdout as {status, error, exit_code}."""
174
175 def test_error_payload_has_exactly_three_keys(self, tmp_path: pathlib.Path) -> None:
176 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
177 d = json.loads(r.output)
178 assert {"status", "error", "exit_code"}.issubset(d.keys())
179
180 def test_error_status_on_failure(self, tmp_path: pathlib.Path) -> None:
181 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
182 d = json.loads(r.output)
183 assert d["status"] == "error"
184
185 def test_error_message_nonempty(self, tmp_path: pathlib.Path) -> None:
186 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
187 d = json.loads(r.output)
188 assert d["error"]
189
190 def test_exit_code_nonzero_on_error(self, tmp_path: pathlib.Path) -> None:
191 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
192 assert r.exit_code != 0
193
194 def test_unknown_remote_error_is_json_in_json_mode(self, tmp_path: pathlib.Path) -> None:
195 """Unknown remote name → JSON error on stdout, not prose on stderr."""
196 from muse.cli.app import main as cli
197
198 repo = _init_repo(tmp_path)
199 with patch("muse.cli.commands.ls_remote.HttpTransport"):
200 r = runner.invoke(
201 cli,
202 ["ls-remote", "--json", "ghost-remote"],
203 env={"MUSE_REPO_ROOT": str(repo)},
204 )
205 assert r.exit_code != 0
206 d = json.loads(r.output)
207 assert d["status"] == "error"
208
209 def test_transport_error_is_json_in_json_mode(self, tmp_path: pathlib.Path) -> None:
210 """Transport error → JSON payload on stdout in json mode, no prose."""
211 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("refused", 0))
212 assert r.exit_code != 0
213 d = json.loads(r.output) # must be valid JSON
214 assert d["status"] == "error"
215
216
217 # ---------------------------------------------------------------------------
218 # Data integrity — sha256: normalization
219 # ---------------------------------------------------------------------------
220
221 class TestDataIntegrity:
222 """Remote-provided OIDs must be normalized to sha256: prefix."""
223
224 def test_bare_hex_oid_normalized_in_json(self, tmp_path: pathlib.Path) -> None:
225 """When remote returns bare hex, output has sha256: prefix."""
226 info = _make_remote_info(branches={"main": _FAKE_BARE_OID})
227 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
228 d = json.loads(r.output)
229 assert d["branches"]["main"].startswith("sha256:"), (
230 f"Expected sha256: prefix, got: {d['branches']['main']!r}"
231 )
232
233 def test_already_prefixed_oid_unchanged(self, tmp_path: pathlib.Path) -> None:
234 """When remote returns sha256:-prefixed OID, output is identical."""
235 info = _make_remote_info(branches={"main": _FAKE_OID})
236 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
237 d = json.loads(r.output)
238 assert d["branches"]["main"] == _FAKE_OID
239
240 def test_bare_hex_oid_normalized_in_text(self, tmp_path: pathlib.Path) -> None:
241 """Text output also normalizes bare hex to sha256:."""
242 from muse.cli.app import main as cli
243 repo = _init_repo(tmp_path)
244 info = _make_remote_info(branches={"main": _FAKE_BARE_OID})
245 with patch("muse.cli.commands.ls_remote.HttpTransport") as MockTransport:
246 MockTransport.return_value.fetch_remote_info.return_value = info
247 r = runner.invoke(cli, ["ls-remote", _REMOTE_NAME],
248 env={"MUSE_REPO_ROOT": str(repo)})
249 assert r.exit_code == 0
250 assert "sha256:" in r.output
251
252 def test_multiple_branches_all_normalized(self, tmp_path: pathlib.Path) -> None:
253 """All branch OIDs are normalized, not just the first one."""
254 branches = {f"b{i}": "f" * 64 for i in range(5)}
255 info = _make_remote_info(branches=branches)
256 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
257 d = json.loads(r.output)
258 for name, oid in d["branches"].items():
259 assert oid.startswith("sha256:"), f"branch {name!r} not normalized: {oid!r}"
260
261 def test_branch_values_are_strings(self, tmp_path: pathlib.Path) -> None:
262 r = _lr(tmp_path, _REMOTE_NAME)
263 d = json.loads(r.output)
264 for oid in d["branches"].values():
265 assert isinstance(oid, str)
266
267
268 # ---------------------------------------------------------------------------
269 # No-prose pollution
270 # ---------------------------------------------------------------------------
271
272 class TestNoProsePollution:
273 def test_stdout_is_valid_json_in_json_mode(self, tmp_path: pathlib.Path) -> None:
274 r = _lr(tmp_path, _REMOTE_NAME)
275 json.loads(r.output) # must not raise
276
277 def test_no_emoji_in_json_stdout(self, tmp_path: pathlib.Path) -> None:
278 r = _lr(tmp_path, _REMOTE_NAME)
279 assert "❌" not in r.output
280 assert "✅" not in r.output
281
282 def test_error_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
283 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("boom", 0))
284 json.loads(r.output) # must not raise
285
286 def test_no_traceback_in_json_mode(self, tmp_path: pathlib.Path) -> None:
287 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("boom", 0))
288 assert "Traceback" not in r.output
289 assert "Traceback" not in r.stderr
290
291 def test_ansi_in_json_output_is_encoded(self, tmp_path: pathlib.Path) -> None:
292 """ANSI in remote branch names must be JSON-encoded, not emitted raw."""
293 ansi_branch = "\x1b[31mbad\x1b[0m"
294 info = _make_remote_info(branches={ansi_branch: _FAKE_BARE_OID})
295 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
296 assert r.exit_code == 0
297 assert "\x1b" not in r.output
298 d = json.loads(r.output)
299 assert ansi_branch in d["branches"]
300
301
302 # ---------------------------------------------------------------------------
303 # TypedDicts
304 # ---------------------------------------------------------------------------
305
306 class TestTypedDicts:
307 def test_ls_remote_json_typeddict_exists(self) -> None:
308 from muse.cli.commands.ls_remote import _LsRemoteJson
309 assert _LsRemoteJson is not None
310
311 def test_ls_remote_error_json_typeddict_exists(self) -> None:
312 from muse.cli.commands.ls_remote import _LsRemoteErrorJson
313 assert _LsRemoteErrorJson is not None
314
315 def test_ls_remote_json_has_status_annotation(self) -> None:
316 from muse.cli.commands.ls_remote import _LsRemoteJson
317 hints = get_type_hints(_LsRemoteJson)
318 assert "status" in hints
319
320 def test_ls_remote_json_has_all_new_fields(self) -> None:
321 from muse.cli.commands.ls_remote import _LsRemoteJson
322 hints = get_type_hints(_LsRemoteJson)
323 for field in ("status", "error", "remote", "url", "duration_ms", "exit_code"):
324 assert field in hints, f"Missing annotation: {field!r}"
325
326
327 # ---------------------------------------------------------------------------
328 # Docstring coverage
329 # ---------------------------------------------------------------------------
330
331 class TestDocstring:
332 def _doc(self) -> str:
333 import muse.cli.commands.ls_remote as mod
334 return mod.__doc__ or ""
335
336 def test_docstring_documents_status(self) -> None:
337 assert "status" in self._doc()
338
339 def test_docstring_documents_error(self) -> None:
340 assert "error" in self._doc()
341
342 def test_docstring_documents_remote(self) -> None:
343 assert "remote" in self._doc()
344
345 def test_docstring_documents_url(self) -> None:
346 assert "url" in self._doc()
347
348 def test_docstring_documents_duration_ms(self) -> None:
349 assert "duration_ms" in self._doc()
350
351 def test_docstring_documents_exit_code(self) -> None:
352 assert "exit_code" in self._doc()
353
354 def test_docstring_documents_error_schema(self) -> None:
355 assert "error" in self._doc() and "exit_code" in self._doc()
356
357
358 class TestRegisterFlags:
359 def test_json_short_flag(self) -> None:
360 import argparse
361 from muse.cli.commands.ls_remote import register
362 p = argparse.ArgumentParser()
363 subs = p.add_subparsers()
364 register(subs)
365 args = p.parse_args(["ls-remote", "-j"])
366 assert args.json_out is True
367
368 def test_json_long_flag(self) -> None:
369 import argparse
370 from muse.cli.commands.ls_remote import register
371 p = argparse.ArgumentParser()
372 subs = p.add_subparsers()
373 register(subs)
374 args = p.parse_args(["ls-remote", "--json"])
375 assert args.json_out is True
376
377 def test_default_no_json(self) -> None:
378 import argparse
379 from muse.cli.commands.ls_remote import register
380 p = argparse.ArgumentParser()
381 subs = p.add_subparsers()
382 register(subs)
383 # Command-specific required args may differ; just check dest exists when possible
384 try:
385 args = p.parse_args(["ls-remote"])
386 assert args.json_out is False
387 except SystemExit:
388 pass # required positional args missing — flag default still correct
File History 4 commits
sha256:81ae324db5ad375fbfe4834c6fcb378312cafad3cc92dec5d3e5c427306621a2 fix: remove commit_exists filter from have anchors — server… Sonnet 4.6 patch 20 days ago
sha256:36c3cb3e76619d4c30a6d9bf81b5ec4ff148e30dcfed913e3114ca7b43b81c7e fix: rename objects→blobs in push client and all stale test… Sonnet 4.6 patch 22 days ago
sha256:c06a9b9b9fee26c68ea725b44d54b2c0a171301ce9de746d5b656617b4463a9a fix: repair four test failures from post-migration audit Sonnet 4.6 patch 28 days ago
sha256:1900655993c83c4107067375548a7be823e471d2515830842f1a12cba4bd3cdf fix: unified object store migration — idempotent writes, JS… Sonnet 4.6 minor ⚠ 28 days ago