"""Comprehensive tests for ``muse.core.agent_slots``. Covers all eight categories: 1. Unit — _toml_escape, _load_raw, _dump round-trip 2. Integration — get_next_account, register_slot, list_slots, peek_next_account 3. E2E — full read-write cycle through public API with real tmp files 4. Stress — 500 sequential accounts, 100-slot registry 5. Data integrity — monotonic counter, slot persistence, msign_path format 6. Performance — slot operations complete within budget 7. Security — symlink guard on write, 0o600 file mode, lock file created 8. Docstrings — all public callables and the module have docstrings """ from __future__ import annotations import pathlib import types import stat import time from typing import Any import pytest # --------------------------------------------------------------------------- # Constants # --------------------------------------------------------------------------- _TEST_HUB = "https://localhost:1337" _TEST_HOSTNAME = "localhost:1337" _TEST_HUB2 = "https://staging.musehub.ai" _TEST_HOSTNAME2 = "staging.musehub.ai" # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def slots_dir(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path: """Redirect the agent-slots store to a temp directory.""" fake_dir = tmp_path / "agent_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 slots_file(slots_dir: pathlib.Path) -> pathlib.Path: """Return the path to the isolated agent-slots.toml.""" return slots_dir / "agent-slots.toml" # --------------------------------------------------------------------------- # 1. Unit — pure helpers # --------------------------------------------------------------------------- class TestTomlEscape: """Unit tests for _toml_escape.""" def test_escapes_backslash(self) -> None: from muse.core.agent_slots import _toml_escape assert _toml_escape("a\\b") == "a\\\\b" def test_escapes_double_quote(self) -> None: from muse.core.agent_slots import _toml_escape assert _toml_escape('say "hello"') == 'say \\"hello\\"' def test_plain_string_unchanged(self) -> None: from muse.core.agent_slots import _toml_escape assert _toml_escape("localhost:1337") == "localhost:1337" def test_both_special_chars(self) -> None: from muse.core.agent_slots import _toml_escape raw = 'path\\to\\"file"' escaped = _toml_escape(raw) assert "\\\\" in escaped assert '\\"' in escaped class TestLoadRaw: """Unit tests for _load_raw.""" def test_returns_empty_dict_when_absent(self, tmp_path: pathlib.Path) -> None: from muse.core.agent_slots import _load_raw assert _load_raw(tmp_path / "nonexistent.toml") == {} def test_returns_empty_dict_on_corrupt_file(self, tmp_path: pathlib.Path) -> None: from muse.core.agent_slots import _load_raw p = tmp_path / "bad.toml" p.write_bytes(b"\xff\xfe corrupt") assert _load_raw(p) == {} def test_loads_valid_toml(self, tmp_path: pathlib.Path) -> None: from muse.core.agent_slots import _load_raw p = tmp_path / "slots.toml" p.write_text( '["localhost:1337"]\nnext_account = 3\n', encoding="utf-8", ) data = _load_raw(p) assert data["localhost:1337"]["next_account"] == 3 class TestDump: """Unit tests for _dump round-trip.""" def test_empty_dict_produces_empty_string(self) -> None: from muse.core.agent_slots import _dump assert _dump({}) == "" def test_round_trip_preserves_next_account(self) -> None: from muse.core.agent_slots import _dump, _load_raw import tempfile, pathlib as pl data = {"localhost:1337": {"next_account": 7, "slots": {"orchestra": 1}}} text = _dump(data) with tempfile.NamedTemporaryFile( mode="w", suffix=".toml", delete=False, encoding="utf-8" ) as f: f.write(text) tmp = pl.Path(f.name) try: loaded = _load_raw(tmp) assert loaded["localhost:1337"]["next_account"] == 7 assert loaded["localhost:1337"]["slots"]["orchestra"] == 1 finally: tmp.unlink(missing_ok=True) def test_slots_sorted_in_output(self) -> None: from muse.core.agent_slots import _dump data = { "localhost:1337": { "next_account": 5, "slots": {"zzz": 3, "aaa": 1}, } } text = _dump(data) pos_aaa = text.index("aaa") pos_zzz = text.index("zzz") assert pos_aaa < pos_zzz, "slots should be alphabetically sorted" def test_hostnames_sorted_in_output(self) -> None: from muse.core.agent_slots import _dump data = { "z.example.com": {"next_account": 1}, "a.example.com": {"next_account": 2}, } text = _dump(data) assert text.index("a.example.com") < text.index("z.example.com") # --------------------------------------------------------------------------- # 2. Integration — public API with isolated store # --------------------------------------------------------------------------- class TestGetNextAccount: """Integration tests for get_next_account.""" def test_first_call_returns_1(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account assert get_next_account(_TEST_HUB) == 1 def test_increments_on_each_call(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account a = get_next_account(_TEST_HUB) b = get_next_account(_TEST_HUB) c = get_next_account(_TEST_HUB) assert b == a + 1 assert c == b + 1 def test_independent_per_hub(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account a1 = get_next_account(_TEST_HUB) b1 = get_next_account(_TEST_HUB2) a2 = get_next_account(_TEST_HUB) b2 = get_next_account(_TEST_HUB2) assert a1 == 1 assert b1 == 1 assert a2 == 2 assert b2 == 2 def test_minimum_account_is_1(self, slots_dir: pathlib.Path, slots_file: pathlib.Path) -> None: """Even if the TOML contains 0, we clamp to 1.""" from muse.core.agent_slots import get_next_account slots_file.write_text( '["localhost:1337"]\nnext_account = 0\n', encoding="utf-8" ) assert get_next_account(_TEST_HUB) == 1 class TestPeekNextAccount: """Integration tests for peek_next_account.""" def test_does_not_increment(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import peek_next_account, get_next_account v1 = peek_next_account(_TEST_HUB) v2 = peek_next_account(_TEST_HUB) assert v1 == v2 == 1 # Now increment and verify peek catches up actual = get_next_account(_TEST_HUB) assert actual == 1 assert peek_next_account(_TEST_HUB) == 2 def test_returns_1_when_file_absent(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import peek_next_account assert peek_next_account(_TEST_HUB) == 1 def test_reflects_current_state(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account, peek_next_account for _ in range(5): get_next_account(_TEST_HUB) assert peek_next_account(_TEST_HUB) == 6 class TestRegisterSlot: """Integration tests for register_slot.""" def test_register_returns_slot_with_correct_fields( self, slots_dir: pathlib.Path ) -> None: from muse.core.agent_slots import register_slot slot = register_slot(_TEST_HUB, "orchestra", 1) assert slot["name"] == "orchestra" assert slot["account"] == 1 assert slot["hub"] == _TEST_HOSTNAME assert "msign_path" in slot def test_msign_path_contains_account(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot slot = register_slot(_TEST_HUB, "mixer", 7) assert "7'" in slot["msign_path"] def test_msign_path_starts_with_m(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot slot = register_slot(_TEST_HUB, "test", 2) assert slot["msign_path"].startswith("m/") def test_overwrite_updates_account(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot, list_slots register_slot(_TEST_HUB, "alpha", 1) register_slot(_TEST_HUB, "alpha", 99) slots = list_slots(_TEST_HUB) matched = [s for s in slots if s["name"] == "alpha"] assert len(matched) == 1 assert matched[0]["account"] == 99 def test_persists_to_file(self, slots_dir: pathlib.Path, slots_file: pathlib.Path) -> None: from muse.core.agent_slots import register_slot register_slot(_TEST_HUB, "persistent", 3) assert slots_file.exists() content = slots_file.read_text(encoding="utf-8") assert "persistent" in content class TestListSlots: """Integration tests for list_slots.""" def test_empty_when_no_file(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import list_slots assert list_slots(_TEST_HUB) == [] def test_returns_registered_slots(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot, list_slots register_slot(_TEST_HUB, "a", 1) register_slot(_TEST_HUB, "b", 2) slots = list_slots(_TEST_HUB) assert len(slots) == 2 def test_sorted_by_account_ascending(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot, list_slots register_slot(_TEST_HUB, "z", 10) register_slot(_TEST_HUB, "a", 3) register_slot(_TEST_HUB, "m", 7) slots = list_slots(_TEST_HUB) accounts = [s["account"] for s in slots] assert accounts == sorted(accounts) def test_isolated_per_hub(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot, list_slots register_slot(_TEST_HUB, "hub1-agent", 1) register_slot(_TEST_HUB2, "hub2-agent", 2) slots1 = list_slots(_TEST_HUB) slots2 = list_slots(_TEST_HUB2) assert len(slots1) == 1 and slots1[0]["name"] == "hub1-agent" assert len(slots2) == 1 and slots2[0]["name"] == "hub2-agent" def test_all_fields_present(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot, list_slots register_slot(_TEST_HUB, "full-check", 5) slot = list_slots(_TEST_HUB)[0] assert "name" in slot assert "account" in slot assert "hub" in slot assert "msign_path" in slot # --------------------------------------------------------------------------- # 3. E2E — full read-write cycle with real tmp files # --------------------------------------------------------------------------- class TestE2EReadWriteCycle: """End-to-end: write through public API, read back through public API.""" def test_get_register_list_roundtrip(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account, register_slot, list_slots acct = get_next_account(_TEST_HUB) register_slot(_TEST_HUB, "roundtrip", acct) slots = list_slots(_TEST_HUB) assert any(s["name"] == "roundtrip" and s["account"] == acct for s in slots) def test_peek_before_and_after_register(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account, peek_next_account, register_slot before = peek_next_account(_TEST_HUB) acct = get_next_account(_TEST_HUB) assert acct == before register_slot(_TEST_HUB, "e2e", acct) after = peek_next_account(_TEST_HUB) assert after == acct + 1 def test_multiple_hubs_in_same_file(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot, list_slots, get_next_account register_slot(_TEST_HUB, "local-agent", get_next_account(_TEST_HUB)) register_slot(_TEST_HUB2, "staging-agent", get_next_account(_TEST_HUB2)) local = list_slots(_TEST_HUB) staging = list_slots(_TEST_HUB2) assert local[0]["name"] == "local-agent" assert staging[0]["name"] == "staging-agent" def test_file_contains_both_hubs(self, slots_dir: pathlib.Path, slots_file: pathlib.Path) -> None: from muse.core.agent_slots import register_slot register_slot(_TEST_HUB, "a", 1) register_slot(_TEST_HUB2, "b", 1) content = slots_file.read_text(encoding="utf-8") assert _TEST_HOSTNAME in content assert _TEST_HOSTNAME2 in content # --------------------------------------------------------------------------- # 4. Stress — many accounts and slots # --------------------------------------------------------------------------- class TestStress: """Stress tests.""" def test_500_sequential_get_next_account(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account accounts = [get_next_account(_TEST_HUB) for _ in range(500)] assert accounts == list(range(1, 501)) def test_100_slots_registered_and_listed(self, slots_dir: 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:03d}", i) slots = list_slots(_TEST_HUB) assert len(slots) == 100 def test_repeated_peek_is_stable(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import peek_next_account values = {peek_next_account(_TEST_HUB) for _ in range(100)} assert values == {1} def test_interleaved_hubs_stay_independent(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account for i in range(50): get_next_account(_TEST_HUB) get_next_account(_TEST_HUB2) from muse.core.agent_slots import peek_next_account assert peek_next_account(_TEST_HUB) == 51 assert peek_next_account(_TEST_HUB2) == 51 # --------------------------------------------------------------------------- # 5. Data integrity # --------------------------------------------------------------------------- class TestDataIntegrity: """Data integrity tests.""" def test_counter_is_monotonic(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account prev = 0 for _ in range(20): cur = get_next_account(_TEST_HUB) assert cur > prev prev = cur def test_peek_never_exceeds_next(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import get_next_account, peek_next_account for _ in range(10): get_next_account(_TEST_HUB) assert peek_next_account(_TEST_HUB) == 11 def test_slot_hub_field_is_hostname_not_url(self, slots_dir: pathlib.Path) -> None: from muse.core.agent_slots import register_slot slot = register_slot(_TEST_HUB, "test", 1) assert slot["hub"] == _TEST_HOSTNAME assert "http://" not in slot["hub"] def test_msign_path_schema(self, slots_dir: pathlib.Path) -> None: """msign_path must be m/purpose'/domain'/entity_agent'/account'.""" from muse.core.agent_slots import register_slot from muse.core.hdkeys import DOMAIN_IDENTITY, ENTITY_AGENT, MUSE_PURPOSE slot = register_slot(_TEST_HUB, "schema-check", 4) expected = f"m/{MUSE_PURPOSE}'/{DOMAIN_IDENTITY}'/{ENTITY_AGENT}'/4'" assert slot["msign_path"] == expected def test_file_permissions_after_register( self, slots_dir: pathlib.Path, slots_file: pathlib.Path ) -> None: from muse.core.agent_slots import register_slot register_slot(_TEST_HUB, "perm", 1) mode = stat.S_IMODE(slots_file.stat().st_mode) assert mode == 0o600 def test_next_account_survives_restart(self, slots_dir: pathlib.Path) -> None: """Simulate process restart: counter must be read back from file.""" from muse.core.agent_slots import get_next_account, _SLOTS_FILE, _load_raw for _ in range(5): get_next_account(_TEST_HUB) # Reload from file as a fresh process would data = _load_raw(_SLOTS_FILE) assert data[_TEST_HOSTNAME]["next_account"] == 6 # --------------------------------------------------------------------------- # 6. Performance # --------------------------------------------------------------------------- class TestPerformance: """Performance tests.""" def test_100_get_next_account_under_3_seconds( self, slots_dir: pathlib.Path ) -> None: from muse.core.agent_slots import get_next_account start = time.monotonic() for _ in range(100): get_next_account(_TEST_HUB) elapsed = time.monotonic() - start assert elapsed < 3.0, f"100 increments took {elapsed:.3f}s" def test_100_register_slot_under_3_seconds( self, slots_dir: pathlib.Path ) -> None: from muse.core.agent_slots import register_slot start = time.monotonic() for i in range(1, 101): register_slot(_TEST_HUB, f"perf-{i}", i) elapsed = time.monotonic() - start assert elapsed < 3.0, f"100 register_slot calls took {elapsed:.3f}s" def test_list_100_slots_under_1_second(self, slots_dir: 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) start = time.monotonic() slots = list_slots(_TEST_HUB) elapsed = time.monotonic() - start assert len(slots) == 100 assert elapsed < 1.0, f"list_slots for 100 entries took {elapsed:.3f}s" # --------------------------------------------------------------------------- # 7. Security # --------------------------------------------------------------------------- class TestSecurity: """Security tests.""" def test_symlink_guard_blocks_write( self, slots_dir: pathlib.Path, slots_file: pathlib.Path ) -> None: """Writing must refuse if agent-slots.toml is a symlink.""" decoy = slots_dir / "decoy.toml" decoy.write_text("", encoding="utf-8") 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_file_mode_is_0600( self, slots_dir: pathlib.Path, slots_file: pathlib.Path ) -> None: from muse.core.agent_slots import register_slot register_slot(_TEST_HUB, "mode-check", 1) mode = stat.S_IMODE(slots_file.stat().st_mode) assert mode == 0o600, f"Expected 0o600, got {oct(mode)}" def test_lock_file_created_in_slots_dir( self, slots_dir: pathlib.Path ) -> None: from muse.core.agent_slots import get_next_account get_next_account(_TEST_HUB) lock = slots_dir / ".agent-slots.lock" assert lock.exists() def test_concurrent_writes_no_data_loss(self, slots_dir: pathlib.Path) -> None: """Two threads incrementing the counter must not produce duplicates.""" import threading from muse.core.agent_slots import get_next_account results: list[int] = [] lock = threading.Lock() def worker() -> None: for _ in range(25): v = get_next_account(_TEST_HUB) with lock: results.append(v) threads = [threading.Thread(target=worker) for _ in range(4)] for t in threads: t.start() for t in threads: t.join() assert len(results) == 100 assert len(set(results)) == 100, "Concurrent increments produced duplicates!" assert sorted(results) == list(range(1, 101)) def test_toml_injection_in_hub_hostname_escaped( self, slots_dir: pathlib.Path ) -> None: """Hub hostnames with TOML special chars must be escaped in output.""" from muse.core.agent_slots import register_slot, list_slots, _dump, _load_raw, _SLOTS_FILE # Use a hostname containing a double-quote (pathological but valid to escape) tricky_hub = 'host"with"quotes:8080' register_slot(tricky_hub, "injector", 1) # File must be loadable (no TOML parse error) data = _load_raw(_SLOTS_FILE) assert tricky_hub in data # --------------------------------------------------------------------------- # 8. Docstrings # --------------------------------------------------------------------------- class TestDocstrings: """Verify every public callable in agent_slots.py has a docstring.""" def _public_names(self) -> list[tuple[str, types.FunctionType | type]]: import inspect import muse.core.agent_slots as mod return [ (name, obj) for name, obj in inspect.getmembers(mod) if not name.startswith("_") and (inspect.isfunction(obj) or inspect.isclass(obj)) and obj.__module__ == mod.__name__ ] def test_all_public_functions_have_docstrings(self) -> None: for name, obj in self._public_names(): assert obj.__doc__, ( f"muse.core.agent_slots.{name} is missing a docstring" ) def test_module_has_docstring(self) -> None: import muse.core.agent_slots as mod assert mod.__doc__, "muse.core.agent_slots module is missing a docstring" def test_agent_slot_typed_dict_has_docstring(self) -> None: from muse.core.agent_slots import AgentSlot assert AgentSlot.__doc__