"""Supercharge tests for ``muse init`` JSON agent-readiness. Verifies the enhanced JSON schema that makes ``muse init --json`` fully consumable by agents without defensive ``dict.get`` guards: { "status": "ok", // always present; "ok" | "error" "error": "", // always present; non-empty on failure "warnings": [], // always present; symlink-skip notices etc. "repo_id": "", "branch": "main", "domain": "code", "path": "/abs/.muse", "reinitialized": false, "bare": false, "schema_version": 1, "created_at": "2026-...", // ISO 8601 UTC "duration_ms": 0.0, // wall-clock time for the init "exit_code": 0 } Error payloads also carry consistent shape: { "status": "error", "error": "", "warnings": [], "exit_code": 1 } """ from __future__ import annotations from collections.abc import Mapping import datetime import json import os import pathlib import pytest from tests.cli_test_helper import CliRunner, InvokeResult from muse.core.paths import muse_dir, repo_json_path runner = CliRunner() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _init(repo: pathlib.Path, *extra_args: str) -> InvokeResult: repo.mkdir(parents=True, exist_ok=True) saved = os.getcwd() try: os.chdir(repo) return runner.invoke(None, ["init", *extra_args]) finally: os.chdir(saved) def _init_json(repo: pathlib.Path, *extra_args: str) -> Mapping[str, object]: result = _init(repo, "--json", *extra_args) assert result.exit_code == 0, f"muse init --json failed: {result.output}" return json.loads(result.output) def _init_json_fail(repo: pathlib.Path, *extra_args: str) -> tuple[dict, int]: """Invoke muse init expecting non-zero exit; return (payload, exit_code).""" result = _init(repo, "--json", *extra_args) assert result.exit_code != 0, f"Expected failure but got exit 0: {result.output}" return json.loads(result.output), result.exit_code # --------------------------------------------------------------------------- # TestJsonSchemaAgent # # Every field an agent depends on must be present in every success response, # with the correct type, so agents never need ``dict.get`` guards. # --------------------------------------------------------------------------- class TestJsonSchemaAgent: REQUIRED_KEYS = { "status", "error", "warnings", "repo_id", "branch", "domain", "path", "reinitialized", "bare", "schema_version", "created_at", "duration_ms", "exit_code", "muse_version", "schema", "timestamp", "remotes", } def test_all_required_keys_present_on_fresh_init( self, tmp_path: pathlib.Path ) -> None: data = _init_json(tmp_path / "repo") missing = self.REQUIRED_KEYS - set(data) assert not missing, f"Missing keys in init JSON: {missing}" def test_all_required_keys_present_on_reinit( self, tmp_path: pathlib.Path ) -> None: repo = tmp_path / "repo" _init(repo) data = _init_json(repo, "--force") missing = self.REQUIRED_KEYS - set(data) assert not missing, f"Missing keys after --force: {missing}" def test_all_required_keys_present_on_bare_init( self, tmp_path: pathlib.Path ) -> None: data = _init_json(tmp_path / "repo", "--bare") missing = self.REQUIRED_KEYS - set(data) assert not missing, f"Missing keys in bare init JSON: {missing}" def test_no_extra_unknown_keys(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") extra = set(data) - self.REQUIRED_KEYS assert not extra, f"Unexpected extra keys: {extra}" def test_field_types_correct(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert isinstance(data["status"], str) assert isinstance(data["error"], str) assert isinstance(data["warnings"], list) assert isinstance(data["repo_id"], str) assert isinstance(data["branch"], str) assert isinstance(data["domain"], str) assert isinstance(data["path"], str) assert isinstance(data["reinitialized"], bool) assert isinstance(data["bare"], bool) assert isinstance(data["schema_version"], int) assert isinstance(data["created_at"], str) assert isinstance(data["duration_ms"], float) assert isinstance(data["exit_code"], int) # --------------------------------------------------------------------------- # TestStatusField # --------------------------------------------------------------------------- class TestStatusField: def test_status_ok_on_success(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert data["status"] == "ok" def test_status_ok_on_bare(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo", "--bare") assert data["status"] == "ok" def test_status_ok_on_reinit(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" _init(repo) data = _init_json(repo, "--force") assert data["status"] == "ok" def test_error_field_empty_on_success(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert data["error"] == "" def test_status_error_on_bad_branch(self, tmp_path: pathlib.Path) -> None: data, code = _init_json_fail(tmp_path / "repo", "--default-branch", "bad..branch") assert data["status"] == "error" assert code != 0 def test_status_error_on_bad_domain(self, tmp_path: pathlib.Path) -> None: data, code = _init_json_fail(tmp_path / "repo", "--domain", "BAD_DOMAIN") assert data["status"] == "error" def test_status_error_on_existing_no_force(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" _init(repo) data, _ = _init_json_fail(repo) assert data["status"] == "error" # --------------------------------------------------------------------------- # TestErrorPayloadShape # # Error payloads must carry a consistent, agent-parseable shape so agents # never have to guess which fields are present after a non-zero exit. # --------------------------------------------------------------------------- class TestErrorPayloadShape: ERROR_KEYS = {"status", "error", "warnings", "exit_code"} def _assert_error_shape(self, data: Mapping[str, object], code: int) -> None: missing = self.ERROR_KEYS - set(data) assert not missing, f"Error payload missing keys: {missing}" assert data["status"] == "error" assert isinstance(data["error"], str) and data["error"] assert isinstance(data["warnings"], list) assert isinstance(data["exit_code"], int) assert data["exit_code"] == code def test_bad_branch_error_shape(self, tmp_path: pathlib.Path) -> None: data, code = _init_json_fail( tmp_path / "repo", "--default-branch", "bad..branch" ) self._assert_error_shape(data, code) def test_bad_domain_error_shape(self, tmp_path: pathlib.Path) -> None: data, code = _init_json_fail(tmp_path / "repo", "--domain", "BAD!") self._assert_error_shape(data, code) def test_already_exists_error_shape(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" _init(repo) data, code = _init_json_fail(repo) self._assert_error_shape(data, code) def test_symlink_template_error_shape(self, tmp_path: pathlib.Path) -> None: tmpl_target = tmp_path / "real_dir" tmpl_target.mkdir() symlink = tmp_path / "link_tmpl" symlink.symlink_to(tmpl_target) repo = tmp_path / "repo" data, code = _init_json_fail(repo, "--template", str(symlink)) self._assert_error_shape(data, code) def test_nonexistent_template_error_shape(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" data, code = _init_json_fail(repo, "--template", str(tmp_path / "no_such")) self._assert_error_shape(data, code) # --------------------------------------------------------------------------- # TestExitCodeField # --------------------------------------------------------------------------- class TestExitCodeField: def test_exit_code_zero_on_success(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert data["exit_code"] == 0 def test_exit_code_zero_on_bare(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo", "--bare") assert data["exit_code"] == 0 def test_exit_code_zero_on_reinit(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" _init(repo) data = _init_json(repo, "--force") assert data["exit_code"] == 0 def test_exit_code_nonzero_on_bad_branch(self, tmp_path: pathlib.Path) -> None: data, code = _init_json_fail( tmp_path / "repo", "--default-branch", "bad..branch" ) assert data["exit_code"] == code assert data["exit_code"] != 0 def test_exit_code_matches_process_exit_code( self, tmp_path: pathlib.Path ) -> None: repo = tmp_path / "repo" _init(repo) result = _init(repo, "--json") # no --force → should fail data = json.loads(result.output) assert data["exit_code"] == result.exit_code # --------------------------------------------------------------------------- # TestWarningsField # --------------------------------------------------------------------------- class TestWarningsField: def test_warnings_empty_on_clean_init(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert data["warnings"] == [] def test_warnings_empty_on_bare_init(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo", "--bare") assert data["warnings"] == [] def test_warnings_empty_on_reinit_no_template( self, tmp_path: pathlib.Path ) -> None: repo = tmp_path / "repo" _init(repo) data = _init_json(repo, "--force") assert data["warnings"] == [] def test_warnings_populated_when_template_has_symlinks( self, tmp_path: pathlib.Path ) -> None: tmpl = tmp_path / "tmpl" tmpl.mkdir() (tmpl / "legit.txt").write_text("hello") (tmpl / "malicious_link").symlink_to("/etc/passwd") repo = tmp_path / "repo" data = _init_json(repo, "--template", str(tmpl)) assert len(data["warnings"]) >= 1 joined = " ".join(data["warnings"]) assert "malicious_link" in joined or "symlink" in joined.lower() def test_warnings_populated_when_template_has_muse_dir( self, tmp_path: pathlib.Path ) -> None: tmpl = tmp_path / "tmpl" tmpl.mkdir() muse_dir(tmpl).mkdir() (repo_json_path(tmpl)).write_text('{"repo_id": "malicious"}') repo = tmp_path / "repo" data = _init_json(repo, "--template", str(tmpl)) assert len(data["warnings"]) >= 1 joined = " ".join(data["warnings"]) assert ".muse" in joined def test_warnings_list_always_list_not_null( self, tmp_path: pathlib.Path ) -> None: """warnings must be a list in every code path, never None or absent.""" repo = tmp_path / "repo" data = _init_json(repo) assert isinstance(data["warnings"], list) def test_multiple_symlinks_produce_multiple_warnings( self, tmp_path: pathlib.Path ) -> None: tmpl = tmp_path / "tmpl" tmpl.mkdir() for i in range(3): (tmpl / f"link_{i}").symlink_to("/etc/passwd") repo = tmp_path / "repo" data = _init_json(repo, "--template", str(tmpl)) assert len(data["warnings"]) >= 3 # --------------------------------------------------------------------------- # TestDurationMs # --------------------------------------------------------------------------- class TestDurationMs: def test_duration_ms_present(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert "duration_ms" in data def test_duration_ms_is_non_negative(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert data["duration_ms"] >= 0.0 def test_duration_ms_is_float(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert isinstance(data["duration_ms"], float) def test_no_legacy_timing_keys(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert "elapsed_ms" not in data assert "elapsed" not in data assert "elapsed_seconds" not in data # --------------------------------------------------------------------------- # TestCreatedAt # --------------------------------------------------------------------------- class TestCreatedAt: def test_created_at_present(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") assert "created_at" in data def test_created_at_is_iso8601_utc(self, tmp_path: pathlib.Path) -> None: data = _init_json(tmp_path / "repo") ts = data["created_at"] # Must parse as a datetime with timezone info dt = datetime.datetime.fromisoformat(ts) assert dt.tzinfo is not None, "created_at must be timezone-aware" def test_created_at_is_recent(self, tmp_path: pathlib.Path) -> None: before = datetime.datetime.now(datetime.timezone.utc) data = _init_json(tmp_path / "repo") after = datetime.datetime.now(datetime.timezone.utc) ts = datetime.datetime.fromisoformat(data["created_at"]) assert before <= ts <= after def test_created_at_changes_on_reinit(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" d1 = _init_json(repo) import time; time.sleep(0.01) d2 = _init_json(repo, "--force") # New reinit → new timestamp (re-runs init, new created_at) assert d2["created_at"] >= d1["created_at"] def test_created_at_matches_repo_json(self, tmp_path: pathlib.Path) -> None: repo = tmp_path / "repo" data = _init_json(repo) stored = json.loads((repo_json_path(repo)).read_text()) assert stored["created_at"] == data["created_at"] # --------------------------------------------------------------------------- # TestTypeDict — no _InitJson / _InitErrorJson TypedDict currently exists; # these tests verify the module exposes them after the supercharge. # --------------------------------------------------------------------------- class TestInitJsonTypedDict: def test_init_json_typed_dict_exists(self) -> None: import muse.cli.commands.init as m assert hasattr(m, "_InitJson"), ( "_InitJson TypedDict must be defined in init.py" ) def test_init_error_json_typed_dict_exists(self) -> None: import muse.cli.commands.init as m assert hasattr(m, "_InitErrorJson"), ( "_InitErrorJson TypedDict must be defined in init.py" ) def test_init_json_has_status_key(self) -> None: import muse.cli.commands.init as m hints = m._InitJson.__annotations__ assert "status" in hints def test_init_json_has_exit_code_key(self) -> None: import muse.cli.commands.init as m hints = m._InitJson.__annotations__ assert "exit_code" in hints def test_init_json_has_warnings_key(self) -> None: import muse.cli.commands.init as m hints = m._InitJson.__annotations__ assert "warnings" in hints def test_init_json_has_duration_ms_key(self) -> None: import muse.cli.commands.init as m hints = m._InitJson.__annotations__ assert "duration_ms" in hints # --------------------------------------------------------------------------- # TestDocstringSchema # # The module-level docstring is the canonical API contract for agents. # It must document all fields that the supercharge adds. # --------------------------------------------------------------------------- class TestDocstringSchema: def test_docstring_documents_status_field(self) -> None: import muse.cli.commands.init as m assert "status" in (m.__doc__ or ""), ( "Module docstring must document the 'status' field" ) def test_docstring_documents_exit_code_field(self) -> None: import muse.cli.commands.init as m assert "exit_code" in (m.__doc__ or ""), ( "Module docstring must document the 'exit_code' field" ) def test_docstring_documents_warnings_field(self) -> None: import muse.cli.commands.init as m assert "warnings" in (m.__doc__ or ""), ( "Module docstring must document the 'warnings' field" ) def test_docstring_documents_duration_ms_field(self) -> None: import muse.cli.commands.init as m assert "duration_ms" in (m.__doc__ or ""), ( "Module docstring must document the 'duration_ms' field" ) def test_docstring_documents_created_at_field(self) -> None: import muse.cli.commands.init as m assert "created_at" in (m.__doc__ or ""), ( "Module docstring must document the 'created_at' field" ) def test_docstring_documents_error_schema(self) -> None: import muse.cli.commands.init as m assert "error" in (m.__doc__ or ""), ( "Module docstring must document the error payload shape" ) # --------------------------------------------------------------------------- # TestRegisterFlags — argparse-level verification # --------------------------------------------------------------------------- class TestRegisterFlags: """Verify that register() wires --json / -j correctly.""" def _make_parser(self) -> "argparse.ArgumentParser": import argparse from muse.cli.commands.init import register ap = argparse.ArgumentParser() subs = ap.add_subparsers() register(subs) return ap def test_json_flag_long(self) -> None: ns = self._make_parser().parse_args(["init", "--json"]) assert ns.json_out is True def test_j_alias(self) -> None: ns = self._make_parser().parse_args(["init", "-j"]) assert ns.json_out is True def test_default_is_text(self) -> None: ns = self._make_parser().parse_args(["init"]) assert ns.json_out is False def test_dest_is_json_out(self) -> None: ns = self._make_parser().parse_args(["init", "-j"]) assert hasattr(ns, "json_out") assert not hasattr(ns, "fmt")