gabriel / muse public
test_agent_json_schema.py python
453 lines 16.2 KB
Raw
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ breaking 146 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
40 import base64
41 import hashlib
42 import json
43 import pathlib
44
45 import pytest
46
47 from tests.cli_test_helper import CliRunner
48
49 cli = None
50 runner = CliRunner()
51
52 _TEST_HUB = "http://localhost:10003"
53 _TEST_HOSTNAME = "localhost:10003"
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 from muse.core.identity import IdentityEntry, save_identity
99 entry: IdentityEntry = {
100 "type": "human",
101 "handle": "gabriel",
102 "key_source": "hd",
103 "mnemonic": _TEST_MNEMONIC,
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 ) -> dict:
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:
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])
128
129
130 def _register(name: str, account: int, slots: None = None) -> dict:
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 = base64.urlsafe_b64decode(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 = base64.urlsafe_b64decode(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 = base64.urlsafe_b64decode(data["public_key_b64"] + "==")
178 expected = hashlib.sha256(pub_bytes).hexdigest()
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 test_II1_returns_json_array(
227 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
228 ) -> None:
229 result = runner.invoke(
230 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
231 )
232 assert result.exit_code == 0
233 data = json.loads(result.output.strip().splitlines()[0])
234 assert isinstance(data, list)
235
236 def test_II2_empty_array_when_no_slots(
237 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
238 ) -> None:
239 result = runner.invoke(
240 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
241 )
242 assert result.exit_code == 0
243 data = json.loads(result.output.strip().splitlines()[0])
244 assert data == []
245
246 def test_II3_each_entry_has_required_keys(
247 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
248 ) -> None:
249 from muse.core.agent_slots import register_slot
250 register_slot(_TEST_HUB, "orchestra", 1)
251 register_slot(_TEST_HUB, "mixer", 2)
252
253 result = runner.invoke(
254 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
255 )
256 assert result.exit_code == 0
257 entries = json.loads(result.output.strip().splitlines()[0])
258 assert entries
259 for entry in entries:
260 missing = _LIST_ENTRY_REQUIRED_KEYS - set(entry.keys())
261 assert not missing, f"Missing keys in list entry: {missing}"
262
263 def test_II4_entries_sorted_by_account(
264 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
265 ) -> None:
266 from muse.core.agent_slots import register_slot
267 register_slot(_TEST_HUB, "z-agent", 5)
268 register_slot(_TEST_HUB, "a-agent", 2)
269 register_slot(_TEST_HUB, "m-agent", 9)
270
271 result = runner.invoke(
272 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
273 )
274 assert result.exit_code == 0
275 entries = json.loads(result.output.strip().splitlines()[0])
276 accounts = [e["account"] for e in entries]
277 assert accounts == sorted(accounts)
278
279 def test_II5_hub_is_hostname_not_url(
280 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
281 ) -> None:
282 from muse.core.agent_slots import register_slot
283 register_slot(_TEST_HUB, "test-slot", 3)
284
285 result = runner.invoke(
286 cli, ["agent", "list", "--hub", _TEST_HUB, "--json"]
287 )
288 assert result.exit_code == 0
289 entries = json.loads(result.output.strip().splitlines()[0])
290 assert entries
291 for entry in entries:
292 assert entry["hub"] == _TEST_HOSTNAME, (
293 f"Expected hostname {_TEST_HOSTNAME!r}, got {entry['hub']!r}"
294 )
295
296
297 # ---------------------------------------------------------------------------
298 # III register schema
299 # ---------------------------------------------------------------------------
300
301
302 class TestRegisterSchemaIII:
303 def test_III1_all_required_keys_present(
304 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
305 ) -> None:
306 data = _register("orchestra", 1)
307 missing = _REGISTER_REQUIRED_KEYS - set(data.keys())
308 assert not missing, f"Missing keys in register response: {missing}"
309
310 def test_III2_status_is_ok(
311 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
312 ) -> None:
313 data = _register("orchestra", 1)
314 assert data["status"] == "ok"
315
316 def test_III3_hub_is_hostname(
317 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
318 ) -> None:
319 data = _register("test-agent", 4)
320 assert data["hub"] == _TEST_HOSTNAME, (
321 f"Expected hostname {_TEST_HOSTNAME!r}, got {data['hub']!r}"
322 )
323
324 def test_III4_msign_path_contains_account_index(
325 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
326 ) -> None:
327 data = _register("my-agent", 11)
328 assert "11'" in data["msign_path"]
329 assert data["msign_path"].startswith("m/")
330
331
332 # ---------------------------------------------------------------------------
333 # IV Error paths — JSON errors when --json is passed
334 # ---------------------------------------------------------------------------
335
336
337 class TestErrorPathsIV:
338 def test_IV1_keygen_no_identity_json_error(
339 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path,
340 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
341 ) -> None:
342 """No identity registered → exit 1 + JSON error on stdout."""
343 result = runner.invoke(
344 cli,
345 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
346 )
347 assert result.exit_code == 1
348 # The first JSON line on stdout must parse
349 json_line = next(
350 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
351 None,
352 )
353 assert json_line is not None, f"No JSON in output:\n{result.output}"
354 data = json.loads(json_line)
355 assert "error" in data
356
357 def test_IV2_keygen_no_mnemonic_json_error(
358 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path,
359 monkeypatch: pytest.MonkeyPatch
360 ) -> None:
361 """Identity exists but has no mnemonic → exit 1 + JSON error."""
362 # Disable keychain so no leftover entry from a previous test run leaks in.
363 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
364 from muse.core.identity import IdentityEntry, save_identity
365 entry: IdentityEntry = {"type": "human", "handle": "gabriel"}
366 save_identity(_TEST_HUB, entry)
367
368 result = runner.invoke(
369 cli,
370 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
371 )
372 assert result.exit_code == 1
373 json_line = next(
374 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
375 None,
376 )
377 assert json_line is not None, f"No JSON in output:\n{result.output}"
378 data = json.loads(json_line)
379 assert "error" in data
380
381 def test_IV3_keygen_negative_account_json_error(
382 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path
383 ) -> None:
384 """Negative account index with --json → exit 1 + JSON error."""
385 result = runner.invoke(
386 cli,
387 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "-1", "--json"],
388 )
389 assert result.exit_code == 1
390 json_line = next(
391 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
392 None,
393 )
394 assert json_line is not None, f"No JSON in output:\n{result.output}"
395 data = json.loads(json_line)
396 assert "error" in data
397
398 def test_IV4_keygen_no_hub_json_error(
399 self, identity_with_mnemonic: None, isolated_slots: pathlib.Path,
400 tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
401 ) -> None:
402 """No hub configured, no --hub flag, --json → exit 1 + JSON error."""
403 monkeypatch.chdir(tmp_path)
404 result = runner.invoke(
405 cli,
406 ["agent", "keygen", "--account", "1", "--json"],
407 )
408 assert result.exit_code == 1
409 json_line = next(
410 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
411 None,
412 )
413 assert json_line is not None, f"No JSON in output:\n{result.output}"
414 data = json.loads(json_line)
415 assert "error" in data
416
417 def test_IV5_error_has_error_key(
418 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path
419 ) -> None:
420 """JSON error responses always have an 'error' key."""
421 result = runner.invoke(
422 cli,
423 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
424 )
425 assert result.exit_code == 1
426 json_line = next(
427 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
428 None,
429 )
430 assert json_line is not None
431 data = json.loads(json_line)
432 assert "error" in data, f"No 'error' key in: {data}"
433 assert isinstance(data["error"], str)
434 assert data["error"] # non-empty
435
436 def test_IV6_error_has_message_key(
437 self, isolated_identity: pathlib.Path, isolated_slots: pathlib.Path
438 ) -> None:
439 """JSON error responses always have a 'message' key."""
440 result = runner.invoke(
441 cli,
442 ["agent", "keygen", "--hub", _TEST_HUB, "--account", "1", "--json"],
443 )
444 assert result.exit_code == 1
445 json_line = next(
446 (ln for ln in result.output.splitlines() if ln.strip().startswith("{")),
447 None,
448 )
449 assert json_line is not None
450 data = json.loads(json_line)
451 assert "message" in data, f"No 'message' key in: {data}"
452 assert isinstance(data["message"], str)
453 assert data["message"] # non-empty
File History 1 commit
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 146 days ago