gabriel / muse public
test_core_agent_slots.py python
583 lines 22.2 KB
Raw
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ breaking 10 hours ago
1 """Comprehensive tests for ``muse.core.agent_slots``.
2
3 Covers all eight categories:
4 1. Unit — _toml_escape, _load_raw, _dump round-trip
5 2. Integration — get_next_account, register_slot, list_slots, peek_next_account
6 3. E2E — full read-write cycle through public API with real tmp files
7 4. Stress — 500 sequential accounts, 100-slot registry
8 5. Data integrity — monotonic counter, slot persistence, msign_path format
9 6. Performance — slot operations complete within budget
10 7. Security — symlink guard on write, 0o600 file mode, lock file created
11 8. Docstrings — all public callables and the module have docstrings
12 """
13
14 from __future__ import annotations
15
16 import pathlib
17 import types
18 import stat
19 import time
20 from typing import Any
21
22 import pytest
23
24 # ---------------------------------------------------------------------------
25 # Constants
26 # ---------------------------------------------------------------------------
27
28 _TEST_HUB = "https://localhost:1337"
29 _TEST_HOSTNAME = "localhost:1337"
30 _TEST_HUB2 = "https://staging.musehub.ai"
31 _TEST_HOSTNAME2 = "staging.musehub.ai"
32
33
34 # ---------------------------------------------------------------------------
35 # Fixtures
36 # ---------------------------------------------------------------------------
37
38
39 @pytest.fixture()
40 def slots_dir(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
41 """Redirect the agent-slots store to a temp directory."""
42 fake_dir = tmp_path / "agent_slots"
43 fake_dir.mkdir()
44 fake_file = fake_dir / "agent-slots.toml"
45
46 monkeypatch.setattr("muse.core.agent_slots._SLOTS_DIR", fake_dir)
47 monkeypatch.setattr("muse.core.agent_slots._SLOTS_FILE", fake_file)
48
49 return fake_dir
50
51
52 @pytest.fixture()
53 def slots_file(slots_dir: pathlib.Path) -> pathlib.Path:
54 """Return the path to the isolated agent-slots.toml."""
55 return slots_dir / "agent-slots.toml"
56
57
58 # ---------------------------------------------------------------------------
59 # 1. Unit — pure helpers
60 # ---------------------------------------------------------------------------
61
62
63 class TestTomlEscape:
64 """Unit tests for _toml_escape."""
65
66 def test_escapes_backslash(self) -> None:
67 from muse.core.agent_slots import _toml_escape
68 assert _toml_escape("a\\b") == "a\\\\b"
69
70 def test_escapes_double_quote(self) -> None:
71 from muse.core.agent_slots import _toml_escape
72 assert _toml_escape('say "hello"') == 'say \\"hello\\"'
73
74 def test_plain_string_unchanged(self) -> None:
75 from muse.core.agent_slots import _toml_escape
76 assert _toml_escape("localhost:1337") == "localhost:1337"
77
78 def test_both_special_chars(self) -> None:
79 from muse.core.agent_slots import _toml_escape
80 raw = 'path\\to\\"file"'
81 escaped = _toml_escape(raw)
82 assert "\\\\" in escaped
83 assert '\\"' in escaped
84
85
86 class TestLoadRaw:
87 """Unit tests for _load_raw."""
88
89 def test_returns_empty_dict_when_absent(self, tmp_path: pathlib.Path) -> None:
90 from muse.core.agent_slots import _load_raw
91 assert _load_raw(tmp_path / "nonexistent.toml") == {}
92
93 def test_returns_empty_dict_on_corrupt_file(self, tmp_path: pathlib.Path) -> None:
94 from muse.core.agent_slots import _load_raw
95 p = tmp_path / "bad.toml"
96 p.write_bytes(b"\xff\xfe corrupt")
97 assert _load_raw(p) == {}
98
99 def test_loads_valid_toml(self, tmp_path: pathlib.Path) -> None:
100 from muse.core.agent_slots import _load_raw
101 p = tmp_path / "slots.toml"
102 p.write_text(
103 '["localhost:1337"]\nnext_account = 3\n',
104 encoding="utf-8",
105 )
106 data = _load_raw(p)
107 assert data["localhost:1337"]["next_account"] == 3
108
109
110 class TestDump:
111 """Unit tests for _dump round-trip."""
112
113 def test_empty_dict_produces_empty_string(self) -> None:
114 from muse.core.agent_slots import _dump
115 assert _dump({}) == ""
116
117 def test_round_trip_preserves_next_account(self) -> None:
118 from muse.core.agent_slots import _dump, _load_raw
119 import tempfile, pathlib as pl
120 data = {"localhost:1337": {"next_account": 7, "slots": {"orchestra": 1}}}
121 text = _dump(data)
122 with tempfile.NamedTemporaryFile(
123 mode="w", suffix=".toml", delete=False, encoding="utf-8"
124 ) as f:
125 f.write(text)
126 tmp = pl.Path(f.name)
127 try:
128 loaded = _load_raw(tmp)
129 assert loaded["localhost:1337"]["next_account"] == 7
130 assert loaded["localhost:1337"]["slots"]["orchestra"] == 1
131 finally:
132 tmp.unlink(missing_ok=True)
133
134 def test_slots_sorted_in_output(self) -> None:
135 from muse.core.agent_slots import _dump
136 data = {
137 "localhost:1337": {
138 "next_account": 5,
139 "slots": {"zzz": 3, "aaa": 1},
140 }
141 }
142 text = _dump(data)
143 pos_aaa = text.index("aaa")
144 pos_zzz = text.index("zzz")
145 assert pos_aaa < pos_zzz, "slots should be alphabetically sorted"
146
147 def test_hostnames_sorted_in_output(self) -> None:
148 from muse.core.agent_slots import _dump
149 data = {
150 "z.example.com": {"next_account": 1},
151 "a.example.com": {"next_account": 2},
152 }
153 text = _dump(data)
154 assert text.index("a.example.com") < text.index("z.example.com")
155
156
157 # ---------------------------------------------------------------------------
158 # 2. Integration — public API with isolated store
159 # ---------------------------------------------------------------------------
160
161
162 class TestGetNextAccount:
163 """Integration tests for get_next_account."""
164
165 def test_first_call_returns_1(self, slots_dir: pathlib.Path) -> None:
166 from muse.core.agent_slots import get_next_account
167 assert get_next_account(_TEST_HUB) == 1
168
169 def test_increments_on_each_call(self, slots_dir: pathlib.Path) -> None:
170 from muse.core.agent_slots import get_next_account
171 a = get_next_account(_TEST_HUB)
172 b = get_next_account(_TEST_HUB)
173 c = get_next_account(_TEST_HUB)
174 assert b == a + 1
175 assert c == b + 1
176
177 def test_independent_per_hub(self, slots_dir: pathlib.Path) -> None:
178 from muse.core.agent_slots import get_next_account
179 a1 = get_next_account(_TEST_HUB)
180 b1 = get_next_account(_TEST_HUB2)
181 a2 = get_next_account(_TEST_HUB)
182 b2 = get_next_account(_TEST_HUB2)
183 assert a1 == 1
184 assert b1 == 1
185 assert a2 == 2
186 assert b2 == 2
187
188 def test_minimum_account_is_1(self, slots_dir: pathlib.Path, slots_file: pathlib.Path) -> None:
189 """Even if the TOML contains 0, we clamp to 1."""
190 from muse.core.agent_slots import get_next_account
191 slots_file.write_text(
192 '["localhost:1337"]\nnext_account = 0\n', encoding="utf-8"
193 )
194 assert get_next_account(_TEST_HUB) == 1
195
196
197 class TestPeekNextAccount:
198 """Integration tests for peek_next_account."""
199
200 def test_does_not_increment(self, slots_dir: pathlib.Path) -> None:
201 from muse.core.agent_slots import peek_next_account, get_next_account
202 v1 = peek_next_account(_TEST_HUB)
203 v2 = peek_next_account(_TEST_HUB)
204 assert v1 == v2 == 1
205 # Now increment and verify peek catches up
206 actual = get_next_account(_TEST_HUB)
207 assert actual == 1
208 assert peek_next_account(_TEST_HUB) == 2
209
210 def test_returns_1_when_file_absent(self, slots_dir: pathlib.Path) -> None:
211 from muse.core.agent_slots import peek_next_account
212 assert peek_next_account(_TEST_HUB) == 1
213
214 def test_reflects_current_state(self, slots_dir: pathlib.Path) -> None:
215 from muse.core.agent_slots import get_next_account, peek_next_account
216 for _ in range(5):
217 get_next_account(_TEST_HUB)
218 assert peek_next_account(_TEST_HUB) == 6
219
220
221 class TestRegisterSlot:
222 """Integration tests for register_slot."""
223
224 def test_register_returns_slot_with_correct_fields(
225 self, slots_dir: pathlib.Path
226 ) -> None:
227 from muse.core.agent_slots import register_slot
228 slot = register_slot(_TEST_HUB, "orchestra", 1)
229 assert slot["name"] == "orchestra"
230 assert slot["account"] == 1
231 assert slot["hub"] == _TEST_HOSTNAME
232 assert "msign_path" in slot
233
234 def test_msign_path_contains_account(self, slots_dir: pathlib.Path) -> None:
235 from muse.core.agent_slots import register_slot
236 slot = register_slot(_TEST_HUB, "mixer", 7)
237 assert "7'" in slot["msign_path"]
238
239 def test_msign_path_starts_with_m(self, slots_dir: pathlib.Path) -> None:
240 from muse.core.agent_slots import register_slot
241 slot = register_slot(_TEST_HUB, "test", 2)
242 assert slot["msign_path"].startswith("m/")
243
244 def test_overwrite_updates_account(self, slots_dir: pathlib.Path) -> None:
245 from muse.core.agent_slots import register_slot, list_slots
246 register_slot(_TEST_HUB, "alpha", 1)
247 register_slot(_TEST_HUB, "alpha", 99)
248 slots = list_slots(_TEST_HUB)
249 matched = [s for s in slots if s["name"] == "alpha"]
250 assert len(matched) == 1
251 assert matched[0]["account"] == 99
252
253 def test_persists_to_file(self, slots_dir: pathlib.Path, slots_file: pathlib.Path) -> None:
254 from muse.core.agent_slots import register_slot
255 register_slot(_TEST_HUB, "persistent", 3)
256 assert slots_file.exists()
257 content = slots_file.read_text(encoding="utf-8")
258 assert "persistent" in content
259
260
261 class TestListSlots:
262 """Integration tests for list_slots."""
263
264 def test_empty_when_no_file(self, slots_dir: pathlib.Path) -> None:
265 from muse.core.agent_slots import list_slots
266 assert list_slots(_TEST_HUB) == []
267
268 def test_returns_registered_slots(self, slots_dir: pathlib.Path) -> None:
269 from muse.core.agent_slots import register_slot, list_slots
270 register_slot(_TEST_HUB, "a", 1)
271 register_slot(_TEST_HUB, "b", 2)
272 slots = list_slots(_TEST_HUB)
273 assert len(slots) == 2
274
275 def test_sorted_by_account_ascending(self, slots_dir: pathlib.Path) -> None:
276 from muse.core.agent_slots import register_slot, list_slots
277 register_slot(_TEST_HUB, "z", 10)
278 register_slot(_TEST_HUB, "a", 3)
279 register_slot(_TEST_HUB, "m", 7)
280 slots = list_slots(_TEST_HUB)
281 accounts = [s["account"] for s in slots]
282 assert accounts == sorted(accounts)
283
284 def test_isolated_per_hub(self, slots_dir: pathlib.Path) -> None:
285 from muse.core.agent_slots import register_slot, list_slots
286 register_slot(_TEST_HUB, "hub1-agent", 1)
287 register_slot(_TEST_HUB2, "hub2-agent", 2)
288 slots1 = list_slots(_TEST_HUB)
289 slots2 = list_slots(_TEST_HUB2)
290 assert len(slots1) == 1 and slots1[0]["name"] == "hub1-agent"
291 assert len(slots2) == 1 and slots2[0]["name"] == "hub2-agent"
292
293 def test_all_fields_present(self, slots_dir: pathlib.Path) -> None:
294 from muse.core.agent_slots import register_slot, list_slots
295 register_slot(_TEST_HUB, "full-check", 5)
296 slot = list_slots(_TEST_HUB)[0]
297 assert "name" in slot
298 assert "account" in slot
299 assert "hub" in slot
300 assert "msign_path" in slot
301
302
303 # ---------------------------------------------------------------------------
304 # 3. E2E — full read-write cycle with real tmp files
305 # ---------------------------------------------------------------------------
306
307
308 class TestE2EReadWriteCycle:
309 """End-to-end: write through public API, read back through public API."""
310
311 def test_get_register_list_roundtrip(self, slots_dir: pathlib.Path) -> None:
312 from muse.core.agent_slots import get_next_account, register_slot, list_slots
313
314 acct = get_next_account(_TEST_HUB)
315 register_slot(_TEST_HUB, "roundtrip", acct)
316 slots = list_slots(_TEST_HUB)
317 assert any(s["name"] == "roundtrip" and s["account"] == acct for s in slots)
318
319 def test_peek_before_and_after_register(self, slots_dir: pathlib.Path) -> None:
320 from muse.core.agent_slots import get_next_account, peek_next_account, register_slot
321
322 before = peek_next_account(_TEST_HUB)
323 acct = get_next_account(_TEST_HUB)
324 assert acct == before
325 register_slot(_TEST_HUB, "e2e", acct)
326 after = peek_next_account(_TEST_HUB)
327 assert after == acct + 1
328
329 def test_multiple_hubs_in_same_file(self, slots_dir: pathlib.Path) -> None:
330 from muse.core.agent_slots import register_slot, list_slots, get_next_account
331
332 register_slot(_TEST_HUB, "local-agent", get_next_account(_TEST_HUB))
333 register_slot(_TEST_HUB2, "staging-agent", get_next_account(_TEST_HUB2))
334
335 local = list_slots(_TEST_HUB)
336 staging = list_slots(_TEST_HUB2)
337 assert local[0]["name"] == "local-agent"
338 assert staging[0]["name"] == "staging-agent"
339
340 def test_file_contains_both_hubs(self, slots_dir: pathlib.Path, slots_file: pathlib.Path) -> None:
341 from muse.core.agent_slots import register_slot
342
343 register_slot(_TEST_HUB, "a", 1)
344 register_slot(_TEST_HUB2, "b", 1)
345 content = slots_file.read_text(encoding="utf-8")
346 assert _TEST_HOSTNAME in content
347 assert _TEST_HOSTNAME2 in content
348
349
350 # ---------------------------------------------------------------------------
351 # 4. Stress — many accounts and slots
352 # ---------------------------------------------------------------------------
353
354
355 class TestStress:
356 """Stress tests."""
357
358 def test_500_sequential_get_next_account(self, slots_dir: pathlib.Path) -> None:
359 from muse.core.agent_slots import get_next_account
360 accounts = [get_next_account(_TEST_HUB) for _ in range(500)]
361 assert accounts == list(range(1, 501))
362
363 def test_100_slots_registered_and_listed(self, slots_dir: pathlib.Path) -> None:
364 from muse.core.agent_slots import register_slot, list_slots
365 for i in range(1, 101):
366 register_slot(_TEST_HUB, f"agent-{i:03d}", i)
367 slots = list_slots(_TEST_HUB)
368 assert len(slots) == 100
369
370 def test_repeated_peek_is_stable(self, slots_dir: pathlib.Path) -> None:
371 from muse.core.agent_slots import peek_next_account
372 values = {peek_next_account(_TEST_HUB) for _ in range(100)}
373 assert values == {1}
374
375 def test_interleaved_hubs_stay_independent(self, slots_dir: pathlib.Path) -> None:
376 from muse.core.agent_slots import get_next_account
377 for i in range(50):
378 get_next_account(_TEST_HUB)
379 get_next_account(_TEST_HUB2)
380 from muse.core.agent_slots import peek_next_account
381 assert peek_next_account(_TEST_HUB) == 51
382 assert peek_next_account(_TEST_HUB2) == 51
383
384
385 # ---------------------------------------------------------------------------
386 # 5. Data integrity
387 # ---------------------------------------------------------------------------
388
389
390 class TestDataIntegrity:
391 """Data integrity tests."""
392
393 def test_counter_is_monotonic(self, slots_dir: pathlib.Path) -> None:
394 from muse.core.agent_slots import get_next_account
395 prev = 0
396 for _ in range(20):
397 cur = get_next_account(_TEST_HUB)
398 assert cur > prev
399 prev = cur
400
401 def test_peek_never_exceeds_next(self, slots_dir: pathlib.Path) -> None:
402 from muse.core.agent_slots import get_next_account, peek_next_account
403 for _ in range(10):
404 get_next_account(_TEST_HUB)
405 assert peek_next_account(_TEST_HUB) == 11
406
407 def test_slot_hub_field_is_hostname_not_url(self, slots_dir: pathlib.Path) -> None:
408 from muse.core.agent_slots import register_slot
409 slot = register_slot(_TEST_HUB, "test", 1)
410 assert slot["hub"] == _TEST_HOSTNAME
411 assert "http://" not in slot["hub"]
412
413 def test_msign_path_schema(self, slots_dir: pathlib.Path) -> None:
414 """msign_path must be the full canonical, hub-scoped muse_path shape
415 (musehub#221) — not a hand-rolled, incomplete four-segment string."""
416 from muse.core.agent_slots import register_slot
417 from muse.core.hdkeys import DOMAIN_IDENTITY, ENTITY_AGENT, hub_index, muse_path
418 slot = register_slot(_TEST_HUB, "schema-check", 4)
419 expected = muse_path(DOMAIN_IDENTITY, ENTITY_AGENT, 4, hub=hub_index(_TEST_HOSTNAME))
420 assert slot["msign_path"] == expected
421
422 def test_file_permissions_after_register(
423 self, slots_dir: pathlib.Path, slots_file: pathlib.Path
424 ) -> None:
425 from muse.core.agent_slots import register_slot
426 register_slot(_TEST_HUB, "perm", 1)
427 mode = stat.S_IMODE(slots_file.stat().st_mode)
428 assert mode == 0o600
429
430 def test_next_account_survives_restart(self, slots_dir: pathlib.Path) -> None:
431 """Simulate process restart: counter must be read back from file."""
432 from muse.core.agent_slots import get_next_account, _SLOTS_FILE, _load_raw
433 for _ in range(5):
434 get_next_account(_TEST_HUB)
435 # Reload from file as a fresh process would
436 data = _load_raw(_SLOTS_FILE)
437 assert data[_TEST_HOSTNAME]["next_account"] == 6
438
439
440 # ---------------------------------------------------------------------------
441 # 6. Performance
442 # ---------------------------------------------------------------------------
443
444
445 class TestPerformance:
446 """Performance tests."""
447
448 def test_100_get_next_account_under_3_seconds(
449 self, slots_dir: pathlib.Path
450 ) -> None:
451 from muse.core.agent_slots import get_next_account
452 start = time.monotonic()
453 for _ in range(100):
454 get_next_account(_TEST_HUB)
455 elapsed = time.monotonic() - start
456 assert elapsed < 3.0, f"100 increments took {elapsed:.3f}s"
457
458 def test_100_register_slot_under_3_seconds(
459 self, slots_dir: pathlib.Path
460 ) -> None:
461 from muse.core.agent_slots import register_slot
462 start = time.monotonic()
463 for i in range(1, 101):
464 register_slot(_TEST_HUB, f"perf-{i}", i)
465 elapsed = time.monotonic() - start
466 assert elapsed < 3.0, f"100 register_slot calls took {elapsed:.3f}s"
467
468 def test_list_100_slots_under_1_second(self, slots_dir: pathlib.Path) -> None:
469 from muse.core.agent_slots import register_slot, list_slots
470 for i in range(1, 101):
471 register_slot(_TEST_HUB, f"agent-{i}", i)
472 start = time.monotonic()
473 slots = list_slots(_TEST_HUB)
474 elapsed = time.monotonic() - start
475 assert len(slots) == 100
476 assert elapsed < 1.0, f"list_slots for 100 entries took {elapsed:.3f}s"
477
478
479 # ---------------------------------------------------------------------------
480 # 7. Security
481 # ---------------------------------------------------------------------------
482
483
484 class TestSecurity:
485 """Security tests."""
486
487 def test_symlink_guard_blocks_write(
488 self, slots_dir: pathlib.Path, slots_file: pathlib.Path
489 ) -> None:
490 """Writing must refuse if agent-slots.toml is a symlink."""
491 decoy = slots_dir / "decoy.toml"
492 decoy.write_text("", encoding="utf-8")
493 slots_file.symlink_to(decoy)
494
495 from muse.core.agent_slots import register_slot
496 with pytest.raises(OSError, match="symlink"):
497 register_slot(_TEST_HUB, "malicious", 1)
498
499 def test_file_mode_is_0600(
500 self, slots_dir: pathlib.Path, slots_file: pathlib.Path
501 ) -> None:
502 from muse.core.agent_slots import register_slot
503 register_slot(_TEST_HUB, "mode-check", 1)
504 mode = stat.S_IMODE(slots_file.stat().st_mode)
505 assert mode == 0o600, f"Expected 0o600, got {oct(mode)}"
506
507 def test_lock_file_created_in_slots_dir(
508 self, slots_dir: pathlib.Path
509 ) -> None:
510 from muse.core.agent_slots import get_next_account
511 get_next_account(_TEST_HUB)
512 lock = slots_dir / ".agent-slots.lock"
513 assert lock.exists()
514
515 def test_concurrent_writes_no_data_loss(self, slots_dir: pathlib.Path) -> None:
516 """Two threads incrementing the counter must not produce duplicates."""
517 import threading
518 from muse.core.agent_slots import get_next_account
519
520 results: list[int] = []
521 lock = threading.Lock()
522
523 def worker() -> None:
524 for _ in range(25):
525 v = get_next_account(_TEST_HUB)
526 with lock:
527 results.append(v)
528
529 threads = [threading.Thread(target=worker) for _ in range(4)]
530 for t in threads:
531 t.start()
532 for t in threads:
533 t.join()
534
535 assert len(results) == 100
536 assert len(set(results)) == 100, "Concurrent increments produced duplicates!"
537 assert sorted(results) == list(range(1, 101))
538
539 def test_toml_injection_in_hub_hostname_escaped(
540 self, slots_dir: pathlib.Path
541 ) -> None:
542 """Hub hostnames with TOML special chars must be escaped in output."""
543 from muse.core.agent_slots import register_slot, list_slots, _dump, _load_raw, _SLOTS_FILE
544 # Use a hostname containing a double-quote (pathological but valid to escape)
545 tricky_hub = 'host"with"quotes:8080'
546 register_slot(tricky_hub, "injector", 1)
547 # File must be loadable (no TOML parse error)
548 data = _load_raw(_SLOTS_FILE)
549 assert tricky_hub in data
550
551
552 # ---------------------------------------------------------------------------
553 # 8. Docstrings
554 # ---------------------------------------------------------------------------
555
556
557 class TestDocstrings:
558 """Verify every public callable in agent_slots.py has a docstring."""
559
560 def _public_names(self) -> list[tuple[str, types.FunctionType | type]]:
561 import inspect
562 import muse.core.agent_slots as mod
563 return [
564 (name, obj)
565 for name, obj in inspect.getmembers(mod)
566 if not name.startswith("_")
567 and (inspect.isfunction(obj) or inspect.isclass(obj))
568 and obj.__module__ == mod.__name__
569 ]
570
571 def test_all_public_functions_have_docstrings(self) -> None:
572 for name, obj in self._public_names():
573 assert obj.__doc__, (
574 f"muse.core.agent_slots.{name} is missing a docstring"
575 )
576
577 def test_module_has_docstring(self) -> None:
578 import muse.core.agent_slots as mod
579 assert mod.__doc__, "muse.core.agent_slots module is missing a docstring"
580
581 def test_agent_slot_typed_dict_has_docstring(self) -> None:
582 from muse.core.agent_slots import AgentSlot
583 assert AgentSlot.__doc__
File History 1 commit
sha256:08c083095bcaffb4c43ce947668fac93bf5d261e13d321776170c4019dd1d77d fix: migration must never update identity.toml on a failed … Sonnet 5 minor ⚠ 10 hours ago