"""Comprehensive tests for ``muse agent`` CLI commands. Covers all eight required categories: 1. Unit — pure helper functions (_derive_agent_seed, _fingerprint, etc.) 2. Integration — run_keygen / run_list / run_register with a real (tmp) identity store 3. E2E — full CLI via CliRunner 4. Stress — many accounts, repeated derivation 5. Data integrity — determinism, isolation between accounts 6. Performance — keygen completes within budget 7. Security — negative accounts rejected, symlink guard, no mnemonic in output 8. Docstrings — all public callables are documented """ from __future__ import annotations import argparse import json import pathlib import time from typing import Any import pytest from tests.cli_test_helper import CliRunner from muse.core.paths import muse_dir from muse.core.types import b64url_decode, public_key_fingerprint cli = None # argparse migration — CliRunner ignores this arg runner = CliRunner() # --------------------------------------------------------------------------- # Constants — fixed test mnemonic (never used in production) # --------------------------------------------------------------------------- _TEST_MNEMONIC = ( "abandon abandon abandon abandon abandon abandon abandon abandon " "abandon abandon abandon about" ) _TEST_HUB = "https://localhost:1337" _TEST_HOSTNAME = "localhost:1337" # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def isolated_identity( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> pathlib.Path: """Redirect identity store to tmp_path so tests never touch ~/.muse/identity.toml.""" fake_dir = tmp_path / "muse_dir" fake_dir.mkdir() fake_file = fake_dir / "identity.toml" keys_dir = fake_dir / "keys" keys_dir.mkdir() monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", fake_dir) monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", fake_file) return fake_dir @pytest.fixture() def isolated_slots( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> pathlib.Path: """Redirect agent-slots store to tmp_path.""" fake_dir = tmp_path / "muse_dir_slots" fake_dir.mkdir() fake_file = fake_dir / "agent-slots.toml" monkeypatch.setattr("muse.core.agent_slots._SLOTS_DIR", fake_dir) monkeypatch.setattr("muse.core.agent_slots._SLOTS_FILE", fake_file) return fake_dir @pytest.fixture() def identity_with_mnemonic( isolated_identity: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: """Save a test identity entry backed by an in-memory keychain.""" from muse.core.identity import IdentityEntry, save_identity # Patch keychain to an in-memory store so the mnemonic survives the # save_identity → load_identity round-trip without touching the real OS keychain. _kc: dict[str, str] = {} monkeypatch.setattr("muse.core.keychain.is_available", lambda: True) monkeypatch.setattr("muse.core.keychain.store", lambda m: _kc.__setitem__("mnemonic", m)) monkeypatch.setattr("muse.core.keychain.load", lambda: _kc.get("mnemonic")) entry: IdentityEntry = { "type": "human", "handle": "gabriel", "hd_path": "m/1075233755'/0'/0'/0'/0'/0'", } save_identity(_TEST_HUB, entry, mnemonic=_TEST_MNEMONIC) @pytest.fixture() def repo_with_hub( tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> pathlib.Path: """Minimal .muse/ repo with hub configured so --hub can be omitted.""" dot_muse = muse_dir(tmp_path) dot_muse.mkdir() (dot_muse / "HEAD").write_text("ref: refs/heads/main\n", encoding="utf-8") (dot_muse / "refs" / "heads").mkdir(parents=True) (dot_muse / "objects").mkdir() (dot_muse / "commits").mkdir() (dot_muse / "snapshots").mkdir() (dot_muse / "config.toml").write_text( f'[hub]\nurl = "{_TEST_HUB}"\n', encoding="utf-8" ) monkeypatch.chdir(tmp_path) return tmp_path # --------------------------------------------------------------------------- # 1. Unit — pure helpers # --------------------------------------------------------------------------- class TestDeriveAgentSeed: """Unit tests for _derive_agent_seed.""" def test_returns_64_bytes(self) -> None: from muse.cli.commands.agent import _derive_agent_seed result = _derive_agent_seed(_TEST_MNEMONIC, 0) assert len(result) == 64 def test_is_bytes(self) -> None: from muse.cli.commands.agent import _derive_agent_seed result = _derive_agent_seed(_TEST_MNEMONIC, 1) assert isinstance(result, (bytes, bytearray)) def test_different_accounts_produce_different_seeds(self) -> None: from muse.cli.commands.agent import _derive_agent_seed s0 = _derive_agent_seed(_TEST_MNEMONIC, 0) s1 = _derive_agent_seed(_TEST_MNEMONIC, 1) assert s0 != s1 def test_same_account_is_deterministic(self) -> None: from muse.cli.commands.agent import _derive_agent_seed s_a = _derive_agent_seed(_TEST_MNEMONIC, 5) s_b = _derive_agent_seed(_TEST_MNEMONIC, 5) assert s_a == s_b def test_different_mnemonics_produce_different_seeds(self) -> None: from muse.cli.commands.agent import _derive_agent_seed mnemonic2 = ( "zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo wrong" ) s1 = _derive_agent_seed(_TEST_MNEMONIC, 0) s2 = _derive_agent_seed(mnemonic2, 0) assert s1 != s2 class TestSubSeedToPublic: """Unit tests for _sub_seed_to_public.""" def test_returns_32_bytes(self) -> None: from muse.cli.commands.agent import _derive_agent_seed, _sub_seed_to_public sub_seed = _derive_agent_seed(_TEST_MNEMONIC, 0) pub = _sub_seed_to_public(sub_seed) assert len(pub) == 32 def test_deterministic(self) -> None: from muse.cli.commands.agent import _derive_agent_seed, _sub_seed_to_public sub_seed = _derive_agent_seed(_TEST_MNEMONIC, 0) assert _sub_seed_to_public(sub_seed) == _sub_seed_to_public(sub_seed) def test_different_seeds_different_pubkeys(self) -> None: from muse.cli.commands.agent import _derive_agent_seed, _sub_seed_to_public s0 = _derive_agent_seed(_TEST_MNEMONIC, 0) s1 = _derive_agent_seed(_TEST_MNEMONIC, 1) assert _sub_seed_to_public(s0) != _sub_seed_to_public(s1) class TestRequireMnemonic: """Unit tests for _require_mnemonic.""" def test_returns_mnemonic_from_identity( self, identity_with_mnemonic: None ) -> None: from muse.cli.commands.agent import _require_mnemonic result = _require_mnemonic(_TEST_HUB) assert result == _TEST_MNEMONIC def test_raises_when_no_identity(self, isolated_identity: pathlib.Path) -> None: from muse.cli.commands.agent import _require_mnemonic with pytest.raises(SystemExit) as exc_info: _require_mnemonic(_TEST_HUB) assert exc_info.value.code == 1 def test_raises_when_identity_has_no_mnemonic( self, isolated_identity: pathlib.Path, monkeypatch: pytest.MonkeyPatch ) -> None: monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled") from muse.core.identity import IdentityEntry, save_identity from muse.cli.commands.agent import _require_mnemonic entry: IdentityEntry = {"type": "human", "handle": "gabriel"} save_identity(_TEST_HUB, entry) with pytest.raises(SystemExit) as exc_info: _require_mnemonic(_TEST_HUB) assert exc_info.value.code == 1 class TestResolveHubUrl: """Unit tests for _resolve_hub_url.""" def test_returns_args_hub_when_provided(self) -> None: from muse.cli.commands.agent import _resolve_hub_url assert _resolve_hub_url("http://localhost:9999") == "http://localhost:9999" def test_raises_when_no_hub(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: from muse.cli.commands.agent import _resolve_hub_url monkeypatch.chdir(tmp_path) with pytest.raises(SystemExit) as exc_info: _resolve_hub_url(None) assert exc_info.value.code == 1 def test_reads_from_repo_config( self, repo_with_hub: pathlib.Path ) -> None: from muse.cli.commands.agent import _resolve_hub_url url = _resolve_hub_url(None) assert url == _TEST_HUB # --------------------------------------------------------------------------- # 2. Integration — run_keygen / run_list / run_register # --------------------------------------------------------------------------- class TestRunKeygen: """Integration tests for run_keygen.""" def test_keygen_produces_valid_output( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: import argparse from muse.cli.commands.agent import run_keygen args = argparse.Namespace(hub=_TEST_HUB, account=1, name=None, json_out=True) import io, contextlib, sys out = io.StringIO() with contextlib.redirect_stdout(out): run_keygen(args) payload = json.loads(out.getvalue()) assert payload["status"] == "ok" assert payload["account"] == 1 assert len(payload["fingerprint"]) == 71 # Sub-seed decodes to 64 bytes seed_bytes = b64url_decode(payload["hd_seed_b64"]) assert len(seed_bytes) == 64 def test_keygen_msign_path_format( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.cli.commands.agent import run_keygen import io, contextlib args = argparse.Namespace(hub=_TEST_HUB, account=3, name=None, json_out=True) out = io.StringIO() with contextlib.redirect_stdout(out): run_keygen(args) payload = json.loads(out.getvalue()) # Path: m/purpose'/domain_identity'/entity_agent'/account' assert payload["msign_path"].endswith("'/3'") assert payload["msign_path"].startswith("m/") def test_keygen_negative_account_rejected( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.cli.commands.agent import run_keygen args = argparse.Namespace(hub=_TEST_HUB, account=-1, name=None, json_out=True) with pytest.raises(SystemExit) as exc_info: run_keygen(args) assert exc_info.value.code == 1 class TestRunList: """Integration tests for run_list.""" def test_list_empty( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.cli.commands.agent import run_list import io, contextlib args = argparse.Namespace(hub=_TEST_HUB, json_out=True) out = io.StringIO() with contextlib.redirect_stdout(out): run_list(args) result = json.loads(out.getvalue()) assert result["slots"] == [] def test_list_shows_registered_slots( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.core.agent_slots import register_slot from muse.cli.commands.agent import run_list import io, contextlib register_slot(_TEST_HUB, "orchestra", 1) register_slot(_TEST_HUB, "mixer", 2) args = argparse.Namespace(hub=_TEST_HUB, json_out=True) out = io.StringIO() with contextlib.redirect_stdout(out): run_list(args) slots = json.loads(out.getvalue())["slots"] assert len(slots) == 2 names = {s["name"] for s in slots} assert names == {"orchestra", "mixer"} def test_list_sorted_by_account( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.core.agent_slots import register_slot from muse.cli.commands.agent import run_list import io, contextlib register_slot(_TEST_HUB, "b-agent", 5) register_slot(_TEST_HUB, "a-agent", 2) args = argparse.Namespace(hub=_TEST_HUB, json_out=True) out = io.StringIO() with contextlib.redirect_stdout(out): run_list(args) slots = json.loads(out.getvalue())["slots"] accounts = [s["account"] for s in slots] assert accounts == sorted(accounts) class TestRunRegister: """Integration tests for run_register.""" def test_register_creates_slot( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.cli.commands.agent import run_register, run_list import io, contextlib args = argparse.Namespace(hub=_TEST_HUB, account=1, name="orchestra", json_out=True) out = io.StringIO() with contextlib.redirect_stdout(out): run_register(args) payload = json.loads(out.getvalue()) assert payload["status"] == "ok" assert payload["name"] == "orchestra" assert payload["account"] == 1 def test_register_persists_across_calls( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.cli.commands.agent import run_register, run_list import io, contextlib reg_args = argparse.Namespace(hub=_TEST_HUB, account=7, name="test-agent", json_out=False) with contextlib.redirect_stdout(io.StringIO()): run_register(reg_args) list_args = argparse.Namespace(hub=_TEST_HUB, json_out=True) out = io.StringIO() with contextlib.redirect_stdout(out): run_list(list_args) slots = json.loads(out.getvalue())["slots"] assert any(s["name"] == "test-agent" and s["account"] == 7 for s in slots) def test_register_negative_account_rejected( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: import argparse from muse.cli.commands.agent import run_register args = argparse.Namespace(hub=_TEST_HUB, account=-5, name="bad", json_out=False) with pytest.raises(SystemExit) as exc_info: run_register(args) assert exc_info.value.code == 1 # --------------------------------------------------------------------------- # 3. E2E — full CLI via CliRunner # --------------------------------------------------------------------------- class TestAgentKeygenE2E: """End-to-end tests: muse agent keygen via CliRunner.""" def test_keygen_json_exit_0( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"] ) assert result.exit_code == 0 payload = json.loads(result.stdout.split("\n")[0]) assert payload["status"] == "ok" assert payload["account"] == 1 def test_keygen_human_readable( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "2"] ) assert result.exit_code == 0 assert "MUSE_AGENT_HD_SEED=" in result.output def test_keygen_no_hub_and_no_config_fails(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.chdir(tmp_path) result = runner.invoke( cli, ["agent", "keygen", "--account", "1", "--json"] ) assert result.exit_code != 0 def test_keygen_no_identity_fails( self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"] ) assert result.exit_code != 0 def test_keygen_requires_account( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--json"] ) assert result.exit_code != 0 def test_keygen_with_name_includes_name_in_json( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--name", "orchestra", "--json"], ) assert result.exit_code == 0 payload = json.loads(result.stdout.split("\n")[0]) assert payload["name"] == "orchestra" class TestAgentListE2E: """End-to-end tests: muse agent list via CliRunner.""" def test_list_empty_json( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "list", "--hub", _TEST_HUB, "--json"] ) assert result.exit_code == 0 assert json.loads(result.stdout.split("\n")[0])["slots"] == [] def test_list_after_register( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: runner.invoke( cli, ["agent", "register", "--hub", _TEST_HUB, "--account", "3", "--name", "my-bot", "--json"], ) result = runner.invoke( cli, ["agent", "list", "--hub", _TEST_HUB, "--json"] ) assert result.exit_code == 0 slots = json.loads(result.stdout.split("\n")[0])["slots"] assert any(s["name"] == "my-bot" for s in slots) def test_list_human_readable_empty( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke(cli, ["agent", "list", "--hub", _TEST_HUB]) assert result.exit_code == 0 assert "No registered" in result.output class TestAgentRegisterE2E: """End-to-end tests: muse agent register via CliRunner.""" def test_register_json_ok( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "register", "--hub", _TEST_HUB, "--account", "4", "--name", "test", "--json"], ) assert result.exit_code == 0 payload = json.loads(result.stdout.split("\n")[0]) assert payload["status"] == "ok" assert payload["account"] == 4 def test_register_requires_account( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "register", "--hub", _TEST_HUB, "--name", "test", "--json"], ) assert result.exit_code != 0 def test_register_requires_name( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "register", "--hub", _TEST_HUB, "--account", "1", "--json"], ) assert result.exit_code != 0 # --------------------------------------------------------------------------- # 4. Stress — many accounts, repeated operations # --------------------------------------------------------------------------- class TestStress: """Stress tests — many accounts, repeated derivations.""" def test_100_different_accounts_all_unique(self) -> None: from muse.cli.commands.agent import _derive_agent_seed seeds = [bytes(_derive_agent_seed(_TEST_MNEMONIC, i)) for i in range(100)] assert len(set(seeds)) == 100 def test_repeated_derivation_consistent(self) -> None: from muse.cli.commands.agent import _derive_agent_seed for _ in range(50): s = _derive_agent_seed(_TEST_MNEMONIC, 42) assert len(s) == 64 def test_register_and_list_100_slots( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: from muse.core.agent_slots import register_slot, list_slots for i in range(1, 101): register_slot(_TEST_HUB, f"agent-{i}", i) slots = list_slots(_TEST_HUB) assert len(slots) == 100 accounts = [s["account"] for s in slots] assert accounts == sorted(accounts) # --------------------------------------------------------------------------- # 5. Data integrity — determinism and isolation # --------------------------------------------------------------------------- class TestDataIntegrity: """Data integrity tests.""" def test_keygen_account_0_and_1_produce_different_seeds(self) -> None: from muse.cli.commands.agent import _derive_agent_seed, _sub_seed_to_public s0 = _derive_agent_seed(_TEST_MNEMONIC, 0) s1 = _derive_agent_seed(_TEST_MNEMONIC, 1) p0 = _sub_seed_to_public(s0) p1 = _sub_seed_to_public(s1) assert p0 != p1 def test_hd_seed_b64_decodes_to_64_bytes( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"] ) assert result.exit_code == 0 payload = json.loads(result.stdout.split("\n")[0]) raw = b64url_decode(payload["hd_seed_b64"]) assert len(raw) == 64 def test_fingerprint_matches_sha256_of_public_key( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"] ) assert result.exit_code == 0 payload = json.loads(result.stdout.split("\n")[0]) pub_bytes = b64url_decode(payload["public_key_b64"]) expected_fp = public_key_fingerprint(pub_bytes) assert payload["fingerprint"] == expected_fp def test_same_account_produces_same_output_in_separate_invocations( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: r1 = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "7", "--json"] ) r2 = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "7", "--json"] ) p1 = json.loads(r1.stdout.split("\n")[0]) p2 = json.loads(r2.stdout.split("\n")[0]) assert p1["fingerprint"] == p2["fingerprint"] assert p1["hd_seed_b64"] == p2["hd_seed_b64"] def test_slot_overwrite_updates_account( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: from muse.core.agent_slots import register_slot, list_slots register_slot(_TEST_HUB, "shared-name", 1) register_slot(_TEST_HUB, "shared-name", 2) slots = list_slots(_TEST_HUB) matched = [s for s in slots if s["name"] == "shared-name"] assert len(matched) == 1 assert matched[0]["account"] == 2 def test_msign_path_contains_account_index( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "9", "--json"] ) assert result.exit_code == 0 payload = json.loads(result.stdout.split("\n")[0]) assert "9'" in payload["msign_path"] # --------------------------------------------------------------------------- # 6. Performance — keygen completes within budget # --------------------------------------------------------------------------- class TestPerformance: """Performance tests — keygen latency budget.""" def test_keygen_under_2_seconds( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: from muse.cli.commands.agent import _derive_agent_seed, _sub_seed_to_public start = time.monotonic() sub_seed = _derive_agent_seed(_TEST_MNEMONIC, 1) _sub_seed_to_public(sub_seed) elapsed = time.monotonic() - start assert elapsed < 2.0, f"Keygen took {elapsed:.3f}s — expected < 2s" def test_10_sequential_keygens_under_5_seconds( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: from muse.cli.commands.agent import _derive_agent_seed, _sub_seed_to_public start = time.monotonic() for i in range(10): sub = _derive_agent_seed(_TEST_MNEMONIC, i) _sub_seed_to_public(sub) elapsed = time.monotonic() - start assert elapsed < 5.0, f"10 keygens took {elapsed:.3f}s — expected < 5s" # --------------------------------------------------------------------------- # 7. Security # --------------------------------------------------------------------------- class TestSecurity: """Security tests.""" def test_negative_account_rejected_in_keygen( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "-1", "--json"], ) assert result.exit_code != 0 def test_negative_account_rejected_in_register( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "register", "--hub", _TEST_HUB, "--account", "-3", "--name", "bad", "--json"], ) assert result.exit_code != 0 def test_mnemonic_not_in_keygen_json_output( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: result = runner.invoke( cli, ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"] ) assert result.exit_code == 0 assert "abandon" not in result.output # mnemonic word not leaked def test_mnemonic_not_in_list_output( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: from muse.core.agent_slots import register_slot register_slot(_TEST_HUB, "safe", 1) result = runner.invoke( cli, ["agent", "list", "--hub", _TEST_HUB, "--json"] ) assert result.exit_code == 0 assert "abandon" not in result.output def test_slots_file_symlink_guard( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path, ) -> None: """agent-slots.toml cannot be a symlink — _save raises OSError.""" from muse.core.agent_slots import _SLOTS_FILE, _SLOTS_DIR # Create a decoy file, then replace agent-slots.toml with a symlink to it decoy = isolated_slots / "decoy.toml" decoy.write_text("", encoding="utf-8") slots_file = isolated_slots / "agent-slots.toml" slots_file.symlink_to(decoy) from muse.core.agent_slots import register_slot with pytest.raises(OSError, match="symlink"): register_slot(_TEST_HUB, "malicious", 1) def test_slots_dir_not_world_readable( self, identity_with_mnemonic: None, isolated_slots: pathlib.Path ) -> None: """After writing, the slots file should have mode 0o600.""" from muse.core.agent_slots import register_slot import stat as stat_mod register_slot(_TEST_HUB, "check-perms", 1) slots_file = isolated_slots / "agent-slots.toml" mode = stat_mod.S_IMODE(slots_file.stat().st_mode) assert mode == 0o600, f"Expected 0o600 but got {oct(mode)}" # --------------------------------------------------------------------------- # 8. Docstrings — every public callable is documented # --------------------------------------------------------------------------- class TestRegisterFlags: """Argparse registration tests for ``muse agent`` subcommands.""" def _parse_keygen(self, *args: str) -> argparse.Namespace: from muse.cli.commands.agent import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["agent", "keygen", *args]) def _parse_list(self, *args: str) -> argparse.Namespace: from muse.cli.commands.agent import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["agent", "list", *args]) def _parse_register(self, *args: str) -> argparse.Namespace: from muse.cli.commands.agent import register p = argparse.ArgumentParser() sub = p.add_subparsers() register(sub) return p.parse_args(["agent", "register", "--account", "1", "--name", "test", *args]) # keygen def test_keygen_default_json_out_is_false(self) -> None: ns = self._parse_keygen("--account", "1") assert ns.json_out is False def test_keygen_json_flag_sets_json_out(self) -> None: ns = self._parse_keygen("--account", "1", "--json") assert ns.json_out is True def test_keygen_j_shorthand_sets_json_out(self) -> None: ns = self._parse_keygen("--account", "1", "-j") assert ns.json_out is True def test_keygen_account_flag(self) -> None: ns = self._parse_keygen("--account", "5") assert ns.account == 5 def test_keygen_hub_default(self) -> None: ns = self._parse_keygen("--account", "1") assert ns.hub is None def test_keygen_name_default(self) -> None: ns = self._parse_keygen("--account", "1") assert ns.name is None # list def test_list_default_json_out_is_false(self) -> None: ns = self._parse_list() assert ns.json_out is False def test_list_json_flag_sets_json_out(self) -> None: ns = self._parse_list("--json") assert ns.json_out is True def test_list_j_shorthand_sets_json_out(self) -> None: ns = self._parse_list("-j") assert ns.json_out is True # register def test_register_default_json_out_is_false(self) -> None: ns = self._parse_register() assert ns.json_out is False def test_register_json_flag_sets_json_out(self) -> None: ns = self._parse_register("--json") assert ns.json_out is True def test_register_j_shorthand_sets_json_out(self) -> None: ns = self._parse_register("-j") assert ns.json_out is True def test_register_account_and_name_required(self) -> None: p = argparse.ArgumentParser() sub = p.add_subparsers() from muse.cli.commands.agent import register register(sub) with pytest.raises(SystemExit): p.parse_args(["agent", "register"]) class TestDocstrings: """Verify every public function/class in agent.py has a docstring.""" def _public_names(self) -> list[str]: import inspect import muse.cli.commands.agent as mod names = [] for name, obj in inspect.getmembers(mod): if name.startswith("_"): continue if inspect.isfunction(obj) or inspect.isclass(obj): if obj.__module__ == mod.__name__: names.append((name, obj)) return names def test_all_public_functions_have_docstrings(self) -> None: for name, obj in self._public_names(): assert obj.__doc__, f"muse.cli.commands.agent.{name} is missing a docstring" def test_module_has_docstring(self) -> None: import muse.cli.commands.agent as mod assert mod.__doc__, "muse.cli.commands.agent module is missing a docstring" def test_typed_dicts_have_docstrings(self) -> None: from muse.cli.commands.agent import _KeygenJson, _RegisterJson assert _KeygenJson.__doc__ assert _RegisterJson.__doc__