"""Supercharge tests for ``muse agent``, ``muse agent-config``, and ``muse agent-map``. Seven-Tier Coverage Matrix -------------------------- Tier 1 — TestTypedDictAgent _KeygenJson, _RegisterJson, _ListJson must have schema_version, exit_code, duration_ms annotations. _ListJson must exist (does not yet). Tier 2 — TestTypedDictAgentConfig _InitJson, _SyncJson, _ReadJson, _StatusJson, _InspectJson, _SetJson must exist and carry schema_version, exit_code, duration_ms. Tier 3 — TestTypedDictAgentMap _AgentMapJson must exist and carry schema_version, exit_code, duration_ms, mode. Tier 4 — TestUnitEmitError _emit_error json/human paths, error/message keys, always exits 1. Tier 5 — TestAliasRegistration -j alias present on agent keygen, agent list, agent register, agent-map. Tier 6 — TestDocstrings run_list, run_keygen, run_register mention schema_version; register() docstring for agent mentions -j. Tier 7 — TestEndToEnd / TestStress / TestDataIntegrity / TestSecurity / TestPerformance Live CLI invocations, stress runs, type invariants, hostile input survival, timing bounds. All tests should be RED until the implementation adds the missing TypedDicts, envelope fields, and -j aliases described in the task. """ from __future__ import annotations import argparse import io import json import os import pathlib import sys import threading import time from typing import get_type_hints import pytest from tests.cli_test_helper import CliRunner, InvokeResult runner = CliRunner() # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _TEST_HUB = "http://test.example.com" _TEST_MNEMONIC = ( "abandon abandon abandon abandon abandon abandon abandon abandon " "abandon abandon abandon about" ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _invoke(repo: pathlib.Path, args: list[str]) -> InvokeResult: """Invoke runner with CWD set to *repo*.""" saved = os.getcwd() try: os.chdir(repo) return runner.invoke(None, args) finally: os.chdir(saved) # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def cfg_repo(tmp_path: pathlib.Path) -> pathlib.Path: """Minimal muse repo suitable for agent-config tests.""" saved = os.getcwd() try: os.chdir(tmp_path) r = runner.invoke(None, ["init"]) assert r.exit_code == 0, f"init failed: {r.output}" finally: os.chdir(saved) return tmp_path @pytest.fixture() def identity_repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Repo with a synthetic identity entry for agent keygen/register tests.""" saved = os.getcwd() try: os.chdir(tmp_path) r = runner.invoke(None, ["init"]) assert r.exit_code == 0, f"init failed: {r.output}" finally: os.chdir(saved) # Inject a fake identity so _require_mnemonic does not bail identity_dir = pathlib.Path.home() / ".muse" identity_dir.mkdir(parents=True, exist_ok=True) identity_file = identity_dir / "identity.toml" # Back up existing identity if present backup: bytes | None = None if identity_file.exists(): backup = identity_file.read_bytes() monkeypatch.setattr( "muse.core.identity.load_identity", lambda url: {"mnemonic": _TEST_MNEMONIC, "handle": "test-agent"}, ) yield tmp_path # Restore backup if backup is not None: identity_file.write_bytes(backup) # =========================================================================== # Tier 1 — TestTypedDictAgent # =========================================================================== class TestTypedDictAgent: """_KeygenJson/_RegisterJson/_ListJson must carry envelope fields.""" # ── _KeygenJson ────────────────────────────────────────────────────────── def test_keygen_json_has_schema_version(self) -> None: from muse.cli.commands.agent import _KeygenJson assert "schema" in get_type_hints(_KeygenJson) def test_keygen_json_has_exit_code(self) -> None: from muse.cli.commands.agent import _KeygenJson assert "exit_code" in get_type_hints(_KeygenJson) def test_keygen_json_has_duration_ms(self) -> None: from muse.cli.commands.agent import _KeygenJson assert "duration_ms" in get_type_hints(_KeygenJson) def test_keygen_json_existing_fields_preserved(self) -> None: from muse.cli.commands.agent import _KeygenJson hints = get_type_hints(_KeygenJson) for field in ("status", "hub", "account", "msign_path", "public_key_b64", "fingerprint", "hd_seed_b64"): assert field in hints, f"_KeygenJson missing existing field: {field}" # ── _RegisterJson ──────────────────────────────────────────────────────── def test_register_json_has_schema_version(self) -> None: from muse.cli.commands.agent import _RegisterJson assert "schema" in get_type_hints(_RegisterJson) def test_register_json_has_exit_code(self) -> None: from muse.cli.commands.agent import _RegisterJson assert "exit_code" in get_type_hints(_RegisterJson) def test_register_json_has_duration_ms(self) -> None: from muse.cli.commands.agent import _RegisterJson assert "duration_ms" in get_type_hints(_RegisterJson) def test_register_json_existing_fields_preserved(self) -> None: from muse.cli.commands.agent import _RegisterJson hints = get_type_hints(_RegisterJson) for field in ("status", "name", "account", "hub", "msign_path"): assert field in hints, f"_RegisterJson missing existing field: {field}" # ── _ListJson ──────────────────────────────────────────────────────────── def test_list_json_exists(self) -> None: """_ListJson TypedDict must be importable — does not yet exist.""" from muse.cli.commands.agent import _ListJson # noqa: F401 def test_list_json_has_schema_version(self) -> None: from muse.cli.commands.agent import _ListJson assert "schema" in get_type_hints(_ListJson) def test_list_json_has_exit_code(self) -> None: from muse.cli.commands.agent import _ListJson assert "exit_code" in get_type_hints(_ListJson) def test_list_json_has_duration_ms(self) -> None: from muse.cli.commands.agent import _ListJson assert "duration_ms" in get_type_hints(_ListJson) def test_list_json_has_slots_field(self) -> None: from muse.cli.commands.agent import _ListJson assert "slots" in get_type_hints(_ListJson) def test_list_json_has_mode_field(self) -> None: from muse.cli.commands.agent import _ListJson assert "mode" in get_type_hints(_ListJson) # =========================================================================== # Tier 2 — TestTypedDictAgentConfig # =========================================================================== class TestTypedDictAgentConfig: """_InitJson/_SyncJson/_ReadJson/_StatusJson/_InspectJson/_SetJson must exist.""" # ── _InitJson ───────────────────────────────────────────────────────────── def test_init_json_exists(self) -> None: from muse.cli.commands.agent_config import _InitJson # noqa: F401 def test_init_json_has_schema_version(self) -> None: from muse.cli.commands.agent_config import _InitJson assert "schema" in get_type_hints(_InitJson) def test_init_json_has_exit_code(self) -> None: from muse.cli.commands.agent_config import _InitJson assert "exit_code" in get_type_hints(_InitJson) def test_init_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_config import _InitJson assert "duration_ms" in get_type_hints(_InitJson) def test_init_json_has_path_field(self) -> None: from muse.cli.commands.agent_config import _InitJson assert "path" in get_type_hints(_InitJson) # ── _SyncJson ───────────────────────────────────────────────────────────── def test_sync_json_exists(self) -> None: from muse.cli.commands.agent_config import _SyncJson # noqa: F401 def test_sync_json_has_schema_version(self) -> None: from muse.cli.commands.agent_config import _SyncJson assert "schema" in get_type_hints(_SyncJson) def test_sync_json_has_exit_code(self) -> None: from muse.cli.commands.agent_config import _SyncJson assert "exit_code" in get_type_hints(_SyncJson) def test_sync_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_config import _SyncJson assert "duration_ms" in get_type_hints(_SyncJson) def test_sync_json_has_adapters_field(self) -> None: from muse.cli.commands.agent_config import _SyncJson assert "adapters" in get_type_hints(_SyncJson) # ── _ReadJson ───────────────────────────────────────────────────────────── def test_read_json_exists(self) -> None: from muse.cli.commands.agent_config import _ReadJson # noqa: F401 def test_read_json_has_schema_version(self) -> None: from muse.cli.commands.agent_config import _ReadJson assert "schema" in get_type_hints(_ReadJson) def test_read_json_has_exit_code(self) -> None: from muse.cli.commands.agent_config import _ReadJson assert "exit_code" in get_type_hints(_ReadJson) def test_read_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_config import _ReadJson assert "duration_ms" in get_type_hints(_ReadJson) def test_read_json_has_content_field(self) -> None: from muse.cli.commands.agent_config import _ReadJson assert "content" in get_type_hints(_ReadJson) # ── _StatusJson ─────────────────────────────────────────────────────────── def test_status_json_exists(self) -> None: from muse.cli.commands.agent_config import _StatusJson # noqa: F401 def test_status_json_has_schema_version(self) -> None: from muse.cli.commands.agent_config import _StatusJson assert "schema" in get_type_hints(_StatusJson) def test_status_json_has_exit_code(self) -> None: from muse.cli.commands.agent_config import _StatusJson assert "exit_code" in get_type_hints(_StatusJson) def test_status_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_config import _StatusJson assert "duration_ms" in get_type_hints(_StatusJson) def test_status_json_has_ready_field(self) -> None: from muse.cli.commands.agent_config import _StatusJson assert "ready" in get_type_hints(_StatusJson) # ── _InspectJson ────────────────────────────────────────────────────────── def test_inspect_json_exists(self) -> None: from muse.cli.commands.agent_config import _InspectJson # noqa: F401 def test_inspect_json_has_schema_version(self) -> None: from muse.cli.commands.agent_config import _InspectJson assert "schema" in get_type_hints(_InspectJson) def test_inspect_json_has_exit_code(self) -> None: from muse.cli.commands.agent_config import _InspectJson assert "exit_code" in get_type_hints(_InspectJson) def test_inspect_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_config import _InspectJson assert "duration_ms" in get_type_hints(_InspectJson) def test_inspect_json_has_context_field(self) -> None: from muse.cli.commands.agent_config import _InspectJson assert "context" in get_type_hints(_InspectJson) # ── _SetJson ────────────────────────────────────────────────────────────── def test_set_json_exists(self) -> None: from muse.cli.commands.agent_config import _SetJson # noqa: F401 def test_set_json_has_schema_version(self) -> None: from muse.cli.commands.agent_config import _SetJson assert "schema" in get_type_hints(_SetJson) def test_set_json_has_exit_code(self) -> None: from muse.cli.commands.agent_config import _SetJson assert "exit_code" in get_type_hints(_SetJson) def test_set_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_config import _SetJson assert "duration_ms" in get_type_hints(_SetJson) def test_set_json_has_adapters_field(self) -> None: from muse.cli.commands.agent_config import _SetJson assert "adapters" in get_type_hints(_SetJson) # =========================================================================== # Tier 3 — TestTypedDictAgentMap # =========================================================================== class TestTypedDictAgentMap: """_AgentMapJson must exist and carry envelope fields.""" def test_agent_map_json_exists(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson # noqa: F401 def test_agent_map_json_has_schema_version(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson assert "schema" in get_type_hints(_AgentMapJson) def test_agent_map_json_has_exit_code(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson assert "exit_code" in get_type_hints(_AgentMapJson) def test_agent_map_json_has_duration_ms(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson assert "duration_ms" in get_type_hints(_AgentMapJson) def test_agent_map_json_has_mode(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson assert "mode" in get_type_hints(_AgentMapJson) def test_agent_map_json_has_track(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson assert "track" in get_type_hints(_AgentMapJson) def test_agent_map_json_has_attributions(self) -> None: from muse.cli.commands.agent_map import _AgentMapJson assert "attributions" in get_type_hints(_AgentMapJson) def test_bar_attribution_existing_fields_preserved(self) -> None: from muse.cli.commands.agent_map import BarAttribution hints = get_type_hints(BarAttribution) for field in ("bar", "author", "commit_id", "message"): assert field in hints, f"BarAttribution missing field: {field}" # =========================================================================== # Tier 4 — TestUnitFingerprint / TestUnitEmitError # =========================================================================== class TestUnitEmitError: """_emit_error json/human paths.""" def _capture_emit_error(self, error: str, message: str, as_json: bool) -> tuple[str, str, int]: """Run _emit_error, return (stdout_text, stderr_text, exit_code).""" from muse.cli.commands.agent import _emit_error stdout_buf = io.StringIO() stderr_buf = io.StringIO() exit_code = 0 orig_stdout, orig_stderr = sys.stdout, sys.stderr sys.stdout = stdout_buf sys.stderr = stderr_buf try: _emit_error(error, message, as_json) except SystemExit as exc: exit_code = int(exc.code) if exc.code is not None else 0 finally: sys.stdout = orig_stdout sys.stderr = orig_stderr return stdout_buf.getvalue(), stderr_buf.getvalue(), exit_code def test_json_mode_writes_to_stdout(self) -> None: stdout, _, _ = self._capture_emit_error("err_code", "msg text", True) assert stdout.strip() != "" def test_json_mode_valid_json(self) -> None: stdout, _, _ = self._capture_emit_error("err_code", "msg text", True) json.loads(stdout) # must not raise def test_json_mode_has_error_key(self) -> None: stdout, _, _ = self._capture_emit_error("err_code", "msg text", True) d = json.loads(stdout) assert "error" in d def test_json_mode_has_message_key(self) -> None: stdout, _, _ = self._capture_emit_error("err_code", "msg text", True) d = json.loads(stdout) assert "message" in d def test_json_mode_error_value_correct(self) -> None: stdout, _, _ = self._capture_emit_error("my_error", "some message", True) d = json.loads(stdout) assert d["error"] == "my_error" def test_json_mode_message_value_correct(self) -> None: stdout, _, _ = self._capture_emit_error("err", "expected message", True) d = json.loads(stdout) assert d["message"] == "expected message" def test_json_mode_exits_1(self) -> None: _, _, code = self._capture_emit_error("err", "msg", True) assert code == 1 def test_human_mode_writes_to_stderr(self) -> None: _, stderr, _ = self._capture_emit_error("err", "human message", False) assert stderr.strip() != "" def test_human_mode_exits_1(self) -> None: _, _, code = self._capture_emit_error("err", "msg", False) assert code == 1 def test_human_mode_stdout_empty(self) -> None: stdout, _, _ = self._capture_emit_error("err", "msg", False) assert stdout.strip() == "" # =========================================================================== # Tier 5 — TestAliasRegistration # =========================================================================== class TestAliasRegistration: """``-j`` alias must be registered on all three agent subcommands and agent-map.""" def _build_agent_parser(self) -> argparse.ArgumentParser: from muse.cli.commands.agent import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p def test_agent_keygen_j_alias_parses(self) -> None: p = self._build_agent_parser() ns = p.parse_args(["agent", "keygen", "--account", "0", "--hub", _TEST_HUB, "-j"]) assert getattr(ns, "json_out", False) is True def test_agent_list_j_alias_parses(self) -> None: p = self._build_agent_parser() ns = p.parse_args(["agent", "list", "--hub", _TEST_HUB, "-j"]) assert getattr(ns, "json_out", False) is True def test_agent_register_j_alias_parses(self) -> None: p = self._build_agent_parser() ns = p.parse_args([ "agent", "register", "--account", "1", "--name", "test-agent", "--hub", _TEST_HUB, "-j", ]) assert getattr(ns, "json_out", False) is True def test_agent_map_j_alias_registered(self) -> None: from muse.cli.commands.agent_map import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) ns = p.parse_args(["agent-map", "tracks/test.mid", "-j"]) # -j should set json_out/json/as_json to True assert ( getattr(ns, "json_out", False) is True or getattr(ns, "json", False) is True or getattr(ns, "as_json", False) is True ) def test_agent_keygen_json_flag_still_works(self) -> None: p = self._build_agent_parser() ns = p.parse_args(["agent", "keygen", "--account", "0", "--hub", _TEST_HUB, "--json"]) assert getattr(ns, "json_out", False) is True def test_agent_list_json_flag_still_works(self) -> None: p = self._build_agent_parser() ns = p.parse_args(["agent", "list", "--hub", _TEST_HUB, "--json"]) assert getattr(ns, "json_out", False) is True def test_agent_register_json_flag_still_works(self) -> None: p = self._build_agent_parser() ns = p.parse_args([ "agent", "register", "--account", "1", "--name", "test-agent", "--hub", _TEST_HUB, "--json", ]) assert getattr(ns, "json_out", False) is True # =========================================================================== # Tier 6 — TestDocstrings # =========================================================================== class TestDocstrings: """Key functions must document new envelope fields in their docstrings.""" def test_run_keygen_docstring_mentions_schema_version(self) -> None: from muse.cli.commands.agent import run_keygen assert run_keygen.__doc__ is not None assert "schema" in run_keygen.__doc__ def test_run_list_docstring_mentions_schema_version(self) -> None: from muse.cli.commands.agent import run_list assert run_list.__doc__ is not None assert "schema" in run_list.__doc__ def test_run_register_docstring_mentions_schema_version(self) -> None: from muse.cli.commands.agent import run_register assert run_register.__doc__ is not None assert "schema" in run_register.__doc__ def test_register_docstring_mentions_j_alias(self) -> None: from muse.cli.commands.agent import register assert register.__doc__ is not None assert "-j" in register.__doc__ def test_run_list_docstring_mentions_duration_ms(self) -> None: from muse.cli.commands.agent import run_list assert run_list.__doc__ is not None assert "duration_ms" in run_list.__doc__ def test_run_keygen_docstring_mentions_duration_ms(self) -> None: from muse.cli.commands.agent import run_keygen assert run_keygen.__doc__ is not None assert "duration_ms" in run_keygen.__doc__ # =========================================================================== # Tier 7a — TestEndToEnd # =========================================================================== class TestEndToEnd: """Live CLI invocations against real (tmp) repos.""" # ── agent list -j ──────────────────────────────────────────────────────── def test_agent_list_j_exits_zero(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) assert r.exit_code == 0, r.output def test_agent_list_j_valid_json(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) assert r.exit_code == 0, r.output json.loads(r.output) def test_agent_list_j_has_schema_version(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_agent_list_j_has_exit_code(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) d = json.loads(r.output) assert "exit_code" in d def test_agent_list_j_has_duration_ms(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) d = json.loads(r.output) assert "duration_ms" in d def test_agent_list_j_has_slots(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) d = json.loads(r.output) assert "slots" in d def test_agent_list_j_slots_is_list(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) d = json.loads(r.output) assert isinstance(d["slots"], list) def test_agent_list_j_exit_code_matches_process(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) d = json.loads(r.output) assert d["exit_code"] == r.exit_code def test_agent_list_json_flag_same_as_j(self) -> None: r1 = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) r2 = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "-j"]) d1 = json.loads(r1.output) d2 = json.loads(r2.output) d1.pop("duration_ms", None) d2.pop("duration_ms", None) assert set(d1.keys()) == set(d2.keys()) # ── agent-config init -j ────────────────────────────────────────────────── def test_agent_config_init_j_exits_zero(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "-j"]) assert r.exit_code == 0, r.output def test_agent_config_init_j_valid_json(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "-j"]) assert r.exit_code == 0, r.output json.loads(r.output) def test_agent_config_init_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "-j"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_agent_config_init_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "-j"]) d = json.loads(r.output) assert "exit_code" in d def test_agent_config_init_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "-j"]) d = json.loads(r.output) assert "duration_ms" in d # ── agent-config status -j ──────────────────────────────────────────────── def test_agent_config_status_j_exits_zero(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "status", "-j"]) assert r.exit_code == 0, r.output def test_agent_config_status_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "status", "-j"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_agent_config_status_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "status", "-j"]) d = json.loads(r.output) assert "exit_code" in d def test_agent_config_status_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "status", "-j"]) d = json.loads(r.output) assert "duration_ms" in d # ── agent-config inspect -j ──────────────────────────────────────────────── def test_agent_config_inspect_j_exits_zero(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"]) assert r.exit_code == 0, r.output def test_agent_config_inspect_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_agent_config_inspect_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"]) d = json.loads(r.output) assert "exit_code" in d def test_agent_config_inspect_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init", "-j"]) r = _invoke(cfg_repo, ["agent-config", "inspect", "-j"]) d = json.loads(r.output) assert "duration_ms" in d # ── agent-config read -j (after init) ──────────────────────────────────── def test_agent_config_read_j_exits_zero(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) r = _invoke(cfg_repo, ["agent-config", "read", "-j"]) assert r.exit_code == 0, r.output def test_agent_config_read_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) r = _invoke(cfg_repo, ["agent-config", "read", "-j"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_agent_config_read_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) r = _invoke(cfg_repo, ["agent-config", "read", "-j"]) d = json.loads(r.output) assert "exit_code" in d def test_agent_config_read_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) r = _invoke(cfg_repo, ["agent-config", "read", "-j"]) d = json.loads(r.output) assert "duration_ms" in d # ── agent-config sync -j (after init + set) ────────────────────────────── def test_agent_config_sync_j_exits_zero(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) _invoke(cfg_repo, ["agent-config", "set", "--adapters", "claude"]) r = _invoke(cfg_repo, ["agent-config", "sync", "-j"]) assert r.exit_code == 0, r.output def test_agent_config_sync_j_has_schema_version(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) _invoke(cfg_repo, ["agent-config", "set", "--adapters", "claude"]) r = _invoke(cfg_repo, ["agent-config", "sync", "-j"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_agent_config_sync_j_has_exit_code(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) _invoke(cfg_repo, ["agent-config", "set", "--adapters", "claude"]) r = _invoke(cfg_repo, ["agent-config", "sync", "-j"]) d = json.loads(r.output) assert "exit_code" in d def test_agent_config_sync_j_has_duration_ms(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) _invoke(cfg_repo, ["agent-config", "set", "--adapters", "claude"]) r = _invoke(cfg_repo, ["agent-config", "sync", "-j"]) d = json.loads(r.output) assert "duration_ms" in d # =========================================================================== # Tier 7b — TestStress # =========================================================================== class TestStress: """High-volume calls to verify no crashes, leaks, or race conditions.""" def test_fingerprint_1000_calls_no_crash(self) -> None: from muse.core.types import public_key_fingerprint for i in range(1000): result = public_key_fingerprint(i.to_bytes(4, "big")) assert len(result) == 71 def test_emit_error_json_500_calls(self) -> None: from muse.cli.commands.agent import _emit_error for i in range(500): stdout_buf = io.StringIO() orig = sys.stdout sys.stdout = stdout_buf try: try: _emit_error(f"err_{i}", f"message_{i}", True) except SystemExit: pass finally: sys.stdout = orig d = json.loads(stdout_buf.getvalue()) assert d["error"] == f"err_{i}" def test_agent_config_status_50_times(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) for _ in range(50): r = _invoke(cfg_repo, ["agent-config", "status", "--json"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert "schema" in d def test_fingerprint_thread_safety(self) -> None: from muse.core.types import public_key_fingerprint results: list[str] = [] errors: list[Exception] = [] lock = threading.Lock() def worker(n: int) -> None: try: fp = public_key_fingerprint(n.to_bytes(4, "big")) with lock: results.append(fp) except Exception as exc: with lock: errors.append(exc) threads = [threading.Thread(target=worker, args=(i,)) for i in range(100)] for t in threads: t.start() for t in threads: t.join() assert not errors, f"Thread errors: {errors}" assert len(results) == 100 # =========================================================================== # Tier 7c — TestDataIntegrity # =========================================================================== class TestDataIntegrity: """Type invariants that the JSON envelope must satisfy.""" def test_agent_list_schema_version_is_str(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert isinstance(d["schema"], int) def test_agent_list_exit_code_is_int(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) d = json.loads(r.output) assert isinstance(d["exit_code"], int) def test_agent_list_duration_ms_is_float(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) d = json.loads(r.output) assert isinstance(d["duration_ms"], float) def test_agent_list_slots_is_list(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) d = json.loads(r.output) assert isinstance(d["slots"], list) def test_agent_list_duration_ms_nonnegative(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) d = json.loads(r.output) assert d["duration_ms"] >= 0.0 def test_agent_config_init_schema_version_is_str(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "--json"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert isinstance(d["schema"], int) def test_agent_config_init_exit_code_is_int(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "--json"]) d = json.loads(r.output) assert isinstance(d["exit_code"], int) def test_agent_config_init_duration_ms_is_float(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "--json"]) d = json.loads(r.output) assert isinstance(d["duration_ms"], float) def test_agent_config_status_exit_code_mirrors_process(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) r = _invoke(cfg_repo, ["agent-config", "status", "--json"]) d = json.loads(r.output) assert d["exit_code"] == r.exit_code def test_agent_list_schema_version_nonempty(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) d = json.loads(r.output) assert d["schema"] > 0 # =========================================================================== # Tier 7d — TestSecurity # =========================================================================== class TestSecurity: """Hostile input must not crash or corrupt JSON output.""" def test_hostile_hub_url_in_emit_error_json(self) -> None: from muse.cli.commands.agent import _emit_error hostile = "'; DROP TABLE users; --" stdout_buf = io.StringIO() orig = sys.stdout sys.stdout = stdout_buf try: try: _emit_error("sql_test", hostile, True) except SystemExit: pass finally: sys.stdout = orig # Must still be valid JSON d = json.loads(stdout_buf.getvalue()) assert d["message"] == hostile def test_sql_injection_in_slot_name(self) -> None: """A slot name with SQL injection chars must not crash fingerprint or emit_error.""" from muse.core.types import public_key_fingerprint malicious_name = b"'; DROP TABLE slots; --" fp = public_key_fingerprint(malicious_name) assert len(fp) == 71 def test_unicode_in_track_path_emit_error(self) -> None: from muse.cli.commands.agent import _emit_error unicode_path = "tracks/\u4e2d\u6587\u97f3\u4e50.mid" stdout_buf = io.StringIO() orig = sys.stdout sys.stdout = stdout_buf try: try: _emit_error("unicode_test", unicode_path, True) except SystemExit: pass finally: sys.stdout = orig d = json.loads(stdout_buf.getvalue()) assert d["message"] == unicode_path def test_very_long_account_name_in_emit_error(self) -> None: from muse.cli.commands.agent import _emit_error long_msg = "x" * 10_000 stdout_buf = io.StringIO() orig = sys.stdout sys.stdout = stdout_buf try: try: _emit_error("long_test", long_msg, True) except SystemExit: pass finally: sys.stdout = orig d = json.loads(stdout_buf.getvalue()) assert d["message"] == long_msg def test_null_bytes_in_fingerprint_input(self) -> None: from muse.core.types import public_key_fingerprint data = b"\x00" * 64 fp = public_key_fingerprint(data) assert len(fp) == 71 def test_newline_in_error_message_survives_json(self) -> None: from muse.cli.commands.agent import _emit_error newline_msg = "line one\nline two\nline three" stdout_buf = io.StringIO() orig = sys.stdout sys.stdout = stdout_buf try: try: _emit_error("newline_test", newline_msg, True) except SystemExit: pass finally: sys.stdout = orig d = json.loads(stdout_buf.getvalue()) assert d["message"] == newline_msg def test_agent_list_with_encoded_url_returns_json(self) -> None: encoded_url = "http://test.example.com%2Fpath" r = runner.invoke(None, ["agent", "list", "--hub", encoded_url, "--json"]) # Should either succeed with JSON or emit a JSON error — not a crash output = r.output.strip() assert output, "no output produced" json.loads(output) # must be valid JSON regardless # =========================================================================== # Tier 7e — TestPerformance # =========================================================================== class TestPerformance: """Timing bounds for performance-sensitive paths.""" def test_fingerprint_1000_under_500ms(self) -> None: from muse.core.types import public_key_fingerprint data = b"perf test input" * 4 start = time.perf_counter() for _ in range(1000): public_key_fingerprint(data) elapsed_ms = (time.perf_counter() - start) * 1000 assert elapsed_ms < 500, f"1000 public_key_fingerprint calls took {elapsed_ms:.1f}ms" def test_agent_config_status_completes_quickly(self, cfg_repo: pathlib.Path) -> None: _invoke(cfg_repo, ["agent-config", "init"]) start = time.perf_counter() r = _invoke(cfg_repo, ["agent-config", "status", "--json"]) elapsed_ms = (time.perf_counter() - start) * 1000 assert r.exit_code == 0, r.output assert elapsed_ms < 2000, f"agent-config status took {elapsed_ms:.1f}ms" def test_agent_list_completes_quickly(self) -> None: start = time.perf_counter() r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) elapsed_ms = (time.perf_counter() - start) * 1000 assert r.exit_code == 0, r.output assert elapsed_ms < 2000, f"agent list took {elapsed_ms:.1f}ms" def test_duration_ms_in_list_output_is_plausible(self) -> None: r = runner.invoke(None, ["agent", "list", "--hub", _TEST_HUB, "--json"]) assert r.exit_code == 0, r.output d = json.loads(r.output) # duration_ms must be >= 0 and < 5000 (very generous upper bound) assert 0.0 <= d["duration_ms"] < 5000.0 def test_duration_ms_in_agent_config_init_is_plausible(self, cfg_repo: pathlib.Path) -> None: r = _invoke(cfg_repo, ["agent-config", "init", "--json"]) assert r.exit_code == 0, r.output d = json.loads(r.output) assert 0.0 <= d["duration_ms"] < 5000.0