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