gabriel / muse public
test_init_supercharge.py python
531 lines 18.9 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 init`` JSON agent-readiness.
2
3 Verifies the enhanced JSON schema that makes ``muse init --json`` fully
4 consumable by agents without defensive ``dict.get`` guards:
5
6 {
7 "status": "ok", // always present; "ok" | "error"
8 "error": "", // always present; non-empty on failure
9 "warnings": [], // always present; symlink-skip notices etc.
10 "repo_id": "<sha256:...>",
11 "branch": "main",
12 "domain": "code",
13 "path": "/abs/.muse",
14 "reinitialized": false,
15 "bare": false,
16 "schema_version": 1,
17 "created_at": "2026-...", // ISO 8601 UTC
18 "duration_ms": 0.0, // wall-clock time for the init
19 "exit_code": 0
20 }
21
22 Error payloads also carry consistent shape:
23
24 {
25 "status": "error",
26 "error": "<message>",
27 "warnings": [],
28 "exit_code": 1
29 }
30 """
31
32 from __future__ import annotations
33 from collections.abc import Mapping
34
35 import datetime
36 import json
37 import os
38 import pathlib
39
40 import pytest
41
42 from tests.cli_test_helper import CliRunner, InvokeResult
43 from muse.core.paths import muse_dir, repo_json_path
44
45 runner = CliRunner()
46
47
48 # ---------------------------------------------------------------------------
49 # Helpers
50 # ---------------------------------------------------------------------------
51
52
53 def _init(repo: pathlib.Path, *extra_args: str) -> InvokeResult:
54 repo.mkdir(parents=True, exist_ok=True)
55 saved = os.getcwd()
56 try:
57 os.chdir(repo)
58 return runner.invoke(None, ["init", *extra_args])
59 finally:
60 os.chdir(saved)
61
62
63 def _init_json(repo: pathlib.Path, *extra_args: str) -> Mapping[str, object]:
64 result = _init(repo, "--json", *extra_args)
65 assert result.exit_code == 0, f"muse init --json failed: {result.output}"
66 return json.loads(result.output)
67
68
69 def _init_json_fail(repo: pathlib.Path, *extra_args: str) -> tuple[dict, int]:
70 """Invoke muse init expecting non-zero exit; return (payload, exit_code)."""
71 result = _init(repo, "--json", *extra_args)
72 assert result.exit_code != 0, f"Expected failure but got exit 0: {result.output}"
73 return json.loads(result.output), result.exit_code
74
75
76 # ---------------------------------------------------------------------------
77 # TestJsonSchemaAgent
78 #
79 # Every field an agent depends on must be present in every success response,
80 # with the correct type, so agents never need ``dict.get`` guards.
81 # ---------------------------------------------------------------------------
82
83
84 class TestJsonSchemaAgent:
85 REQUIRED_KEYS = {
86 "status",
87 "error",
88 "warnings",
89 "repo_id",
90 "branch",
91 "domain",
92 "path",
93 "reinitialized",
94 "bare",
95 "schema_version",
96 "created_at",
97 "duration_ms",
98 "exit_code",
99 "muse_version",
100 "schema",
101 "timestamp",
102 "remotes",
103 }
104
105 def test_all_required_keys_present_on_fresh_init(
106 self, tmp_path: pathlib.Path
107 ) -> None:
108 data = _init_json(tmp_path / "repo")
109 missing = self.REQUIRED_KEYS - set(data)
110 assert not missing, f"Missing keys in init JSON: {missing}"
111
112 def test_all_required_keys_present_on_reinit(
113 self, tmp_path: pathlib.Path
114 ) -> None:
115 repo = tmp_path / "repo"
116 _init(repo)
117 data = _init_json(repo, "--force")
118 missing = self.REQUIRED_KEYS - set(data)
119 assert not missing, f"Missing keys after --force: {missing}"
120
121 def test_all_required_keys_present_on_bare_init(
122 self, tmp_path: pathlib.Path
123 ) -> None:
124 data = _init_json(tmp_path / "repo", "--bare")
125 missing = self.REQUIRED_KEYS - set(data)
126 assert not missing, f"Missing keys in bare init JSON: {missing}"
127
128 def test_no_extra_unknown_keys(self, tmp_path: pathlib.Path) -> None:
129 data = _init_json(tmp_path / "repo")
130 extra = set(data) - self.REQUIRED_KEYS
131 assert not extra, f"Unexpected extra keys: {extra}"
132
133 def test_field_types_correct(self, tmp_path: pathlib.Path) -> None:
134 data = _init_json(tmp_path / "repo")
135 assert isinstance(data["status"], str)
136 assert isinstance(data["error"], str)
137 assert isinstance(data["warnings"], list)
138 assert isinstance(data["repo_id"], str)
139 assert isinstance(data["branch"], str)
140 assert isinstance(data["domain"], str)
141 assert isinstance(data["path"], str)
142 assert isinstance(data["reinitialized"], bool)
143 assert isinstance(data["bare"], bool)
144 assert isinstance(data["schema_version"], int)
145 assert isinstance(data["created_at"], str)
146 assert isinstance(data["duration_ms"], float)
147 assert isinstance(data["exit_code"], int)
148
149
150 # ---------------------------------------------------------------------------
151 # TestStatusField
152 # ---------------------------------------------------------------------------
153
154
155 class TestStatusField:
156 def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None:
157 data = _init_json(tmp_path / "repo")
158 assert data["status"] == "ok"
159
160 def test_status_ok_on_bare(self, tmp_path: pathlib.Path) -> None:
161 data = _init_json(tmp_path / "repo", "--bare")
162 assert data["status"] == "ok"
163
164 def test_status_ok_on_reinit(self, tmp_path: pathlib.Path) -> None:
165 repo = tmp_path / "repo"
166 _init(repo)
167 data = _init_json(repo, "--force")
168 assert data["status"] == "ok"
169
170 def test_error_field_empty_on_success(self, tmp_path: pathlib.Path) -> None:
171 data = _init_json(tmp_path / "repo")
172 assert data["error"] == ""
173
174 def test_status_error_on_bad_branch(self, tmp_path: pathlib.Path) -> None:
175 data, code = _init_json_fail(tmp_path / "repo", "--default-branch", "bad..branch")
176 assert data["status"] == "error"
177 assert code != 0
178
179 def test_status_error_on_bad_domain(self, tmp_path: pathlib.Path) -> None:
180 data, code = _init_json_fail(tmp_path / "repo", "--domain", "BAD_DOMAIN")
181 assert data["status"] == "error"
182
183 def test_status_error_on_existing_no_force(self, tmp_path: pathlib.Path) -> None:
184 repo = tmp_path / "repo"
185 _init(repo)
186 data, _ = _init_json_fail(repo)
187 assert data["status"] == "error"
188
189
190 # ---------------------------------------------------------------------------
191 # TestErrorPayloadShape
192 #
193 # Error payloads must carry a consistent, agent-parseable shape so agents
194 # never have to guess which fields are present after a non-zero exit.
195 # ---------------------------------------------------------------------------
196
197
198 class TestErrorPayloadShape:
199 ERROR_KEYS = {"status", "error", "warnings", "exit_code"}
200
201 def _assert_error_shape(self, data: Mapping[str, object], code: int) -> None:
202 missing = self.ERROR_KEYS - set(data)
203 assert not missing, f"Error payload missing keys: {missing}"
204 assert data["status"] == "error"
205 assert isinstance(data["error"], str) and data["error"]
206 assert isinstance(data["warnings"], list)
207 assert isinstance(data["exit_code"], int)
208 assert data["exit_code"] == code
209
210 def test_bad_branch_error_shape(self, tmp_path: pathlib.Path) -> None:
211 data, code = _init_json_fail(
212 tmp_path / "repo", "--default-branch", "bad..branch"
213 )
214 self._assert_error_shape(data, code)
215
216 def test_bad_domain_error_shape(self, tmp_path: pathlib.Path) -> None:
217 data, code = _init_json_fail(tmp_path / "repo", "--domain", "BAD!")
218 self._assert_error_shape(data, code)
219
220 def test_already_exists_error_shape(self, tmp_path: pathlib.Path) -> None:
221 repo = tmp_path / "repo"
222 _init(repo)
223 data, code = _init_json_fail(repo)
224 self._assert_error_shape(data, code)
225
226 def test_symlink_template_error_shape(self, tmp_path: pathlib.Path) -> None:
227 tmpl_target = tmp_path / "real_dir"
228 tmpl_target.mkdir()
229 symlink = tmp_path / "link_tmpl"
230 symlink.symlink_to(tmpl_target)
231
232 repo = tmp_path / "repo"
233 data, code = _init_json_fail(repo, "--template", str(symlink))
234 self._assert_error_shape(data, code)
235
236 def test_nonexistent_template_error_shape(self, tmp_path: pathlib.Path) -> None:
237 repo = tmp_path / "repo"
238 data, code = _init_json_fail(repo, "--template", str(tmp_path / "no_such"))
239 self._assert_error_shape(data, code)
240
241
242 # ---------------------------------------------------------------------------
243 # TestExitCodeField
244 # ---------------------------------------------------------------------------
245
246
247 class TestExitCodeField:
248 def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None:
249 data = _init_json(tmp_path / "repo")
250 assert data["exit_code"] == 0
251
252 def test_exit_code_zero_on_bare(self, tmp_path: pathlib.Path) -> None:
253 data = _init_json(tmp_path / "repo", "--bare")
254 assert data["exit_code"] == 0
255
256 def test_exit_code_zero_on_reinit(self, tmp_path: pathlib.Path) -> None:
257 repo = tmp_path / "repo"
258 _init(repo)
259 data = _init_json(repo, "--force")
260 assert data["exit_code"] == 0
261
262 def test_exit_code_nonzero_on_bad_branch(self, tmp_path: pathlib.Path) -> None:
263 data, code = _init_json_fail(
264 tmp_path / "repo", "--default-branch", "bad..branch"
265 )
266 assert data["exit_code"] == code
267 assert data["exit_code"] != 0
268
269 def test_exit_code_matches_process_exit_code(
270 self, tmp_path: pathlib.Path
271 ) -> None:
272 repo = tmp_path / "repo"
273 _init(repo)
274 result = _init(repo, "--json") # no --force → should fail
275 data = json.loads(result.output)
276 assert data["exit_code"] == result.exit_code
277
278
279 # ---------------------------------------------------------------------------
280 # TestWarningsField
281 # ---------------------------------------------------------------------------
282
283
284 class TestWarningsField:
285 def test_warnings_empty_on_clean_init(self, tmp_path: pathlib.Path) -> None:
286 data = _init_json(tmp_path / "repo")
287 assert data["warnings"] == []
288
289 def test_warnings_empty_on_bare_init(self, tmp_path: pathlib.Path) -> None:
290 data = _init_json(tmp_path / "repo", "--bare")
291 assert data["warnings"] == []
292
293 def test_warnings_empty_on_reinit_no_template(
294 self, tmp_path: pathlib.Path
295 ) -> None:
296 repo = tmp_path / "repo"
297 _init(repo)
298 data = _init_json(repo, "--force")
299 assert data["warnings"] == []
300
301 def test_warnings_populated_when_template_has_symlinks(
302 self, tmp_path: pathlib.Path
303 ) -> None:
304 tmpl = tmp_path / "tmpl"
305 tmpl.mkdir()
306 (tmpl / "legit.txt").write_text("hello")
307 (tmpl / "malicious_link").symlink_to("/etc/passwd")
308
309 repo = tmp_path / "repo"
310 data = _init_json(repo, "--template", str(tmpl))
311 assert len(data["warnings"]) >= 1
312 joined = " ".join(data["warnings"])
313 assert "malicious_link" in joined or "symlink" in joined.lower()
314
315 def test_warnings_populated_when_template_has_muse_dir(
316 self, tmp_path: pathlib.Path
317 ) -> None:
318 tmpl = tmp_path / "tmpl"
319 tmpl.mkdir()
320 muse_dir(tmpl).mkdir()
321 (repo_json_path(tmpl)).write_text('{"repo_id": "malicious"}')
322
323 repo = tmp_path / "repo"
324 data = _init_json(repo, "--template", str(tmpl))
325 assert len(data["warnings"]) >= 1
326 joined = " ".join(data["warnings"])
327 assert ".muse" in joined
328
329 def test_warnings_list_always_list_not_null(
330 self, tmp_path: pathlib.Path
331 ) -> None:
332 """warnings must be a list in every code path, never None or absent."""
333 repo = tmp_path / "repo"
334 data = _init_json(repo)
335 assert isinstance(data["warnings"], list)
336
337 def test_multiple_symlinks_produce_multiple_warnings(
338 self, tmp_path: pathlib.Path
339 ) -> None:
340 tmpl = tmp_path / "tmpl"
341 tmpl.mkdir()
342 for i in range(3):
343 (tmpl / f"link_{i}").symlink_to("/etc/passwd")
344
345 repo = tmp_path / "repo"
346 data = _init_json(repo, "--template", str(tmpl))
347 assert len(data["warnings"]) >= 3
348
349
350 # ---------------------------------------------------------------------------
351 # TestDurationMs
352 # ---------------------------------------------------------------------------
353
354
355 class TestDurationMs:
356 def test_duration_ms_present(self, tmp_path: pathlib.Path) -> None:
357 data = _init_json(tmp_path / "repo")
358 assert "duration_ms" in data
359
360 def test_duration_ms_is_non_negative(self, tmp_path: pathlib.Path) -> None:
361 data = _init_json(tmp_path / "repo")
362 assert data["duration_ms"] >= 0.0
363
364 def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None:
365 data = _init_json(tmp_path / "repo")
366 assert isinstance(data["duration_ms"], float)
367
368 def test_no_legacy_timing_keys(self, tmp_path: pathlib.Path) -> None:
369 data = _init_json(tmp_path / "repo")
370 assert "elapsed_ms" not in data
371 assert "elapsed" not in data
372 assert "elapsed_seconds" not in data
373
374
375 # ---------------------------------------------------------------------------
376 # TestCreatedAt
377 # ---------------------------------------------------------------------------
378
379
380 class TestCreatedAt:
381 def test_created_at_present(self, tmp_path: pathlib.Path) -> None:
382 data = _init_json(tmp_path / "repo")
383 assert "created_at" in data
384
385 def test_created_at_is_iso8601_utc(self, tmp_path: pathlib.Path) -> None:
386 data = _init_json(tmp_path / "repo")
387 ts = data["created_at"]
388 # Must parse as a datetime with timezone info
389 dt = datetime.datetime.fromisoformat(ts)
390 assert dt.tzinfo is not None, "created_at must be timezone-aware"
391
392 def test_created_at_is_recent(self, tmp_path: pathlib.Path) -> None:
393 before = datetime.datetime.now(datetime.timezone.utc)
394 data = _init_json(tmp_path / "repo")
395 after = datetime.datetime.now(datetime.timezone.utc)
396 ts = datetime.datetime.fromisoformat(data["created_at"])
397 assert before <= ts <= after
398
399 def test_created_at_changes_on_reinit(self, tmp_path: pathlib.Path) -> None:
400 repo = tmp_path / "repo"
401 d1 = _init_json(repo)
402 import time; time.sleep(0.01)
403 d2 = _init_json(repo, "--force")
404 # New reinit → new timestamp (re-runs init, new created_at)
405 assert d2["created_at"] >= d1["created_at"]
406
407 def test_created_at_matches_repo_json(self, tmp_path: pathlib.Path) -> None:
408 repo = tmp_path / "repo"
409 data = _init_json(repo)
410 stored = json.loads((repo_json_path(repo)).read_text())
411 assert stored["created_at"] == data["created_at"]
412
413
414 # ---------------------------------------------------------------------------
415 # TestTypeDict — no _InitJson / _InitErrorJson TypedDict currently exists;
416 # these tests verify the module exposes them after the supercharge.
417 # ---------------------------------------------------------------------------
418
419
420 class TestInitJsonTypedDict:
421 def test_init_json_typed_dict_exists(self) -> None:
422 import muse.cli.commands.init as m
423 assert hasattr(m, "_InitJson"), (
424 "_InitJson TypedDict must be defined in init.py"
425 )
426
427 def test_init_error_json_typed_dict_exists(self) -> None:
428 import muse.cli.commands.init as m
429 assert hasattr(m, "_InitErrorJson"), (
430 "_InitErrorJson TypedDict must be defined in init.py"
431 )
432
433 def test_init_json_has_status_key(self) -> None:
434 import muse.cli.commands.init as m
435 hints = m._InitJson.__annotations__
436 assert "status" in hints
437
438 def test_init_json_has_exit_code_key(self) -> None:
439 import muse.cli.commands.init as m
440 hints = m._InitJson.__annotations__
441 assert "exit_code" in hints
442
443 def test_init_json_has_warnings_key(self) -> None:
444 import muse.cli.commands.init as m
445 hints = m._InitJson.__annotations__
446 assert "warnings" in hints
447
448 def test_init_json_has_duration_ms_key(self) -> None:
449 import muse.cli.commands.init as m
450 hints = m._InitJson.__annotations__
451 assert "duration_ms" in hints
452
453
454 # ---------------------------------------------------------------------------
455 # TestDocstringSchema
456 #
457 # The module-level docstring is the canonical API contract for agents.
458 # It must document all fields that the supercharge adds.
459 # ---------------------------------------------------------------------------
460
461
462 class TestDocstringSchema:
463 def test_docstring_documents_status_field(self) -> None:
464 import muse.cli.commands.init as m
465 assert "status" in (m.__doc__ or ""), (
466 "Module docstring must document the 'status' field"
467 )
468
469 def test_docstring_documents_exit_code_field(self) -> None:
470 import muse.cli.commands.init as m
471 assert "exit_code" in (m.__doc__ or ""), (
472 "Module docstring must document the 'exit_code' field"
473 )
474
475 def test_docstring_documents_warnings_field(self) -> None:
476 import muse.cli.commands.init as m
477 assert "warnings" in (m.__doc__ or ""), (
478 "Module docstring must document the 'warnings' field"
479 )
480
481 def test_docstring_documents_duration_ms_field(self) -> None:
482 import muse.cli.commands.init as m
483 assert "duration_ms" in (m.__doc__ or ""), (
484 "Module docstring must document the 'duration_ms' field"
485 )
486
487 def test_docstring_documents_created_at_field(self) -> None:
488 import muse.cli.commands.init as m
489 assert "created_at" in (m.__doc__ or ""), (
490 "Module docstring must document the 'created_at' field"
491 )
492
493 def test_docstring_documents_error_schema(self) -> None:
494 import muse.cli.commands.init as m
495 assert "error" in (m.__doc__ or ""), (
496 "Module docstring must document the error payload shape"
497 )
498
499
500 # ---------------------------------------------------------------------------
501 # TestRegisterFlags — argparse-level verification
502 # ---------------------------------------------------------------------------
503
504
505 class TestRegisterFlags:
506 """Verify that register() wires --json / -j correctly."""
507
508 def _make_parser(self) -> "argparse.ArgumentParser":
509 import argparse
510 from muse.cli.commands.init import register
511 ap = argparse.ArgumentParser()
512 subs = ap.add_subparsers()
513 register(subs)
514 return ap
515
516 def test_json_flag_long(self) -> None:
517 ns = self._make_parser().parse_args(["init", "--json"])
518 assert ns.json_out is True
519
520 def test_j_alias(self) -> None:
521 ns = self._make_parser().parse_args(["init", "-j"])
522 assert ns.json_out is True
523
524 def test_default_is_text(self) -> None:
525 ns = self._make_parser().parse_args(["init"])
526 assert ns.json_out is False
527
528 def test_dest_is_json_out(self) -> None:
529 ns = self._make_parser().parse_args(["init", "-j"])
530 assert hasattr(ns, "json_out")
531 assert not hasattr(ns, "fmt")
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 ⚠ 29 days ago