gabriel / muse public
test_auth_supercharge.py python
1,037 lines 41.3 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 135 days ago
1 """Supercharge tests for ``muse auth``.
2
3 New features under test
4 -----------------------
5 1. ``--type human|agent`` filter on ``whoami --all``
6 2. ``provisioned_by`` field in ``whoami --json`` output (trust chain visibility)
7 3. ``hd_path`` field in ``whoami --json`` output (HD provenance visibility)
8 4. ``_KeygenJson`` TypedDict completeness: ``hd_path``, ``mnemonic_word_count``, ``label``
9 5. ``_ShowJson`` TypedDict completeness: ``algorithm``,
10 ``provisioned_by``, ``provisioned_by_fingerprint``
11 6. Security: invalid ``--type`` value exits non-zero
12 7. ``show --json`` includes ``algorithm`` and ``hd_path``
13
14 Test categories
15 ---------------
16 - unit : TypedDict schema completeness
17 - integration : CLI round-trips via CliRunner with isolated identity files
18 - security : bad --type value rejected, ANSI-safe output
19 - data integrity: provisioned_by, hd_path survive save→load round-trip
20 - performance : _load_all with 50 entries under 200 ms
21 """
22
23 from __future__ import annotations
24 from collections.abc import Mapping
25
26 import json
27 import pathlib
28 import time
29
30 import pytest
31
32 from tests.cli_test_helper import CliRunner, InvokeResult
33 from muse.core.identity import IdentityEntry, save_identity
34
35 cli = None
36 runner = CliRunner()
37
38 HUB = "http://localhost:19111"
39 HOSTNAME = "localhost:19111"
40 HUB2 = "http://localhost:19222"
41 HOSTNAME2 = "localhost:19222"
42
43
44 # ── fixtures ──────────────────────────────────────────────────────────────────
45
46
47 @pytest.fixture
48 def repo(tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> pathlib.Path:
49 """Minimal .muse/ repo with isolated identity home."""
50 from muse._version import __version__
51
52 muse_dir = tmp_path / ".muse"
53 for sub in ("refs/heads", "objects", "commits", "snapshots"):
54 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
55 (muse_dir / "repo.json").write_text(
56 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
57 )
58 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
59 (muse_dir / "refs" / "heads" / "main").write_text("")
60 (muse_dir / "config.toml").write_text("")
61
62 muse_home = tmp_path / ".muse-home"
63 muse_home.mkdir()
64 (muse_home / "identity.toml").write_text("")
65
66 import muse.core.identity as _id_mod
67 monkeypatch.setattr(_id_mod, "_IDENTITY_FILE", muse_home / "identity.toml")
68 monkeypatch.setattr(_id_mod, "_IDENTITY_DIR", muse_home)
69 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
70 monkeypatch.chdir(tmp_path)
71 return tmp_path
72
73
74 def _human(handle: str = "alice") -> IdentityEntry:
75 return {
76 "type": "human",
77 "handle": handle,
78 "algorithm": "ed25519",
79 "fingerprint": "a" * 64,
80 }
81
82
83 def _agent(handle: str = "bot", provisioned_by: str = "alice") -> IdentityEntry:
84 return {
85 "type": "agent",
86 "handle": handle,
87 "algorithm": "ed25519",
88 "fingerprint": "b" * 64,
89 "provisioned_by": provisioned_by,
90 }
91
92
93 def _parse_json(result: InvokeResult) -> Mapping[str, object]:
94 """Parse first JSON structure from result.output (object or array)."""
95 for line in result.output.splitlines():
96 stripped = line.strip()
97 if stripped.startswith("{") or stripped.startswith("["):
98 return json.loads(stripped)
99 raise ValueError(f"No JSON in output:\n{result.output!r}")
100
101
102 # ── Unit: _KeygenJson TypedDict completeness ──────────────────────────────────
103
104
105 class TestKeygenJsonTypedDict:
106 """_KeygenJson TypedDict must declare hd_path and mnemonic_word_count."""
107
108 def test_hd_path_in_keygen_typeddict(self) -> None:
109 from muse.cli.commands.auth import _KeygenJson
110 hints = _KeygenJson.__annotations__
111 assert "hd_path" in hints, (
112 "_KeygenJson missing hd_path — run_keygen already emits it"
113 )
114
115 def test_mnemonic_word_count_in_keygen_typeddict(self) -> None:
116 from muse.cli.commands.auth import _KeygenJson
117 hints = _KeygenJson.__annotations__
118 assert "mnemonic_word_count" in hints, (
119 "_KeygenJson missing mnemonic_word_count — run_keygen already emits it"
120 )
121
122 def test_all_required_keys_present(self) -> None:
123 from muse.cli.commands.auth import _KeygenJson
124 hints = _KeygenJson.__annotations__
125 for key in ("status", "hub", "hostname", "public_key_b64",
126 "fingerprint", "hd_path", "mnemonic_word_count"):
127 assert key in hints, f"_KeygenJson missing key: {key!r}"
128
129
130 # ── Unit: _ShowJson TypedDict completeness ────────────────────────────────────
131
132
133 class TestShowJsonTypedDict:
134 """_ShowJson TypedDict must expose algorithm, provisioned_by,
135 provisioned_by_fingerprint — fields that identity.toml stores but the
136 show command didn't surface."""
137
138 def test_algorithm_in_show_typeddict(self) -> None:
139 from muse.cli.commands.auth import _ShowJson
140 assert "algorithm" in _ShowJson.__annotations__, "_ShowJson missing algorithm"
141
142 def test_provisioned_by_in_show_typeddict(self) -> None:
143 from muse.cli.commands.auth import _ShowJson
144 assert "provisioned_by" in _ShowJson.__annotations__, \
145 "_ShowJson missing provisioned_by (agent trust chain)"
146
147 def test_provisioned_by_fingerprint_in_show_typeddict(self) -> None:
148 from muse.cli.commands.auth import _ShowJson
149 assert "provisioned_by_fingerprint" in _ShowJson.__annotations__, \
150 "_ShowJson missing provisioned_by_fingerprint"
151
152
153 # ── Unit: _WhoamiJson has provisioned_by and hd_path ─────────────────────────
154
155
156 class TestWhoamiJsonTypedDict:
157 def test_provisioned_by_in_whoami_typeddict(self) -> None:
158 from muse.cli.commands.auth import _WhoamiJson
159 assert "provisioned_by" in _WhoamiJson.__annotations__, \
160 "_WhoamiJson missing provisioned_by"
161
162 def test_hd_path_in_whoami_typeddict(self) -> None:
163 from muse.cli.commands.auth import _WhoamiJson
164 assert "hd_path" in _WhoamiJson.__annotations__, \
165 "_WhoamiJson missing hd_path"
166
167
168 # ── Integration: _display_entry emits provisioned_by for agents ───────────────
169
170
171 class TestDisplayEntryProvisionedBy:
172 def test_agent_provisioned_by_in_json(self, capsys: pytest.CaptureFixture[str]) -> None:
173 from muse.cli.commands.auth import _display_entry
174 entry: IdentityEntry = {
175 "type": "agent",
176 "handle": "bot",
177 "algorithm": "ed25519",
178 "fingerprint": "b" * 64,
179 "provisioned_by": "alice",
180 }
181 _display_entry(HOSTNAME, entry, json_output=True)
182 data = json.loads(capsys.readouterr().out)
183 assert data.get("provisioned_by") == "alice"
184
185 def test_human_no_provisioned_by(self, capsys: pytest.CaptureFixture[str]) -> None:
186 from muse.cli.commands.auth import _display_entry
187 entry: IdentityEntry = _human()
188 _display_entry(HOSTNAME, entry, json_output=True)
189 data = json.loads(capsys.readouterr().out)
190 assert "provisioned_by" not in data or data.get("provisioned_by") == ""
191
192 def test_agent_hd_path_in_json(self, capsys: pytest.CaptureFixture[str]) -> None:
193 from muse.cli.commands.auth import _display_entry
194 hd_path = "m/1075233755'/0'/0'/0'/0'/0'"
195 entry: IdentityEntry = {
196 "type": "human",
197 "handle": "gabriel",
198 "algorithm": "ed25519",
199 "fingerprint": "a" * 64,
200 "hd_path": hd_path,
201 }
202 _display_entry(HOSTNAME, entry, json_output=True)
203 data = json.loads(capsys.readouterr().out)
204 assert data.get("hd_path") == hd_path
205
206 def test_no_hd_path_absent_from_json(self, capsys: pytest.CaptureFixture[str]) -> None:
207 from muse.cli.commands.auth import _display_entry
208 entry: IdentityEntry = _human()
209 _display_entry(HOSTNAME, entry, json_output=True)
210 data = json.loads(capsys.readouterr().out)
211 assert "hd_path" not in data or data.get("hd_path") == ""
212
213
214 # ── Integration: whoami --all --type filter ───────────────────────────────────
215
216
217 class TestWhoamiTypeFilter:
218 """``muse auth whoami --all --type TYPE`` filters by identity type."""
219
220 def test_type_human_returns_only_humans(self, repo: pathlib.Path) -> None:
221 save_identity(HUB, _human("alice"))
222 save_identity(HUB2, _agent("bot", "alice"))
223 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "human", "--json"])
224 assert result.exit_code == 0, result.output
225 data = json.loads(result.output)["identities"]
226 assert isinstance(data, list)
227 assert all(e["type"] == "human" for e in data), f"non-human in results: {data}"
228 handles = {e["handle"] for e in data}
229 assert "alice" in handles
230 assert "bot" not in handles
231
232 def test_type_agent_returns_only_agents(self, repo: pathlib.Path) -> None:
233 save_identity(HUB, _human("alice"))
234 save_identity(HUB2, _agent("bot", "alice"))
235 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "agent", "--json"])
236 assert result.exit_code == 0, result.output
237 data = json.loads(result.output)["identities"]
238 assert isinstance(data, list)
239 assert all(e["type"] == "agent" for e in data)
240 handles = {e["handle"] for e in data}
241 assert "bot" in handles
242 assert "alice" not in handles
243
244 def test_type_filter_no_match_exits_nonzero(self, repo: pathlib.Path) -> None:
245 """--type agent when only humans are stored → exit nonzero."""
246 save_identity(HUB, _human("alice"))
247 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "agent"])
248 assert result.exit_code != 0
249
250 def test_type_filter_no_match_json_exits_nonzero(self, repo: pathlib.Path) -> None:
251 save_identity(HUB, _human("alice"))
252 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "agent", "--json"])
253 assert result.exit_code != 0
254
255 def test_type_invalid_value_exits_nonzero(self, repo: pathlib.Path) -> None:
256 """--type must accept only 'human' or 'agent'."""
257 save_identity(HUB, _human())
258 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "superuser"])
259 assert result.exit_code != 0
260
261 def test_type_requires_all_flag(self, repo: pathlib.Path) -> None:
262 """--type without --all should fail or be ignored gracefully."""
263 save_identity(HUB, _human())
264 result = runner.invoke(cli, ["auth", "whoami", "--hub", HUB, "--type", "human", "--json"])
265 # Either succeeds (type flag ignored for single-hub) or fails cleanly
266 # Most important: no crash / traceback
267 assert result.exit_code in (0, 1), f"Unexpected exit code: {result.exit_code}"
268
269 def test_type_filter_counts_correctly(self, repo: pathlib.Path) -> None:
270 """3 humans + 2 agents; --type human → 3 results."""
271 hubs = [f"http://localhost:{19111 + i}" for i in range(5)]
272 for i, hub in enumerate(hubs):
273 if i < 3:
274 save_identity(hub, _human(f"human-{i}"))
275 else:
276 save_identity(hub, _agent(f"bot-{i}", "operator"))
277 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "human", "--json"])
278 assert result.exit_code == 0
279 data = json.loads(result.output)["identities"]
280 assert len(data) == 3
281
282 def test_type_agent_includes_provisioned_by(self, repo: pathlib.Path) -> None:
283 """Agent entries in --type agent output expose provisioned_by."""
284 save_identity(HUB, _agent("bot", "alice"))
285 result = runner.invoke(cli, ["auth", "whoami", "--all", "--type", "agent", "--json"])
286 assert result.exit_code == 0
287 data = json.loads(result.output)["identities"]
288 assert len(data) == 1
289 assert data[0].get("provisioned_by") == "alice"
290
291
292 # ── Integration: whoami --json includes provisioned_by for agents ─────────────
293
294
295 class TestWhoamiProvisionedBy:
296 def test_whoami_json_agent_has_provisioned_by(self, repo: pathlib.Path) -> None:
297 save_identity(HUB, _agent("bot", "alice"))
298 result = runner.invoke(cli, ["auth", "whoami", "--hub", HUB, "--json"])
299 assert result.exit_code == 0
300 data = json.loads(result.output)
301 assert data.get("provisioned_by") == "alice"
302
303 def test_whoami_json_human_no_provisioned_by(self, repo: pathlib.Path) -> None:
304 save_identity(HUB, _human())
305 result = runner.invoke(cli, ["auth", "whoami", "--hub", HUB, "--json"])
306 assert result.exit_code == 0
307 data = json.loads(result.output)
308 assert "provisioned_by" not in data or not data["provisioned_by"]
309
310 def test_whoami_all_json_agent_has_provisioned_by(self, repo: pathlib.Path) -> None:
311 save_identity(HUB, _agent("bot", "alice"))
312 result = runner.invoke(cli, ["auth", "whoami", "--all", "--json"])
313 assert result.exit_code == 0
314 entries = json.loads(result.output)["identities"]
315 bot = next(e for e in entries if e["handle"] == "bot")
316 assert bot.get("provisioned_by") == "alice"
317
318
319 # ── Integration: whoami --json includes hd_path when present ──────────────────
320
321
322 class TestWhoamiHdPath:
323 def test_whoami_json_hd_entry_has_hd_path(self, repo: pathlib.Path) -> None:
324 hd_path = "m/1075233755'/0'/0'/0'/0'/0'"
325 entry: IdentityEntry = {
326 "type": "human",
327 "handle": "gabriel",
328 "algorithm": "ed25519",
329 "fingerprint": "a" * 64,
330 "hd_path": hd_path,
331 }
332 save_identity(HUB, entry)
333 result = runner.invoke(cli, ["auth", "whoami", "--hub", HUB, "--json"])
334 assert result.exit_code == 0
335 data = json.loads(result.output)
336 assert data.get("hd_path") == hd_path
337
338 def test_whoami_json_no_hd_path_when_absent(self, repo: pathlib.Path) -> None:
339 save_identity(HUB, _human())
340 result = runner.invoke(cli, ["auth", "whoami", "--hub", HUB, "--json"])
341 assert result.exit_code == 0
342 data = json.loads(result.output)
343 assert "hd_path" not in data or not data["hd_path"]
344
345
346 # ── Integration: show --json includes algorithm ───────────────────────────────
347
348
349 class TestShowJsonKeyPathAlgorithm:
350 def test_show_json_has_algorithm(self, repo: pathlib.Path) -> None:
351 save_identity(HUB, _human())
352 result = runner.invoke(cli, ["auth", "show", "--hub", HUB, "--json"])
353 assert result.exit_code == 0
354 data = json.loads(result.output)
355 assert "algorithm" in data, f"show --json missing algorithm; got: {list(data)}"
356
357 def test_show_json_algorithm_value_correct(self, repo: pathlib.Path) -> None:
358 save_identity(HUB, _human())
359 result = runner.invoke(cli, ["auth", "show", "--hub", HUB, "--json"])
360 assert result.exit_code == 0
361 data = json.loads(result.output)
362 assert data["algorithm"] == "ed25519"
363
364 def test_show_json_agent_has_provisioned_by(self, repo: pathlib.Path) -> None:
365 save_identity(HUB, _agent("bot", "alice"))
366 result = runner.invoke(cli, ["auth", "show", "--hub", HUB, "--json"])
367 assert result.exit_code == 0
368 data = json.loads(result.output)
369 assert data.get("provisioned_by") == "alice"
370
371
372 # ── Security: type filter input validation ────────────────────────────────────
373
374
375 class TestTypeFilterSecurity:
376 def test_type_with_ansi_injection_rejected(self, repo: pathlib.Path) -> None:
377 save_identity(HUB, _human())
378 result = runner.invoke(
379 cli, ["auth", "whoami", "--all", "--type", "\x1b[31mhuman\x1b[0m"]
380 )
381 assert result.exit_code != 0
382
383 def test_type_with_newline_injection_rejected(self, repo: pathlib.Path) -> None:
384 save_identity(HUB, _human())
385 result = runner.invoke(
386 cli, ["auth", "whoami", "--all", "--type", "human\nevil"]
387 )
388 assert result.exit_code != 0
389
390 def test_type_with_semicolon_rejected(self, repo: pathlib.Path) -> None:
391 save_identity(HUB, _human())
392 result = runner.invoke(
393 cli, ["auth", "whoami", "--all", "--type", "human;rm -rf /"]
394 )
395 assert result.exit_code != 0
396
397
398 # ── Performance: _load_all with 50 entries under 200 ms ──────────────────────
399
400
401 class TestLoadAllPerformance:
402 def test_50_entries_under_200ms(self, tmp_path: pathlib.Path) -> None:
403 from muse.core.identity import _load_all, _dump_identity
404
405 entries: dict[str, IdentityEntry] = {}
406 for i in range(50):
407 hostname = f"localhost:{19500 + i}"
408 entries[hostname] = {
409 "type": "human",
410 "handle": f"user-{i:02d}",
411 "algorithm": "ed25519",
412 "fingerprint": "a" * 64,
413 "hd_path": "m/1075233755'/0'/0'/0'/0'/0'",
414 }
415
416 p = tmp_path / "identity.toml"
417 p.write_text(_dump_identity(entries))
418
419 start = time.monotonic()
420 loaded = _load_all(p)
421 elapsed = time.monotonic() - start
422
423 assert len(loaded) == 50
424 assert elapsed < 0.2, f"_load_all with 50 entries took {elapsed:.3f}s"
425
426
427 # ── Stress: show with 50+ identities ─────────────────────────────────────────
428
429
430 class TestShowStress:
431 """show must handle large identity files without corruption."""
432
433 def test_show_with_50_identities_returns_correct_entry(
434 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
435 ) -> None:
436 """show --hub TARGET picks the right entry from a file with 50 entries."""
437 import muse.core.identity as _id_mod
438 from muse.core.identity import _dump_identity
439
440 identity_file = tmp_path / "identity.toml"
441 monkeypatch.setattr(_id_mod, "_IDENTITY_FILE", identity_file)
442 monkeypatch.setattr(_id_mod, "_IDENTITY_DIR", tmp_path)
443 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
444 monkeypatch.chdir(tmp_path)
445
446 # Create minimal .muse structure
447 from muse._version import __version__
448 muse_dir = tmp_path / ".muse"
449 for sub in ("refs/heads", "objects", "commits", "snapshots"):
450 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
451 (muse_dir / "repo.json").write_text(
452 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
453 )
454 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
455 (muse_dir / "refs" / "heads" / "main").write_text("")
456 (muse_dir / "config.toml").write_text("")
457
458 entries: dict[str, IdentityEntry] = {}
459 target_hub = "http://localhost:20050"
460 for i in range(50):
461 hub = f"http://localhost:{20000 + i}"
462 entries[f"localhost:{20000 + i}"] = {
463 "type": "human",
464 "handle": f"user-{i:02d}",
465 "algorithm": "ed25519",
466 "fingerprint": "a" * 64,
467 }
468 # Override one specific entry as the target
469 entries["localhost:20050"] = {
470 "type": "human",
471 "handle": "target-user",
472 "algorithm": "ed25519",
473 "fingerprint": "f" * 64,
474 }
475 identity_file.write_text(_dump_identity(entries))
476
477 result = runner.invoke(cli, ["auth", "show", "--hub", target_hub, "--json"])
478 assert result.exit_code == 0, result.output
479 data = json.loads(result.output)
480 assert data["handle"] == "target-user"
481 assert data["fingerprint"] == "f" * 64
482
483 def test_show_50_repeated_calls_consistent(
484 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
485 ) -> None:
486 """50 consecutive show calls return identical results."""
487 import muse.core.identity as _id_mod
488 from muse.core.identity import _dump_identity
489
490 identity_file = tmp_path / "identity.toml"
491 monkeypatch.setattr(_id_mod, "_IDENTITY_FILE", identity_file)
492 monkeypatch.setattr(_id_mod, "_IDENTITY_DIR", tmp_path)
493 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
494 monkeypatch.chdir(tmp_path)
495
496 from muse._version import __version__
497 muse_dir = tmp_path / ".muse"
498 for sub in ("refs/heads", "objects", "commits", "snapshots"):
499 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
500 (muse_dir / "repo.json").write_text(
501 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
502 )
503 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
504 (muse_dir / "refs" / "heads" / "main").write_text("")
505 (muse_dir / "config.toml").write_text("")
506
507 entry: IdentityEntry = {
508 "type": "human",
509 "handle": "stable-user",
510 "algorithm": "ed25519",
511 "fingerprint": "b" * 64,
512 }
513 identity_file.write_text(_dump_identity({"localhost:20099": entry}))
514
515 results = set()
516 for _ in range(50):
517 r = runner.invoke(cli, ["auth", "show", "--hub", "http://localhost:20099", "--json"])
518 assert r.exit_code == 0
519 d = json.loads(r.output)
520 d.pop("duration_ms", None)
521 d.pop("timestamp", None)
522 results.add(json.dumps(d, sort_keys=True))
523
524 assert len(results) == 1, "show returned different output across 50 calls"
525
526
527 # ── Stress: logout clear_all_identities with many hubs ───────────────────────
528
529
530 class TestLogoutStress:
531 """logout --all must atomically clear arbitrarily many entries."""
532
533 def test_logout_all_50_hubs(
534 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
535 ) -> None:
536 """logout --all removes all 50 entries in one shot."""
537 import muse.core.identity as _id_mod
538 from muse.core.identity import _dump_identity
539
540 identity_file = tmp_path / "identity.toml"
541 monkeypatch.setattr(_id_mod, "_IDENTITY_FILE", identity_file)
542 monkeypatch.setattr(_id_mod, "_IDENTITY_DIR", tmp_path)
543 monkeypatch.setenv("MUSE_REPO_ROOT", str(tmp_path))
544 monkeypatch.chdir(tmp_path)
545
546 from muse._version import __version__
547 muse_dir = tmp_path / ".muse"
548 for sub in ("refs/heads", "objects", "commits", "snapshots"):
549 (muse_dir / sub).mkdir(parents=True, exist_ok=True)
550 (muse_dir / "repo.json").write_text(
551 json.dumps({"repo_id": "test-repo", "schema_version": __version__, "domain": "code"})
552 )
553 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
554 (muse_dir / "refs" / "heads" / "main").write_text("")
555 (muse_dir / "config.toml").write_text("")
556
557 entries: dict[str, IdentityEntry] = {}
558 for i in range(50):
559 entries[f"localhost:{21000 + i}"] = {
560 "type": "human",
561 "handle": f"user-{i:02d}",
562 "algorithm": "ed25519",
563 "fingerprint": "a" * 64,
564 }
565 identity_file.write_text(_dump_identity(entries))
566
567 result = runner.invoke(cli, ["auth", "logout", "--all", "--json"])
568 assert result.exit_code == 0, result.output
569 data = json.loads(result.output)
570 assert data["status"] == "ok"
571 assert data["count"] == 50
572 assert len(data["hubs"]) == 50
573
574 # File should be empty now
575 remaining = identity_file.read_text().strip()
576 assert remaining == "", f"identity.toml not cleared: {remaining!r}"
577
578
579 # ── Performance: logout --all with 50 hubs under 100 ms ──────────────────────
580
581
582 class TestLogoutPerformance:
583 def test_logout_all_50_hubs_under_100ms(
584 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
585 ) -> None:
586 import muse.core.identity as _id_mod
587 from muse.core.identity import _dump_identity, clear_all_identities
588
589 identity_file = tmp_path / "identity.toml"
590 monkeypatch.setattr(_id_mod, "_IDENTITY_FILE", identity_file)
591 monkeypatch.setattr(_id_mod, "_IDENTITY_DIR", tmp_path)
592
593 entries: dict[str, IdentityEntry] = {}
594 for i in range(50):
595 entries[f"localhost:{22000 + i}"] = {
596 "type": "human",
597 "handle": f"user-{i:02d}",
598 "algorithm": "ed25519",
599 "fingerprint": "a" * 64,
600 }
601 identity_file.write_text(_dump_identity(entries))
602
603 start = time.monotonic()
604 removed = clear_all_identities()
605 elapsed = time.monotonic() - start
606
607 assert len(removed) == 50
608 assert elapsed < 0.1, f"clear_all_identities(50) took {elapsed:.3f}s"
609
610
611 # ── Data integrity: recover produces identical fingerprint from same mnemonic ─
612
613
614 class TestRecoverDataIntegrity:
615 """Recovering from the same mnemonic must reproduce the same fingerprint."""
616
617 _MNEMONIC = (
618 "abandon abandon abandon abandon abandon abandon abandon abandon "
619 "abandon abandon abandon about"
620 )
621
622 def _patch(
623 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
624 ) -> None:
625 import muse.core.keypair as kp_mod
626 import muse.core.identity as id_mod
627
628 fake_home = tmp_path / "home"
629 fake_home.mkdir(parents=True, exist_ok=True)
630 import pathlib as _pl
631 monkeypatch.setattr(_pl.Path, "home", staticmethod(lambda: fake_home))
632 monkeypatch.setattr(kp_mod, "_KEYS_DIR", fake_home / ".muse" / "keys")
633 monkeypatch.setattr(id_mod, "_IDENTITY_DIR", fake_home / ".muse")
634 monkeypatch.setattr(id_mod, "_IDENTITY_FILE", fake_home / ".muse" / "identity.toml")
635 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
636
637 def test_same_mnemonic_same_fingerprint(
638 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
639 ) -> None:
640 """Recovering twice from the same mnemonic gives the same fingerprint."""
641 self._patch(monkeypatch, tmp_path)
642
643 r1 = runner.invoke(
644 cli,
645 ["auth", "recover", "--hub", "http://localhost:19911", "--json"],
646 input=self._MNEMONIC + "\n",
647 )
648 assert r1.exit_code == 0, r1.output
649 d1 = json.loads(r1.output)
650
651 # Force-overwrite on second recover
652 r2 = runner.invoke(
653 cli,
654 ["auth", "recover", "--hub", "http://localhost:19911", "--force", "--json"],
655 input=self._MNEMONIC + "\n",
656 )
657 assert r2.exit_code == 0, r2.output
658 d2 = json.loads(r2.output)
659
660 assert d1["fingerprint"] == d2["fingerprint"], (
661 f"Fingerprint changed between recoveries: {d1['fingerprint']} vs {d2['fingerprint']}"
662 )
663 assert d1["public_key_b64"] == d2["public_key_b64"]
664
665 def test_different_mnemonic_different_fingerprint(
666 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
667 ) -> None:
668 """A different mnemonic produces a different fingerprint."""
669 self._patch(monkeypatch, tmp_path)
670
671 mnemonic_b = (
672 "abandon abandon abandon abandon abandon abandon abandon abandon "
673 "abandon abandon abandon zoo" # intentionally invalid — just needs to pass BIP39
674 )
675 # Use the canonical 12-word test vector for second recover
676 mnemonic_b = (
677 "zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo zoo wrong"
678 )
679
680 r1 = runner.invoke(
681 cli,
682 ["auth", "recover", "--hub", "http://localhost:19912", "--json"],
683 input=self._MNEMONIC + "\n",
684 )
685 assert r1.exit_code == 0, r1.output
686
687 r2 = runner.invoke(
688 cli,
689 ["auth", "recover", "--hub", "http://localhost:19913", "--json"],
690 input=mnemonic_b + "\n",
691 )
692 assert r2.exit_code == 0, r2.output
693
694 d1 = json.loads(r1.output)
695 d2 = json.loads(r2.output)
696 assert d1["fingerprint"] != d2["fingerprint"], (
697 "Different mnemonics must not produce the same fingerprint"
698 )
699
700 def test_recover_hd_path_persisted(
701 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
702 ) -> None:
703 """After recover, identity.toml must contain hd_path."""
704 import muse.core.identity as id_mod
705 self._patch(monkeypatch, tmp_path)
706
707 r = runner.invoke(
708 cli,
709 ["auth", "recover", "--hub", "http://localhost:19914", "--json"],
710 input=self._MNEMONIC + "\n",
711 )
712 assert r.exit_code == 0, r.output
713 data = json.loads(r.output)
714 assert "hd_path" in data
715 assert data["hd_path"].startswith("m/")
716
717 # Also verify TOML on disk
718 loaded = id_mod.load_identity("http://localhost:19914")
719 assert loaded is not None
720 assert loaded.get("hd_path", "").startswith("m/")
721
722 def test_recover_mnemonic_not_in_toml(
723 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
724 ) -> None:
725 """Mnemonic must not be written to identity.toml during recover."""
726 import muse.core.identity as id_mod
727 self._patch(monkeypatch, tmp_path)
728
729 r = runner.invoke(
730 cli,
731 ["auth", "recover", "--hub", "http://localhost:19915"],
732 input=self._MNEMONIC + "\n",
733 )
734 assert r.exit_code == 0, r.output
735
736 import re
737 raw = id_mod._IDENTITY_FILE.read_text()
738 assert re.search(r'^\s*mnemonic\s*=', raw, re.MULTILINE) is None, (
739 f"mnemonic written to TOML:\n{raw}"
740 )
741 assert self._MNEMONIC not in raw
742
743
744 # ── Stress: recover same hub 10 times (--force) ───────────────────────────────
745
746
747 class TestRecoverStress:
748 _MNEMONIC = (
749 "abandon abandon abandon abandon abandon abandon abandon abandon "
750 "abandon abandon abandon about"
751 )
752
753 def _patch(
754 self, monkeypatch: pytest.MonkeyPatch, tmp_path: pathlib.Path
755 ) -> None:
756 import muse.core.keypair as kp_mod
757 import muse.core.identity as id_mod
758
759 fake_home = tmp_path / "home"
760 fake_home.mkdir(parents=True, exist_ok=True)
761 import pathlib as _pl
762 monkeypatch.setattr(_pl.Path, "home", staticmethod(lambda: fake_home))
763 monkeypatch.setattr(kp_mod, "_KEYS_DIR", fake_home / ".muse" / "keys")
764 monkeypatch.setattr(id_mod, "_IDENTITY_DIR", fake_home / ".muse")
765 monkeypatch.setattr(id_mod, "_IDENTITY_FILE", fake_home / ".muse" / "identity.toml")
766 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
767
768 def test_10_recoveries_same_fingerprint(
769 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
770 ) -> None:
771 """10 forced recoveries from the same mnemonic must all yield the same fingerprint."""
772 self._patch(monkeypatch, tmp_path)
773
774 fingerprints: list[str] = []
775 for i in range(10):
776 flags = ["auth", "recover", "--hub", "http://localhost:19920", "--json"]
777 if i > 0:
778 flags.append("--force")
779 r = runner.invoke(cli, flags, input=self._MNEMONIC + "\n")
780 assert r.exit_code == 0, f"recover #{i} failed:\n{r.output}"
781 fingerprints.append(json.loads(r.output)["fingerprint"])
782
783 assert len(set(fingerprints)) == 1, (
784 f"Fingerprint varied across 10 recoveries: {set(fingerprints)}"
785 )
786
787 def test_10_recoveries_different_hubs(
788 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
789 ) -> None:
790 """Recover for 10 different hubs from the same mnemonic — all succeed."""
791 self._patch(monkeypatch, tmp_path)
792
793 fingerprints: list[str] = []
794 for i in range(10):
795 hub = f"http://localhost:{19930 + i}"
796 r = runner.invoke(
797 cli,
798 ["auth", "recover", "--hub", hub, "--json"],
799 input=self._MNEMONIC + "\n",
800 )
801 assert r.exit_code == 0, f"recover for {hub} failed:\n{r.output}"
802 fingerprints.append(json.loads(r.output)["fingerprint"])
803
804 # All 10 should produce the same fingerprint (same mnemonic, human key)
805 assert len(set(fingerprints)) == 1, (
806 "Same mnemonic should give same fingerprint regardless of hub hostname"
807 )
808
809
810 # ── Performance: recover completes within SLA ─────────────────────────────────
811
812
813 class TestRecoverPerformance:
814 _MNEMONIC = (
815 "abandon abandon abandon abandon abandon abandon abandon abandon "
816 "abandon abandon abandon about"
817 )
818
819 def test_recover_under_3_seconds(
820 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
821 ) -> None:
822 """Full recover (PBKDF2 + SLIP-0010 + PEM write) must complete within 3 s."""
823 import muse.core.keypair as kp_mod
824 import muse.core.identity as id_mod
825 import pathlib as _pl
826
827 fake_home = tmp_path / "home"
828 fake_home.mkdir(parents=True, exist_ok=True)
829 monkeypatch.setattr(_pl.Path, "home", staticmethod(lambda: fake_home))
830 monkeypatch.setattr(kp_mod, "_KEYS_DIR", fake_home / ".muse" / "keys")
831 monkeypatch.setattr(id_mod, "_IDENTITY_DIR", fake_home / ".muse")
832 monkeypatch.setattr(id_mod, "_IDENTITY_FILE", fake_home / ".muse" / "identity.toml")
833 monkeypatch.setenv("MUSE_KEYCHAIN_BACKEND", "disabled")
834
835 start = time.monotonic()
836 result = runner.invoke(
837 cli,
838 ["auth", "recover", "--hub", "http://localhost:19940", "--json"],
839 input=self._MNEMONIC + "\n",
840 )
841 elapsed = time.monotonic() - start
842
843 assert result.exit_code == 0, result.output
844 assert elapsed < 3.0, f"recover took {elapsed:.2f}s — exceeds 3 s SLA"
845
846
847 # ── Performance: register latency (stubbed network) ───────────────────────────
848
849
850 class TestRegisterPerformance:
851 """register with a mocked hub must complete within 500 ms."""
852
853 def test_register_under_500ms(
854 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
855 ) -> None:
856 import muse.core.keypair as kp_mod
857 import muse.core.identity as id_mod
858 import pathlib as _pl
859 import unittest.mock
860 import urllib.request
861
862 fake_home = tmp_path / "home"
863 fake_home.mkdir(parents=True, exist_ok=True)
864 monkeypatch.setattr(_pl.Path, "home", staticmethod(lambda: fake_home))
865 monkeypatch.setattr(kp_mod, "_KEYS_DIR", fake_home / ".muse" / "keys")
866 monkeypatch.setattr(id_mod, "_IDENTITY_DIR", fake_home / ".muse")
867 monkeypatch.setattr(id_mod, "_IDENTITY_FILE", fake_home / ".muse" / "identity.toml")
868
869 # Pre-set identity entry and patch keychain so register can sign
870 import muse.core.keychain as _kc
871 _MNEMONIC = "abandon " * 11 + "about"
872 monkeypatch.setattr(_kc, "is_available", lambda: True)
873 monkeypatch.setattr(_kc, "load", lambda: _MNEMONIC)
874 id_mod.save_identity("http://localhost:19950", {
875 "type": "human",
876 "handle": "perf-user",
877 "algorithm": "ed25519",
878 "hd_path": "m/1075233755'/0'/0'/0'/0'/0'",
879 "fingerprint": "a" * 64,
880 })
881
882 # Mock challenge-response round-trip
883 def _fake_urlopen(req, *args, **kwargs):
884 url = req.full_url if hasattr(req, "full_url") else str(req)
885 if "challenge" in url:
886 body = json.dumps({
887 "challengeToken": "deadbeef" * 8, # 64-char hex nonce
888 "isNewKey": True,
889 "algorithm": "ed25519",
890 }).encode()
891 else:
892 body = json.dumps({
893 "token": "test-auth-token",
894 "handle": "perf-user",
895 "identityId": "id-123",
896 "isNewIdentity": True,
897 "authMethod": "ed25519",
898 }).encode()
899
900 class _Resp:
901 def __init__(self) -> None:
902 self.status = 200
903 def read(self, n: int = -1) -> bytes:
904 return body
905 def __enter__(self): return self
906 def __exit__(self, *a): pass
907
908 return _Resp()
909
910 monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
911
912 start = time.monotonic()
913 result = runner.invoke(
914 cli,
915 ["auth", "register", "--hub", "http://localhost:19950",
916 "--handle", "perf-user", "--json"],
917 )
918 elapsed = time.monotonic() - start
919
920 assert result.exit_code == 0, f"register failed:\n{result.output}"
921 assert elapsed < 0.5, f"register took {elapsed:.3f}s — exceeds 500 ms SLA"
922
923
924 # ── Stress: register repeated calls to same hub ───────────────────────────────
925
926
927 class TestRegisterStress:
928 """register is idempotent — repeated calls with --force succeed."""
929
930 def test_5_register_calls_same_hub(
931 self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch
932 ) -> None:
933 import muse.core.keypair as kp_mod
934 import muse.core.identity as id_mod
935 import pathlib as _pl
936 import urllib.request
937
938 fake_home = tmp_path / "home"
939 fake_home.mkdir(parents=True, exist_ok=True)
940 monkeypatch.setattr(_pl.Path, "home", staticmethod(lambda: fake_home))
941 monkeypatch.setattr(kp_mod, "_KEYS_DIR", fake_home / ".muse" / "keys")
942 monkeypatch.setattr(id_mod, "_IDENTITY_DIR", fake_home / ".muse")
943 monkeypatch.setattr(id_mod, "_IDENTITY_FILE", fake_home / ".muse" / "identity.toml")
944
945 # Pre-set identity entry and patch keychain so register can sign
946 import muse.core.keychain as _kc
947 _MNEMONIC = "abandon " * 11 + "about"
948 monkeypatch.setattr(_kc, "is_available", lambda: True)
949 monkeypatch.setattr(_kc, "load", lambda: _MNEMONIC)
950 id_mod.save_identity("http://localhost:19960", {
951 "type": "human",
952 "handle": "stress-user",
953 "algorithm": "ed25519",
954 "hd_path": "m/1075233755'/0'/0'/0'/0'/0'",
955 "fingerprint": "a" * 64,
956 })
957
958 call_count = [0]
959
960 def _fake_urlopen(req, *args, **kwargs):
961 call_count[0] += 1
962 url = req.full_url if hasattr(req, "full_url") else str(req)
963 if "challenge" in url:
964 body = json.dumps({
965 "challengeToken": "cafebabe" * 8, # 64-char hex nonce
966 "isNewKey": False,
967 "algorithm": "ed25519",
968 }).encode()
969 else:
970 body = json.dumps({
971 "token": "auth-token",
972 "handle": "stress-user",
973 "identityId": "id-stress",
974 "isNewIdentity": False,
975 "authMethod": "ed25519",
976 }).encode()
977
978 class _Resp:
979 status = 200
980 def read(self, n: int = -1) -> bytes:
981 return body
982 def __enter__(self): return self
983 def __exit__(self, *a): pass
984
985 return _Resp()
986
987 monkeypatch.setattr(urllib.request, "urlopen", _fake_urlopen)
988
989 handles_seen: list[str] = []
990 for i in range(5):
991 r = runner.invoke(
992 cli,
993 ["auth", "register", "--hub", "http://localhost:19960",
994 "--handle", "stress-user", "--json"],
995 )
996 assert r.exit_code == 0, f"register #{i} failed:\n{r.output}"
997 data = _parse_json(r)
998 handles_seen.append(data.get("handle", ""))
999
1000 assert all(h == "stress-user" for h in handles_seen), (
1001 f"handle inconsistent across 5 registrations: {handles_seen}"
1002 )
1003
1004
1005 # ── TDD: key_path removed from IdentityEntry and _dump_identity (P3) ─────────
1006
1007
1008 class TestKeyPathPurged:
1009 """P3: key_path must be gone from IdentityEntry TypedDict and _dump_identity.
1010
1011 Before fix: key_path: str is in IdentityEntry; _dump_identity serialises it.
1012 After fix: key_path is absent from the TypedDict; _dump_identity never emits it.
1013 """
1014
1015 def test_P3_1_key_path_not_in_identity_entry_typeddict(self) -> None:
1016 """key_path must not appear in IdentityEntry's annotations."""
1017 from muse.core.identity import IdentityEntry
1018 annotations = IdentityEntry.__annotations__
1019 assert "key_path" not in annotations, (
1020 "key_path still in IdentityEntry TypedDict — Phase 3 not complete"
1021 )
1022
1023 def test_P3_2_dump_identity_never_emits_key_path(self) -> None:
1024 """_dump_identity must not write key_path to TOML even when entry has it."""
1025 from muse.core.identity import _dump_identity
1026 entry = {
1027 "type": "human",
1028 "handle": "alice",
1029 "key_path": "/home/alice/.muse/keys/musehub_ai.pem", # must be stripped
1030 "algorithm": "ed25519",
1031 "fingerprint": "abc123",
1032 "hd_path": "m/1075233755'/0'/0'/0'/0'/0'",
1033 }
1034 toml_text = _dump_identity({"musehub.ai": entry})
1035 assert "key_path" not in toml_text, (
1036 f"_dump_identity still emits key_path:\n{toml_text}"
1037 )
File History 2 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 135 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 141 days ago