gabriel / muse public
test_ls_remote_supercharge.py python
348 lines 13.7 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 141 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.pack 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
27 runner = CliRunner()
28
29 # ---------------------------------------------------------------------------
30 # Shared fixtures
31 # ---------------------------------------------------------------------------
32
33 _FAKE_BARE_OID = "a" * 64 # bare hex — simulates non-compliant remote
34 _FAKE_OID = long_id("a" * 64)# canonical form
35 _FAKE_URL = "http://localhost:10003/gabriel/muse"
36 _REMOTE_NAME = "local"
37
38
39 def _init_repo(path: pathlib.Path) -> pathlib.Path:
40 muse = path / ".muse"
41 (muse / "commits").mkdir(parents=True, exist_ok=True)
42 (muse / "snapshots").mkdir(parents=True, exist_ok=True)
43 (muse / "objects").mkdir(parents=True, exist_ok=True)
44 (muse / "refs" / "heads").mkdir(parents=True, exist_ok=True)
45 (muse / "HEAD").write_text("ref: refs/heads/main", encoding="utf-8")
46 (muse / "repo.json").write_text(
47 json.dumps({"repo_id": "test-repo", "domain": "generic"}), encoding="utf-8"
48 )
49 (muse / "config.toml").write_text(
50 f'[remotes.{_REMOTE_NAME}]\nurl = "{_FAKE_URL}"\n', encoding="utf-8"
51 )
52 return path
53
54
55 def _make_remote_info(
56 branches: dict[str, str] | None = None,
57 default: str = "main",
58 ) -> RemoteInfo:
59 # Intentionally use bare hex OIDs to test the defense-in-depth normalization.
60 return RemoteInfo(
61 repo_id="test-repo",
62 domain="generic",
63 branch_heads={"main": _FAKE_BARE_OID} if branches is None else branches,
64 default_branch=default,
65 )
66
67
68 def _lr(
69 tmp_path: pathlib.Path,
70 *args: str,
71 remote_info: RemoteInfo | None = None,
72 transport_error: TransportError | None = None,
73 ) -> InvokeResult:
74 from muse.cli.app import main as cli
75
76 repo = _init_repo(tmp_path)
77 info = remote_info or _make_remote_info()
78
79 with patch("muse.cli.commands.ls_remote.HttpTransport") as MockTransport:
80 instance = MockTransport.return_value
81 if transport_error is not None:
82 instance.fetch_remote_info.side_effect = transport_error
83 else:
84 instance.fetch_remote_info.return_value = info
85 return runner.invoke(
86 cli,
87 ["ls-remote", *args],
88 env={"MUSE_REPO_ROOT": str(repo)},
89 )
90
91
92 # ---------------------------------------------------------------------------
93 # JSON envelope schema
94 # ---------------------------------------------------------------------------
95
96 class TestJsonEnvelopeSchema:
97 """Every required key is present in the success envelope."""
98
99 _REQUIRED_KEYS = {
100 "status", "error", "repo_id", "domain", "default_branch",
101 "branches", "remote", "url", "duration_ms", "exit_code",
102 }
103
104 def test_all_required_keys_present(self, tmp_path: pathlib.Path) -> None:
105 r = _lr(tmp_path, _REMOTE_NAME)
106 assert r.exit_code == 0
107 d = json.loads(r.output)
108 missing = self._REQUIRED_KEYS - d.keys()
109 assert not missing, f"Missing keys: {missing}"
110
111 def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None:
112 r = _lr(tmp_path, _REMOTE_NAME)
113 assert json.loads(r.output)["status"] == "ok"
114
115 def test_error_empty_on_success(self, tmp_path: pathlib.Path) -> None:
116 r = _lr(tmp_path, _REMOTE_NAME)
117 assert json.loads(r.output)["error"] == ""
118
119 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
120 r = _lr(tmp_path, _REMOTE_NAME)
121 assert json.loads(r.output)["exit_code"] == 0
122
123 def test_duration_ms_is_nonneg_float(self, tmp_path: pathlib.Path) -> None:
124 r = _lr(tmp_path, _REMOTE_NAME)
125 d = json.loads(r.output)
126 assert isinstance(d["duration_ms"], float)
127 assert d["duration_ms"] >= 0.0
128
129 def test_remote_field_reflects_name(self, tmp_path: pathlib.Path) -> None:
130 r = _lr(tmp_path, _REMOTE_NAME)
131 d = json.loads(r.output)
132 assert d["remote"] == _REMOTE_NAME
133
134 def test_url_field_reflects_resolved_url(self, tmp_path: pathlib.Path) -> None:
135 r = _lr(tmp_path, _REMOTE_NAME)
136 d = json.loads(r.output)
137 assert d["url"] == _FAKE_URL
138
139 def test_remote_null_when_url_passed_directly(self, tmp_path: pathlib.Path) -> None:
140 """When the caller passes a full URL, no remote name was resolved — remote=null."""
141 r = _lr(tmp_path, _FAKE_URL)
142 d = json.loads(r.output)
143 assert d["remote"] is None
144
145 def test_url_present_when_url_passed_directly(self, tmp_path: pathlib.Path) -> None:
146 r = _lr(tmp_path, _FAKE_URL)
147 d = json.loads(r.output)
148 assert d["url"] == _FAKE_URL
149
150 def test_repo_id_matches_remote(self, tmp_path: pathlib.Path) -> None:
151 r = _lr(tmp_path, _REMOTE_NAME)
152 d = json.loads(r.output)
153 assert d["repo_id"] == "test-repo"
154
155 def test_domain_matches_remote(self, tmp_path: pathlib.Path) -> None:
156 r = _lr(tmp_path, _REMOTE_NAME)
157 d = json.loads(r.output)
158 assert d["domain"] == "generic"
159
160 def test_branches_is_dict(self, tmp_path: pathlib.Path) -> None:
161 r = _lr(tmp_path, _REMOTE_NAME)
162 d = json.loads(r.output)
163 assert isinstance(d["branches"], dict)
164
165
166 # ---------------------------------------------------------------------------
167 # Error payload shape
168 # ---------------------------------------------------------------------------
169
170 class TestErrorPayloadShape:
171 """In --json mode, errors go to stdout as {status, error, exit_code}."""
172
173 def test_error_payload_has_exactly_three_keys(self, tmp_path: pathlib.Path) -> None:
174 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
175 d = json.loads(r.output)
176 assert set(d.keys()) == {"status", "error", "exit_code"}
177
178 def test_error_status_on_failure(self, tmp_path: pathlib.Path) -> None:
179 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
180 d = json.loads(r.output)
181 assert d["status"] == "error"
182
183 def test_error_message_nonempty(self, tmp_path: pathlib.Path) -> None:
184 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
185 d = json.loads(r.output)
186 assert d["error"]
187
188 def test_exit_code_nonzero_on_error(self, tmp_path: pathlib.Path) -> None:
189 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("down", 0))
190 assert r.exit_code != 0
191
192 def test_unknown_remote_error_is_json_in_json_mode(self, tmp_path: pathlib.Path) -> None:
193 """Unknown remote name → JSON error on stdout, not prose on stderr."""
194 from muse.cli.app import main as cli
195
196 repo = _init_repo(tmp_path)
197 with patch("muse.cli.commands.ls_remote.HttpTransport"):
198 r = runner.invoke(
199 cli,
200 ["ls-remote", "ghost-remote"],
201 env={"MUSE_REPO_ROOT": str(repo)},
202 )
203 assert r.exit_code != 0
204 d = json.loads(r.output)
205 assert d["status"] == "error"
206
207 def test_transport_error_is_json_in_json_mode(self, tmp_path: pathlib.Path) -> None:
208 """Transport error → JSON payload on stdout in json mode, no prose."""
209 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("refused", 0))
210 assert r.exit_code != 0
211 d = json.loads(r.output) # must be valid JSON
212 assert d["status"] == "error"
213
214
215 # ---------------------------------------------------------------------------
216 # Data integrity — sha256: normalization
217 # ---------------------------------------------------------------------------
218
219 class TestDataIntegrity:
220 """Remote-provided OIDs must be normalized to sha256: prefix."""
221
222 def test_bare_hex_oid_normalized_in_json(self, tmp_path: pathlib.Path) -> None:
223 """When remote returns bare hex, output has sha256: prefix."""
224 info = _make_remote_info(branches={"main": _FAKE_BARE_OID})
225 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
226 d = json.loads(r.output)
227 assert d["branches"]["main"].startswith("sha256:"), (
228 f"Expected sha256: prefix, got: {d['branches']['main']!r}"
229 )
230
231 def test_already_prefixed_oid_unchanged(self, tmp_path: pathlib.Path) -> None:
232 """When remote returns sha256:-prefixed OID, output is identical."""
233 info = _make_remote_info(branches={"main": _FAKE_OID})
234 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
235 d = json.loads(r.output)
236 assert d["branches"]["main"] == _FAKE_OID
237
238 def test_bare_hex_oid_normalized_in_text(self, tmp_path: pathlib.Path) -> None:
239 """Text output also normalizes bare hex to sha256:."""
240 info = _make_remote_info(branches={"main": _FAKE_BARE_OID})
241 r = _lr(tmp_path, _REMOTE_NAME, "--format", "text", remote_info=info)
242 assert r.exit_code == 0
243 assert "sha256:" in r.output
244
245 def test_multiple_branches_all_normalized(self, tmp_path: pathlib.Path) -> None:
246 """All branch OIDs are normalized, not just the first one."""
247 branches = {f"b{i}": "f" * 64 for i in range(5)}
248 info = _make_remote_info(branches=branches)
249 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
250 d = json.loads(r.output)
251 for name, oid in d["branches"].items():
252 assert oid.startswith("sha256:"), f"branch {name!r} not normalized: {oid!r}"
253
254 def test_branch_values_are_strings(self, tmp_path: pathlib.Path) -> None:
255 r = _lr(tmp_path, _REMOTE_NAME)
256 d = json.loads(r.output)
257 for oid in d["branches"].values():
258 assert isinstance(oid, str)
259
260
261 # ---------------------------------------------------------------------------
262 # No-prose pollution
263 # ---------------------------------------------------------------------------
264
265 class TestNoProsePollution:
266 def test_stdout_is_valid_json_in_json_mode(self, tmp_path: pathlib.Path) -> None:
267 r = _lr(tmp_path, _REMOTE_NAME)
268 json.loads(r.output) # must not raise
269
270 def test_no_emoji_in_json_stdout(self, tmp_path: pathlib.Path) -> None:
271 r = _lr(tmp_path, _REMOTE_NAME)
272 assert "❌" not in r.output
273 assert "✅" not in r.output
274
275 def test_error_stdout_is_valid_json(self, tmp_path: pathlib.Path) -> None:
276 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("boom", 0))
277 json.loads(r.output) # must not raise
278
279 def test_no_traceback_in_json_mode(self, tmp_path: pathlib.Path) -> None:
280 r = _lr(tmp_path, _REMOTE_NAME, transport_error=TransportError("boom", 0))
281 assert "Traceback" not in r.output
282 assert "Traceback" not in r.stderr
283
284 def test_ansi_in_json_output_is_encoded(self, tmp_path: pathlib.Path) -> None:
285 """ANSI in remote branch names must be JSON-encoded, not emitted raw."""
286 ansi_branch = "\x1b[31mbad\x1b[0m"
287 info = _make_remote_info(branches={ansi_branch: _FAKE_BARE_OID})
288 r = _lr(tmp_path, _REMOTE_NAME, remote_info=info)
289 assert r.exit_code == 0
290 assert "\x1b" not in r.output
291 d = json.loads(r.output)
292 assert ansi_branch in d["branches"]
293
294
295 # ---------------------------------------------------------------------------
296 # TypedDicts
297 # ---------------------------------------------------------------------------
298
299 class TestTypedDicts:
300 def test_ls_remote_json_typeddict_exists(self) -> None:
301 from muse.cli.commands.ls_remote import _LsRemoteJson
302 assert _LsRemoteJson is not None
303
304 def test_ls_remote_error_json_typeddict_exists(self) -> None:
305 from muse.cli.commands.ls_remote import _LsRemoteErrorJson
306 assert _LsRemoteErrorJson is not None
307
308 def test_ls_remote_json_has_status_annotation(self) -> None:
309 from muse.cli.commands.ls_remote import _LsRemoteJson
310 hints = get_type_hints(_LsRemoteJson)
311 assert "status" in hints
312
313 def test_ls_remote_json_has_all_new_fields(self) -> None:
314 from muse.cli.commands.ls_remote import _LsRemoteJson
315 hints = get_type_hints(_LsRemoteJson)
316 for field in ("status", "error", "remote", "url", "duration_ms", "exit_code"):
317 assert field in hints, f"Missing annotation: {field!r}"
318
319
320 # ---------------------------------------------------------------------------
321 # Docstring coverage
322 # ---------------------------------------------------------------------------
323
324 class TestDocstring:
325 def _doc(self) -> str:
326 import muse.cli.commands.ls_remote as mod
327 return mod.__doc__ or ""
328
329 def test_docstring_documents_status(self) -> None:
330 assert "status" in self._doc()
331
332 def test_docstring_documents_error(self) -> None:
333 assert "error" in self._doc()
334
335 def test_docstring_documents_remote(self) -> None:
336 assert "remote" in self._doc()
337
338 def test_docstring_documents_url(self) -> None:
339 assert "url" in self._doc()
340
341 def test_docstring_documents_duration_ms(self) -> None:
342 assert "duration_ms" in self._doc()
343
344 def test_docstring_documents_exit_code(self) -> None:
345 assert "exit_code" in self._doc()
346
347 def test_docstring_documents_error_schema(self) -> None:
348 assert "error" in self._doc() and "exit_code" in self._doc()
File History 1 commit
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 141 days ago