gabriel / muse public
test_agent_json_schema.py python
457 lines 16.4 KB
Raw
sha256:ff478cfdcdd4b7fd6de89cb68896601a981f945634463275ec333bd20ca36402 Merge branch 'dev' into main Human 20 days ago
1 """Tests for the canonical ``muse agent`` JSON schema.
2
3 Coverage
4 --------
5 I keygen schema
6 I1 All required keys present in keygen response
7 I2 status is "ok"
8 I3 hd_seed_b64 decodes to exactly 64 bytes
9 I4 public_key_b64 decodes to exactly 32 bytes
10 I5 fingerprint is sha256 hex of public_key_b64 bytes
11 I6 name is null when --name not provided
12 I7 name reflects --name flag when provided
13 I8 msign_path contains the account index
14 I9 hub is the full URL passed via --hub
15
16 II list schema
17 II1 Returns a JSON array (not object)
18 II2 Empty array when no slots registered
19 II3 Each entry has all required keys
20 II4 Entries are sorted by account index (ascending)
21 II5 hub in each entry is hostname (not full URL)
22
23 III register schema
24 III1 All required keys present in register response
25 III2 status is "ok"
26 III3 hub is hostname (not full URL)
27 III4 msign_path contains the account index
28
29 IV Error paths — JSON errors when --json is passed
30 IV1 keygen with no identity → JSON error, exit 1
31 IV2 keygen with no mnemonic → JSON error, exit 1
32 IV3 keygen with negative account → JSON error, exit 1
33 IV4 keygen with no hub (no config) → JSON error, exit 1
34 IV5 Error responses include "error" key
35 IV6 Error responses include "message" key
36 """
37
38 from __future__ import annotations
39 from collections.abc import Mapping
40
41 import json
42 import pathlib
43
44 import pytest
45
46 from tests.cli_test_helper import CliRunner, InvokeResult
47 from muse.core.types import b64url_decode, public_key_fingerprint
48
49 cli = None
50 runner = CliRunner()
51
52 _TEST_HUB = "https://localhost:1337"
53 _TEST_HOSTNAME = "localhost:1337"
54 _TEST_MNEMONIC = (
55 "abandon abandon abandon abandon abandon abandon abandon abandon "
56 "abandon abandon abandon about"
57 )
58
59 _KEYGEN_REQUIRED_KEYS = {
60 "status", "hub", "account", "name", "msign_path",
61 "public_key_b64", "fingerprint", "hd_seed_b64",
62 }
63 _LIST_ENTRY_REQUIRED_KEYS = {"name", "account", "hub", "msign_path"}
64 _REGISTER_REQUIRED_KEYS = {"status", "name", "account", "hub", "msign_path"}
65
66
67 # ---------------------------------------------------------------------------
68 # Fixtures
69 # ---------------------------------------------------------------------------
70
71
72 @pytest.fixture()
73 def isolated_identity(
74 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
75 ) -> pathlib.Path:
76 fake_dir = tmp_path / "dot_muse"
77 fake_dir.mkdir()
78 fake_file = fake_dir / "identity.toml"
79 monkeypatch.setattr("muse.core.identity._IDENTITY_DIR", fake_dir)
80 monkeypatch.setattr("muse.core.identity._IDENTITY_FILE", fake_file)
81 return fake_dir
82
83
84 @pytest.fixture()
85 def isolated_slots(
86 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
87 ) -> pathlib.Path:
88 fake_dir = tmp_path / "dot_muse_slots"
89 fake_dir.mkdir()
90 fake_file = fake_dir / "agent-slots.toml"
91 monkeypatch.setattr("muse.core.agent_slots._SLOTS_DIR", fake_dir)
92 monkeypatch.setattr("muse.core.agent_slots._SLOTS_FILE", fake_file)
93 return fake_dir
94
95
96 @pytest.fixture()
97 def identity_with_mnemonic(isolated_identity: pathlib.Path) -> None:
98 import muse.core.keychain as _kc
99 from muse.core.identity import IdentityEntry, save_identity
100 _kc.store(_TEST_MNEMONIC)
101 entry: IdentityEntry = {
102 "type": "human",
103 "handle": "gabriel",
104 "hd_path": "m/1075233755'/0'/0'/0'/0'/0'",
105 }
106 save_identity(_TEST_HUB, entry)
107
108
109 def _keygen(
110 *extra_args: str,
111 identity: None = None,
112 slots: None = None,
113 ) -> Mapping[str, object]:
114 result = runner.invoke(
115 cli,
116 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"] + list(extra_args),
117 )
118 assert result.exit_code == 0, f"keygen failed:\n{result.output}"
119 return json.loads(result.output.strip().splitlines()[0])
120
121
122 def _list_slots(slots: None = None) -> list[Mapping[str, str | int | None]]:
123 result = runner.invoke(
124 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
125 )
126 assert result.exit_code == 0, f"list failed:\n{result.output}"
127 return json.loads(result.output.strip().splitlines()[0])["slots"]
128
129
130 def _register(name: str, account: int, slots: None = None) -> Mapping[str, object]:
131 result = runner.invoke(
132 cli,
133 ["agent", "register", "--hub", _TEST_HUB,
134 "--account", str(account), "--name", name, "--json"],
135 )
136 assert result.exit_code == 0, f"register failed:\n{result.output}"
137 return json.loads(result.output.strip().splitlines()[0])
138
139
140 # ---------------------------------------------------------------------------
141 # I keygen schema
142 # ---------------------------------------------------------------------------
143
144
145 class TestKeygenSchemaI:
146 def test_I1_all_required_keys_present(
147 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
148 ) -> None:
149 data = _keygen()
150 missing = _KEYGEN_REQUIRED_KEYS - set(data.keys())
151 assert not missing, f"Missing keys in keygen response: {missing}"
152
153 def test_I2_status_is_ok(
154 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
155 ) -> None:
156 data = _keygen()
157 assert data["status"] == "ok"
158
159 def test_I3_hd_seed_b64_decodes_to_64_bytes(
160 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
161 ) -> None:
162 data = _keygen()
163 raw = b64url_decode(data["hd_seed_b64"])
164 assert len(raw) == 64, f"Expected 64 bytes, got {len(raw)}"
165
166 def test_I4_public_key_b64_decodes_to_32_bytes(
167 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
168 ) -> None:
169 data = _keygen()
170 raw = b64url_decode(data["public_key_b64"])
171 assert len(raw) == 32, f"Expected 32 bytes, got {len(raw)}"
172
173 def test_I5_fingerprint_is_sha256_of_public_key(
174 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
175 ) -> None:
176 data = _keygen()
177 pub_bytes = b64url_decode(data["public_key_b64"])
178 expected = public_key_fingerprint(pub_bytes)
179 assert data["fingerprint"] == expected, (
180 f"Fingerprint mismatch: {data['fingerprint']!r} != {expected!r}"
181 )
182
183 def test_I6_name_is_null_without_name_flag(
184 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
185 ) -> None:
186 data = _keygen()
187 assert data["name"] is None
188
189 def test_I7_name_reflects_name_flag(
190 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
191 ) -> None:
192 result = runner.invoke(
193 cli,
194 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1",
195 "--name", "orchestra", "--json"],
196 )
197 assert result.exit_code == 0
198 data = json.loads(result.output.strip().splitlines()[0])
199 assert data["name"] == "orchestra"
200
201 def test_I8_msign_path_contains_account_index(
202 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
203 ) -> None:
204 result = runner.invoke(
205 cli,
206 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "7", "--json"],
207 )
208 assert result.exit_code == 0
209 data = json.loads(result.output.strip().splitlines()[0])
210 assert "7'" in data["msign_path"]
211 assert data["msign_path"].startswith("m/")
212
213 def test_I9_hub_is_full_url(
214 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
215 ) -> None:
216 data = _keygen()
217 assert data["hub"] == _TEST_HUB
218
219
220 # ---------------------------------------------------------------------------
221 # II list schema
222 # ---------------------------------------------------------------------------
223
224
225 class TestListSchemaII:
226 def _list_data(self, result: "InvokeResult") -> Mapping[str, object]:
227 return json.loads(result.output.strip().splitlines()[0])
228
229 def test_II1_returns_json_object_with_slots(
230 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
231 ) -> None:
232 result = runner.invoke(
233 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
234 )
235 assert result.exit_code == 0
236 data = self._list_data(result)
237 assert isinstance(data, dict)
238 assert "slots" in data
239
240 def test_II2_empty_slots_when_no_slots(
241 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
242 ) -> None:
243 result = runner.invoke(
244 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
245 )
246 assert result.exit_code == 0
247 data = self._list_data(result)
248 assert data["slots"] == []
249
250 def test_II3_each_entry_has_required_keys(
251 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
252 ) -> None:
253 from muse.core.agent_slots import register_slot
254 register_slot(_TEST_HUB, "orchestra", 1)
255 register_slot(_TEST_HUB, "mixer", 2)
256
257 result = runner.invoke(
258 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
259 )
260 assert result.exit_code == 0
261 entries = self._list_data(result)["slots"]
262 assert entries
263 for entry in entries:
264 missing = _LIST_ENTRY_REQUIRED_KEYS - set(entry.keys())
265 assert not missing, f"Missing keys in list entry: {missing}"
266
267 def test_II4_entries_sorted_by_account(
268 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
269 ) -> None:
270 from muse.core.agent_slots import register_slot
271 register_slot(_TEST_HUB, "z-agent", 5)
272 register_slot(_TEST_HUB, "a-agent", 2)
273 register_slot(_TEST_HUB, "m-agent", 9)
274
275 result = runner.invoke(
276 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
277 )
278 assert result.exit_code == 0
279 entries = self._list_data(result)["slots"]
280 accounts = [e["account"] for e in entries]
281 assert accounts == sorted(accounts)
282
283 def test_II5_hub_is_hostname_not_url(
284 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
285 ) -> None:
286 from muse.core.agent_slots import register_slot
287 register_slot(_TEST_HUB, "test-slot", 3)
288
289 result = runner.invoke(
290 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
291 )
292 assert result.exit_code == 0
293 entries = self._list_data(result)["slots"]
294 assert entries
295 for entry in entries:
296 assert entry["hub"] == _TEST_HOSTNAME, (
297 f"Expected hostname {_TEST_HOSTNAME!r}, got {entry['hub']!r}"
298 )
299
300
301 # ---------------------------------------------------------------------------
302 # III register schema
303 # ---------------------------------------------------------------------------
304
305
306 class TestRegisterSchemaIII:
307 def test_III1_all_required_keys_present(
308 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
309 ) -> None:
310 data = _register("orchestra", 1)
311 missing = _REGISTER_REQUIRED_KEYS - set(data.keys())
312 assert not missing, f"Missing keys in register response: {missing}"
313
314 def test_III2_status_is_ok(
315 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
316 ) -> None:
317 data = _register("orchestra", 1)
318 assert data["status"] == "ok"
319
320 def test_III3_hub_is_hostname(
321 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
322 ) -> None:
323 data = _register("test-agent", 4)
324 assert data["hub"] == _TEST_HOSTNAME, (
325 f"Expected hostname {_TEST_HOSTNAME!r}, got {data['hub']!r}"
326 )
327
328 def test_III4_msign_path_contains_account_index(
329 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
330 ) -> None:
331 data = _register("my-agent", 11)
332 assert "11'" in data["msign_path"]
333 assert data["msign_path"].startswith("m/")
334
335
336 # ---------------------------------------------------------------------------
337 # IV Error paths — JSON errors when --json is passed
338 # ---------------------------------------------------------------------------
339
340
341 class TestErrorPathsIV:
342 def test_IV1_keygen_no_identity_json_error(
343 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path,
344 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
345 ) -> None:
346 """No identity registered → exit 1 + JSON error on stdout."""
347 result = runner.invoke(
348 cli,
349 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
350 )
351 assert result.exit_code == 1
352 # The first JSON line on stdout must parse
353 json_line = next(
354 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
355 None,
356 )
357 assert json_line is not None, f"No JSON in output:\n{result.output}"
358 data = json.loads(json_line)
359 assert "error" in data
360
361 def test_IV2_keygen_no_mnemonic_json_error(
362 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path,
363 monkeypatch: pytest.MonkeyPatch
364 ) -> None:
365 """Identity exists but has no mnemonic → exit 1 + JSON error."""
366 # Disable keychain so no leftover entry from a previous test run leaks in.
367 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
368 from muse.core.identity import IdentityEntry, save_identity
369 entry: IdentityEntry = {"type": "human", "handle": "gabriel"}
370 save_identity(_TEST_HUB, entry)
371
372 result = runner.invoke(
373 cli,
374 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
375 )
376 assert result.exit_code == 1
377 json_line = next(
378 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
379 None,
380 )
381 assert json_line is not None, f"No JSON in output:\n{result.output}"
382 data = json.loads(json_line)
383 assert "error" in data
384
385 def test_IV3_keygen_negative_account_json_error(
386 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
387 ) -> None:
388 """Negative account index with --json → exit 1 + JSON error."""
389 result = runner.invoke(
390 cli,
391 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "-1", "--json"],
392 )
393 assert result.exit_code == 1
394 json_line = next(
395 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
396 None,
397 )
398 assert json_line is not None, f"No JSON in output:\n{result.output}"
399 data = json.loads(json_line)
400 assert "error" in data
401
402 def test_IV4_keygen_no_hub_json_error(
403 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path,
404 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
405 ) -> None:
406 """No hub configured, no --hub flag, --json → exit 1 + JSON error."""
407 monkeypatch.chdir(tmp_path)
408 result = runner.invoke(
409 cli,
410 ["agent", "keygen", "--account", "1", "--json"],
411 )
412 assert result.exit_code == 1
413 json_line = next(
414 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
415 None,
416 )
417 assert json_line is not None, f"No JSON in output:\n{result.output}"
418 data = json.loads(json_line)
419 assert "error" in data
420
421 def test_IV5_error_has_error_key(
422 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path
423 ) -> None:
424 """JSON error responses always have an 'error' key."""
425 result = runner.invoke(
426 cli,
427 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
428 )
429 assert result.exit_code == 1
430 json_line = next(
431 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
432 None,
433 )
434 assert json_line is not None
435 data = json.loads(json_line)
436 assert "error" in data, f"No 'error' key in: {data}"
437 assert isinstance(data["error"], str)
438 assert data["error"] # non-empty
439
440 def test_IV6_error_has_message_key(
441 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path
442 ) -> None:
443 """JSON error responses always have a 'message' key."""
444 result = runner.invoke(
445 cli,
446 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
447 )
448 assert result.exit_code == 1
449 json_line = next(
450 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
451 None,
452 )
453 assert json_line is not None
454 data = json.loads(json_line)
455 assert "message" in data, f"No 'message' key in: {data}"
456 assert isinstance(data["message"], str)
457 assert data["message"] # non-empty
File History 1 commit
sha256:ff478cfdcdd4b7fd6de89cb68896601a981f945634463275ec333bd20ca36402 Merge branch 'dev' into main Human 20 days ago