gabriel / muse public
test_migrate_domain_integers.py python
958 lines 37.8 KB
Raw
sha256:2a703f78341332ef0beb9856d2267de6aec89b3883c31519b6900b667d026e62 chore: delete muse/prose domain — hallucinated, never existed Sonnet 4.6 minor ⚠ breaking 3 days ago
1 """TDD tests for ``muse migrate domain-integers``.
2
3 Phase 4 of issue #3. Covers three layers:
4
5 1. Core library (muse.core.domain_migration)
6 - Legacy path detection
7 - Old → new path mapping for all 7 pre-Phase-1 domains
8 - Key re-derivation produces a different (correct) fingerprint
9 - Identity file scanning
10
11 2. Dry-run plan (no writes, no hub calls)
12
13 3. CLI smoke (muse migrate domain-integers --dry-run)
14
15 Background
16 ----------
17 Pre-Phase-1 Muse used sequential domain integers (0–6). Phase 1 switched to
18 hash-derived integers (e.g. DOMAIN_IDENTITY 0 → 1660078172). Users with keys
19 derived at the old paths must re-derive at the new paths and re-register with
20 the hub. This command automates that migration.
21
22 Old → new mapping::
23
24 0 muse/identity → 1660078172
25 1 muse/payments → 284229149
26 2 muse/code → 678195575
27 3 muse/music → 1755707987
28 4 muse/midi → 1444628350
29 6 muse/blockchain → 1556829714
30 """
31
32 from __future__ import annotations
33
34 import json
35 import pathlib
36 from collections.abc import Mapping
37 from unittest.mock import MagicMock, patch
38
39 import pytest
40
41 from muse.core.hdkeys import (
42 DOMAIN_IDENTITY,
43 DOMAIN_PAYMENTS,
44 DOMAIN_CODE,
45 DOMAIN_MUSIC,
46 DOMAIN_MIDI,
47 DOMAIN_BLOCKCHAIN,
48 ENTITY_HUMAN,
49 ENTITY_AGENT,
50 ROLE_SIGN,
51 ROLE_ATTEST,
52 muse_path,
53 )
54 from muse.core.paths import muse_dir
55
56 # ---------------------------------------------------------------------------
57 # Test mnemonic — BIP39 all-zeros vector; deterministic, never touches keychain
58 # ---------------------------------------------------------------------------
59 FAKE_MNEMONIC = (
60 "abandon abandon abandon abandon abandon abandon "
61 "abandon abandon abandon abandon abandon about"
62 )
63
64 # ---------------------------------------------------------------------------
65 # Helpers
66 # ---------------------------------------------------------------------------
67
68
69 def _old_path(domain: int, entity_type: int = 0, entity_id: int = 0, role: int = 0, index: int = 0) -> str:
70 """Build a pre-Phase-1 HD path with a sequential domain integer."""
71 from muse.core.slip010 import MUSE_PURPOSE
72 return f"m/{MUSE_PURPOSE}'/{domain}'/{entity_type}'/{entity_id}'/{role}'/{index}'"
73
74
75 # ============================================================================
76 # 1. Core: legacy path detection
77 # ============================================================================
78
79
80 class TestIsLegacyHdPath:
81 def test_domain_0_identity_is_legacy(self) -> None:
82 from muse.core.domain_migration import is_legacy_hd_path
83 assert is_legacy_hd_path(_old_path(0)) is True
84
85 def test_all_old_sequential_domains_are_legacy(self) -> None:
86 from muse.core.domain_migration import is_legacy_hd_path
87 for d in range(7):
88 assert is_legacy_hd_path(_old_path(d)) is True, f"domain {d} should be legacy"
89
90 def test_new_identity_domain_is_not_legacy(self) -> None:
91 from muse.core.domain_migration import is_legacy_hd_path
92 assert is_legacy_hd_path(muse_path(DOMAIN_IDENTITY)) is False
93
94 def test_all_new_hash_derived_domains_are_not_legacy(self) -> None:
95 from muse.core.domain_migration import is_legacy_hd_path
96 for domain in (DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, DOMAIN_MUSIC,
97 DOMAIN_MIDI, DOMAIN_BLOCKCHAIN):
98 assert is_legacy_hd_path(muse_path(domain)) is False, f"domain {domain} should not be legacy"
99
100 def test_empty_string_is_not_legacy(self) -> None:
101 from muse.core.domain_migration import is_legacy_hd_path
102 assert is_legacy_hd_path("") is False
103
104 def test_non_muse_purpose_is_not_legacy(self) -> None:
105 from muse.core.domain_migration import is_legacy_hd_path
106 assert is_legacy_hd_path("m/44'/0'/0'/0'/0'/0'") is False
107
108 def test_agent_old_path_is_legacy(self) -> None:
109 from muse.core.domain_migration import is_legacy_hd_path
110 # entity_type=1 (agent), old domain=0
111 assert is_legacy_hd_path(_old_path(0, entity_type=1, entity_id=3)) is True
112
113 def test_integer_7_is_not_legacy(self) -> None:
114 # 7 was never a valid old domain — only 0–6 are known legacy integers
115 from muse.core.domain_migration import is_legacy_hd_path
116 assert is_legacy_hd_path(_old_path(7)) is False
117
118
119 # ============================================================================
120 # 2. Core: old → new path mapping
121 # ============================================================================
122
123
124 class TestNewPathForLegacy:
125 @pytest.mark.parametrize("old_int,expected_new", [
126 (0, DOMAIN_IDENTITY),
127 (1, DOMAIN_PAYMENTS),
128 (2, DOMAIN_CODE),
129 (3, DOMAIN_MUSIC),
130 (4, DOMAIN_MIDI),
131 (6, DOMAIN_BLOCKCHAIN),
132 ])
133 def test_all_old_domains_map_to_correct_new_index(self, old_int: int, expected_new: int) -> None:
134 from muse.core.domain_migration import new_path_for_legacy
135 new = new_path_for_legacy(_old_path(old_int))
136 assert f"/{expected_new}'" in new
137
138 def test_preserves_entity_type(self) -> None:
139 from muse.core.domain_migration import new_path_for_legacy
140 old = _old_path(0, entity_type=1)
141 new = new_path_for_legacy(old)
142 assert "/1'/" in new # entity_type=1 preserved
143
144 def test_preserves_entity_id(self) -> None:
145 from muse.core.domain_migration import new_path_for_legacy
146 old = _old_path(0, entity_id=5)
147 new = new_path_for_legacy(old)
148 assert "/5'/" in new # entity_id=5 preserved
149
150 def test_preserves_role(self) -> None:
151 from muse.core.domain_migration import new_path_for_legacy
152 old = _old_path(0, role=ROLE_ATTEST)
153 new = new_path_for_legacy(old)
154 assert f"/{ROLE_ATTEST}'" in new
155
156 def test_preserves_index(self) -> None:
157 from muse.core.domain_migration import new_path_for_legacy
158 old = _old_path(0, index=2)
159 new = new_path_for_legacy(old)
160 assert "/2'" in new # index=2 preserved
161
162 def test_unknown_old_domain_raises(self) -> None:
163 from muse.core.domain_migration import new_path_for_legacy
164 with pytest.raises(ValueError, match="Unknown legacy domain integer"):
165 new_path_for_legacy(_old_path(7))
166
167 def test_already_new_path_raises(self) -> None:
168 from muse.core.domain_migration import new_path_for_legacy
169 with pytest.raises(ValueError):
170 new_path_for_legacy(muse_path(DOMAIN_IDENTITY))
171
172 def test_output_is_valid_muse_path_format(self) -> None:
173 from muse.core.domain_migration import new_path_for_legacy
174 from muse.core.slip010 import MUSE_PURPOSE
175 new = new_path_for_legacy(_old_path(0))
176 assert new.startswith(f"m/{MUSE_PURPOSE}'")
177 parts = new.lstrip("m/").split("/")
178 assert len(parts) == 6
179 assert all(p.endswith("'") for p in parts)
180
181
182 # ============================================================================
183 # 3. Core: key re-derivation
184 # ============================================================================
185
186
187 class TestDeriveNewFingerprint:
188 def test_old_and_new_fingerprints_differ(self) -> None:
189 """Key at new path is different from key at old path — migration has effect."""
190 from muse.core.domain_migration import derive_fingerprint_at_path
191 from muse.core.bip39 import mnemonic_to_seed
192 seed = mnemonic_to_seed(FAKE_MNEMONIC)
193 old = derive_fingerprint_at_path(seed, _old_path(0))
194 new = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
195 assert old != new
196
197 def test_new_fingerprint_is_deterministic(self) -> None:
198 """Same mnemonic + new path always produces the same fingerprint."""
199 from muse.core.domain_migration import derive_fingerprint_at_path
200 from muse.core.bip39 import mnemonic_to_seed
201 seed = mnemonic_to_seed(FAKE_MNEMONIC)
202 fp1 = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
203 fp2 = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
204 assert fp1 == fp2
205
206 def test_new_fingerprint_matches_direct_derive(self) -> None:
207 """derive_fingerprint_at_path agrees with derive_hd_public_info for identity path."""
208 from muse.core.domain_migration import derive_fingerprint_at_path
209 from muse.core.bip39 import mnemonic_to_seed
210 from muse.core.keypair import derive_hd_public_info
211 seed = mnemonic_to_seed(FAKE_MNEMONIC)
212 _, expected_fp = derive_hd_public_info(seed)
213 actual_fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
214 assert actual_fp == expected_fp
215
216 def test_fingerprint_is_sha256_prefixed(self) -> None:
217 from muse.core.domain_migration import derive_fingerprint_at_path
218 from muse.core.bip39 import mnemonic_to_seed
219 seed = mnemonic_to_seed(FAKE_MNEMONIC)
220 fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
221 assert fp.startswith("sha256:")
222 assert len(fp) == 71 # "sha256:" (7) + 64 hex chars
223
224
225 # ============================================================================
226 # 4. Core: scanning identity map for legacy entries
227 # ============================================================================
228
229
230 class TestScanForLegacy:
231 def test_finds_legacy_human_entry(self) -> None:
232 from muse.core.domain_migration import scan_for_legacy
233 identity_map = {
234 "localhost:1337": {
235 "type": "human",
236 "hd_path": _old_path(0),
237 "fingerprint": "a" * 64,
238 }
239 }
240 keys = scan_for_legacy(identity_map)
241 assert "localhost:1337" in keys
242
243 def test_ignores_already_migrated_entry(self) -> None:
244 from muse.core.domain_migration import scan_for_legacy
245 identity_map = {
246 "localhost:1337": {
247 "type": "human",
248 "hd_path": muse_path(DOMAIN_IDENTITY),
249 "fingerprint": "b" * 64,
250 }
251 }
252 assert scan_for_legacy(identity_map) == []
253
254 def test_ignores_entry_with_no_hd_path(self) -> None:
255 from muse.core.domain_migration import scan_for_legacy
256 identity_map = {
257 "localhost:1337": {
258 "type": "human",
259 "fingerprint": "c" * 64,
260 }
261 }
262 assert scan_for_legacy(identity_map) == []
263
264 def test_finds_multiple_legacy_hubs(self) -> None:
265 from muse.core.domain_migration import scan_for_legacy
266 identity_map = {
267 "localhost:1337": {"hd_path": _old_path(0), "fingerprint": "a" * 64},
268 "staging.musehub.ai": {"hd_path": _old_path(2), "fingerprint": "b" * 64},
269 }
270 keys = scan_for_legacy(identity_map)
271 assert set(keys) == {"localhost:1337", "staging.musehub.ai"}
272
273 def test_mixed_returns_only_legacy(self) -> None:
274 from muse.core.domain_migration import scan_for_legacy
275 identity_map = {
276 "localhost:1337": {"hd_path": _old_path(0), "fingerprint": "a" * 64},
277 "staging.musehub.ai": {"hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "b" * 64},
278 }
279 keys = scan_for_legacy(identity_map)
280 assert keys == ["localhost:1337"]
281
282 def test_empty_map_returns_empty(self) -> None:
283 from muse.core.domain_migration import scan_for_legacy
284 assert scan_for_legacy({}) == []
285
286
287 # ============================================================================
288 # 5. Core: MigrationPlan dataclass
289 # ============================================================================
290
291
292 class TestMigrationPlan:
293 def test_plan_has_expected_fields(self) -> None:
294 from muse.core.domain_migration import MigrationPlan
295 plan = MigrationPlan(
296 hub_key="localhost:1337",
297 old_hd_path=_old_path(0),
298 new_hd_path=muse_path(DOMAIN_IDENTITY),
299 old_domain_int=0,
300 new_domain_int=DOMAIN_IDENTITY,
301 domain_name="muse/identity",
302 )
303 assert plan.hub_key == "localhost:1337"
304 assert plan.old_domain_int == 0
305 assert plan.new_domain_int == DOMAIN_IDENTITY
306 assert plan.domain_name == "muse/identity"
307
308 def test_build_plans_from_identity_map(self) -> None:
309 from muse.core.domain_migration import build_plans
310 identity_map = {
311 "localhost:1337": {"hd_path": _old_path(0), "fingerprint": "a" * 64, "type": "human"},
312 "staging.musehub.ai": {"hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "b" * 64},
313 }
314 plans = build_plans(identity_map)
315 assert len(plans) == 1
316 assert plans[0].hub_key == "localhost:1337"
317 assert plans[0].domain_name == "muse/identity"
318 assert plans[0].old_domain_int == 0
319 assert plans[0].new_domain_int == DOMAIN_IDENTITY
320
321 def test_build_plans_all_seven_legacy_domains(self) -> None:
322 from muse.core.domain_migration import build_plans, OLD_DOMAIN_NAMES
323 identity_map = {
324 f"hub{i}": {"hd_path": _old_path(i), "fingerprint": "a" * 64}
325 for i in range(7)
326 }
327 plans = build_plans(identity_map)
328 assert len(plans) == 7
329 plan_by_key = {p.hub_key: p for p in plans}
330 for i, name in OLD_DOMAIN_NAMES.items():
331 assert plan_by_key[f"hub{i}"].domain_name == name
332
333
334 # ============================================================================
335 # 6. Dry-run: no writes, returns plan
336 # ============================================================================
337
338
339 class TestDryRun:
340 def test_dry_run_returns_plans_without_registering(self) -> None:
341 from muse.core.domain_migration import run_migration
342 from muse.core.bip39 import mnemonic_to_seed
343
344 identity_map = {
345 "localhost:1337": {
346 "type": "human",
347 "handle": "gabriel",
348 "hd_path": _old_path(0),
349 "fingerprint": "a" * 64,
350 "algorithm": "ed25519",
351 }
352 }
353 seed = mnemonic_to_seed(FAKE_MNEMONIC)
354
355 # hub_register_fn should NOT be called in dry_run mode
356 hub_register = MagicMock()
357 result = run_migration(
358 identity_map=identity_map,
359 seed=seed,
360 hub_register_fn=hub_register,
361 dry_run=True,
362 )
363
364 hub_register.assert_not_called()
365 assert len(result) == 1
366 assert result[0].old_hd_path == _old_path(0)
367 assert result[0].new_hd_path == muse_path(DOMAIN_IDENTITY)
368 assert result[0].hub_registered is False
369 assert result[0].new_fingerprint != result[0].old_fingerprint
370
371 def test_dry_run_does_not_mutate_identity_map(self) -> None:
372 from muse.core.domain_migration import run_migration
373 from muse.core.bip39 import mnemonic_to_seed
374
375 identity_map = {
376 "localhost:1337": {
377 "type": "human",
378 "hd_path": _old_path(0),
379 "fingerprint": "a" * 64,
380 }
381 }
382 original_fp = identity_map["localhost:1337"]["fingerprint"]
383 seed = mnemonic_to_seed(FAKE_MNEMONIC)
384
385 run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True)
386
387 assert identity_map["localhost:1337"]["fingerprint"] == original_fp
388 assert identity_map["localhost:1337"]["hd_path"] == _old_path(0)
389
390
391 # ============================================================================
392 # 7. Live run: hub called, identity map mutated
393 # ============================================================================
394
395
396 class TestLiveMigration:
397 def test_live_run_calls_hub_register(self) -> None:
398 from muse.core.domain_migration import run_migration
399 from muse.core.bip39 import mnemonic_to_seed
400
401 identity_map = {
402 "localhost:1337": {
403 "type": "human",
404 "handle": "gabriel",
405 "hd_path": _old_path(0),
406 "fingerprint": "a" * 64,
407 "algorithm": "ed25519",
408 }
409 }
410 seed = mnemonic_to_seed(FAKE_MNEMONIC)
411 hub_register = MagicMock(return_value=True)
412
413 run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
414
415 hub_register.assert_called_once()
416 call_kwargs = hub_register.call_args
417 assert "localhost:1337" in str(call_kwargs)
418
419 def test_live_run_updates_fingerprint_in_identity_map(self) -> None:
420 from muse.core.domain_migration import run_migration, derive_fingerprint_at_path
421 from muse.core.bip39 import mnemonic_to_seed
422
423 identity_map = {
424 "localhost:1337": {
425 "type": "human",
426 "handle": "gabriel",
427 "hd_path": _old_path(0),
428 "fingerprint": "a" * 64,
429 "algorithm": "ed25519",
430 }
431 }
432 seed = mnemonic_to_seed(FAKE_MNEMONIC)
433 hub_register = MagicMock(return_value=True)
434
435 run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
436
437 expected_fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
438 assert identity_map["localhost:1337"]["fingerprint"] == expected_fp
439 assert identity_map["localhost:1337"]["hd_path"] == muse_path(DOMAIN_IDENTITY)
440
441 def test_live_run_skips_already_migrated_entries(self) -> None:
442 from muse.core.domain_migration import run_migration
443 from muse.core.bip39 import mnemonic_to_seed
444
445 identity_map = {
446 "localhost:1337": {
447 "type": "human",
448 "hd_path": muse_path(DOMAIN_IDENTITY),
449 "fingerprint": "b" * 64,
450 }
451 }
452 seed = mnemonic_to_seed(FAKE_MNEMONIC)
453 hub_register = MagicMock()
454
455 result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
456
457 hub_register.assert_not_called()
458 assert result == []
459
460 def test_live_run_result_hub_registered_true_on_success(self) -> None:
461 from muse.core.domain_migration import run_migration
462 from muse.core.bip39 import mnemonic_to_seed
463
464 identity_map = {
465 "localhost:1337": {
466 "type": "human",
467 "handle": "gabriel",
468 "hd_path": _old_path(0),
469 "fingerprint": "a" * 64,
470 "algorithm": "ed25519",
471 }
472 }
473 seed = mnemonic_to_seed(FAKE_MNEMONIC)
474 result = run_migration(
475 identity_map=identity_map,
476 seed=seed,
477 hub_register_fn=MagicMock(return_value=True),
478 dry_run=False,
479 )
480 assert result[0].hub_registered is True
481
482 def test_live_run_result_hub_registered_false_on_failure(self) -> None:
483 from muse.core.domain_migration import run_migration
484 from muse.core.bip39 import mnemonic_to_seed
485
486 identity_map = {
487 "localhost:1337": {
488 "type": "human",
489 "handle": "gabriel",
490 "hd_path": _old_path(0),
491 "fingerprint": "a" * 64,
492 "algorithm": "ed25519",
493 }
494 }
495 seed = mnemonic_to_seed(FAKE_MNEMONIC)
496 hub_register = MagicMock(side_effect=RuntimeError("hub unreachable"))
497
498 result = run_migration(
499 identity_map=identity_map,
500 seed=seed,
501 hub_register_fn=hub_register,
502 dry_run=False,
503 )
504 assert result[0].hub_registered is False
505
506 def test_multi_hub_migrates_all_legacy(self) -> None:
507 from muse.core.domain_migration import run_migration
508 from muse.core.bip39 import mnemonic_to_seed
509
510 identity_map = {
511 "localhost:1337": {
512 "type": "human",
513 "handle": "gabriel",
514 "hd_path": _old_path(0),
515 "fingerprint": "a" * 64,
516 "algorithm": "ed25519",
517 },
518 "staging.musehub.ai": {
519 "type": "human",
520 "handle": "gabriel",
521 "hd_path": _old_path(0),
522 "fingerprint": "a" * 64,
523 "algorithm": "ed25519",
524 },
525 }
526 seed = mnemonic_to_seed(FAKE_MNEMONIC)
527 hub_register = MagicMock(return_value=True)
528
529 result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
530
531 assert len(result) == 2
532 assert hub_register.call_count == 2
533
534
535 # ============================================================================
536 # 8. CLI smoke: muse migrate domain-integers --dry-run
537 # ============================================================================
538
539
540 def _write_identity_toml(path: pathlib.Path, data: Mapping[str, Mapping[str, object]]) -> None:
541 """Write a minimal identity.toml without tomli_w dependency."""
542 lines = []
543 for section, fields in data.items():
544 lines.append(f'["{section}"]')
545 for k, v in fields.items():
546 lines.append(f'{k} = "{v}"')
547 lines.append("")
548 path.write_text("\n".join(lines), encoding="utf-8")
549 path.chmod(0o600)
550
551
552 class TestCliDryRun:
553 def test_cli_dry_run_exits_0_with_json(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
554 """CLI smoke: dry-run against a fake identity.toml with one legacy entry."""
555 dot_muse = muse_dir(tmp_path)
556 dot_muse.mkdir()
557 identity_file = dot_muse / "identity.toml"
558 _write_identity_toml(identity_file, {
559 "localhost:1337": {
560 "type": "human",
561 "handle": "gabriel",
562 "algorithm": "ed25519",
563 "fingerprint": "a" * 64,
564 "hd_path": _old_path(0),
565 }
566 })
567
568 import muse.core.identity as id_module
569 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
570 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
571
572 import muse.core.keychain as kc_module
573 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
574
575 from tests.cli_test_helper import CliRunner
576 runner = CliRunner()
577 result = runner.invoke(None, ["migrate", "domain-integers", "--dry-run", "--json"])
578
579 assert result.exit_code == 0, result.output
580 data = json.loads(result.output)
581 assert data["dry_run"] is True
582 assert data["entries_found"] == 1
583 assert data["entries_migrated"] == 0
584
585 def test_cli_no_legacy_entries_exits_0(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
586 """CLI returns 0 and reports 0 entries when identity is already migrated."""
587 dot_muse = muse_dir(tmp_path)
588 dot_muse.mkdir()
589 identity_file = dot_muse / "identity.toml"
590 _write_identity_toml(identity_file, {
591 "localhost:1337": {
592 "type": "human",
593 "handle": "gabriel",
594 "algorithm": "ed25519",
595 "fingerprint": "b" * 64,
596 "hd_path": muse_path(DOMAIN_IDENTITY),
597 }
598 })
599
600 import muse.core.identity as id_module
601 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
602 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
603
604 import muse.core.keychain as kc_module
605 monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC)
606
607 from tests.cli_test_helper import CliRunner
608 runner = CliRunner()
609 result = runner.invoke(None, ["migrate", "domain-integers", "--dry-run", "--json"])
610
611 assert result.exit_code == 0, result.output
612 data = json.loads(result.output)
613 assert data["entries_found"] == 0
614 assert data["entries_migrated"] == 0
615
616
617 # ============================================================================
618 # 9. Integration: real TOML file on disk
619 # ============================================================================
620
621
622 class TestIntegration:
623 def test_migration_against_real_toml_file(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None:
624 """run_migration correctly processes an identity_map loaded from a real TOML file."""
625 from muse.core.bip39 import mnemonic_to_seed
626 from muse.core.domain_migration import run_migration, derive_fingerprint_at_path
627
628 dot_muse = muse_dir(tmp_path)
629 dot_muse.mkdir()
630 identity_file = dot_muse / "identity.toml"
631 _write_identity_toml(identity_file, {
632 "localhost:1337": {
633 "type": "human",
634 "handle": "gabriel",
635 "algorithm": "ed25519",
636 "fingerprint": "a" * 64,
637 "hd_path": _old_path(0),
638 },
639 "staging.musehub.ai": {
640 "type": "human",
641 "handle": "gabriel",
642 "algorithm": "ed25519",
643 "fingerprint": "b" * 64,
644 "hd_path": muse_path(DOMAIN_IDENTITY),
645 },
646 })
647
648 import muse.core.identity as id_module
649 monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse)
650 monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file)
651
652 from muse.core.identity import list_all_identities
653 identity_map = dict(list_all_identities())
654 seed = mnemonic_to_seed(FAKE_MNEMONIC)
655
656 results = run_migration(
657 identity_map=identity_map,
658 seed=seed,
659 hub_register_fn=MagicMock(return_value=True),
660 dry_run=False,
661 )
662
663 assert len(results) == 1
664 assert results[0].hub_key == "localhost:1337"
665 assert results[0].hub_registered is True
666 expected_fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY))
667 assert identity_map["localhost:1337"]["fingerprint"] == expected_fp
668
669
670 # ============================================================================
671 # 10. Stress: large numbers of hubs
672 # ============================================================================
673
674
675 class TestStress:
676 def test_dry_run_fifty_hubs_no_writes(self) -> None:
677 """Dry-run with 50 hubs returns all results without calling hub_register_fn."""
678 from muse.core.domain_migration import run_migration
679 from muse.core.bip39 import mnemonic_to_seed
680
681 NUM_HUBS = 50
682 identity_map = {
683 f"hub{i}.example.com:1337": {
684 "type": "human",
685 "hd_path": _old_path(i % 7),
686 "fingerprint": "a" * 64,
687 }
688 for i in range(NUM_HUBS)
689 }
690 original_paths = {k: v["hd_path"] for k, v in identity_map.items()}
691 seed = mnemonic_to_seed(FAKE_MNEMONIC)
692 hub_register = MagicMock()
693
694 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=True)
695
696 hub_register.assert_not_called()
697 assert len(results) == NUM_HUBS
698 for k, v in identity_map.items():
699 assert v["hd_path"] == original_paths[k]
700
701 def test_live_run_fifty_hubs_all_registered(self) -> None:
702 """Live run with 50 hubs successfully migrates and registers each entry."""
703 from muse.core.domain_migration import run_migration
704 from muse.core.bip39 import mnemonic_to_seed
705
706 NUM_HUBS = 50
707 identity_map = {
708 f"hub{i}.example.com:1337": {
709 "type": "human",
710 "handle": "gabriel",
711 "hd_path": _old_path(i % 7),
712 "fingerprint": "a" * 64,
713 "algorithm": "ed25519",
714 }
715 for i in range(NUM_HUBS)
716 }
717 seed = mnemonic_to_seed(FAKE_MNEMONIC)
718 hub_register = MagicMock(return_value=True)
719
720 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False)
721
722 assert len(results) == NUM_HUBS
723 assert hub_register.call_count == NUM_HUBS
724 assert all(r.hub_registered for r in results)
725
726
727 # ============================================================================
728 # 11. Data integrity: idempotency and partial failures
729 # ============================================================================
730
731
732 class TestDataIntegrity:
733 def test_already_migrated_map_returns_empty(self) -> None:
734 """Running migration on an already-migrated identity_map returns no results."""
735 from muse.core.domain_migration import run_migration
736 from muse.core.bip39 import mnemonic_to_seed
737
738 identity_map = {
739 "localhost:1337": {
740 "type": "human",
741 "hd_path": muse_path(DOMAIN_IDENTITY),
742 "fingerprint": "z" * 64,
743 }
744 }
745 seed = mnemonic_to_seed(FAKE_MNEMONIC)
746 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=False)
747 assert results == []
748 assert identity_map["localhost:1337"]["fingerprint"] == "z" * 64
749
750 def test_partial_failure_updates_successful_entries(self) -> None:
751 """If hub registration fails for one entry, successful entries are still updated."""
752 from muse.core.domain_migration import run_migration
753 from muse.core.bip39 import mnemonic_to_seed
754
755 identity_map = {
756 "ok.musehub.ai": {
757 "type": "human",
758 "handle": "gabriel",
759 "hd_path": _old_path(0),
760 "fingerprint": "a" * 64,
761 "algorithm": "ed25519",
762 },
763 "bad.musehub.ai": {
764 "type": "human",
765 "handle": "gabriel",
766 "hd_path": _old_path(1),
767 "fingerprint": "b" * 64,
768 "algorithm": "ed25519",
769 },
770 }
771 seed = mnemonic_to_seed(FAKE_MNEMONIC)
772
773 def _flaky_register(hub_key: str, new_fingerprint: str, new_hd_path: str, entry: Mapping[str, object]) -> bool:
774 if "bad" in hub_key:
775 raise RuntimeError("network error")
776 return True
777
778 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=_flaky_register, dry_run=False)
779
780 assert len(results) == 2
781 ok_result = next(r for r in results if r.hub_key == "ok.musehub.ai")
782 bad_result = next(r for r in results if r.hub_key == "bad.musehub.ai")
783 assert ok_result.hub_registered is True
784 assert bad_result.hub_registered is False
785 # Both entries get their hd_path updated even when hub registration fails
786 assert identity_map["ok.musehub.ai"]["hd_path"] == muse_path(DOMAIN_IDENTITY)
787 assert identity_map["bad.musehub.ai"]["hd_path"] == muse_path(DOMAIN_PAYMENTS)
788
789 def test_fingerprint_format_stored_in_identity_map(self) -> None:
790 """Fingerprint written to identity_map after live migration is sha256: prefixed."""
791 from muse.core.domain_migration import run_migration
792 from muse.core.bip39 import mnemonic_to_seed
793
794 identity_map = {
795 "localhost:1337": {
796 "type": "human",
797 "handle": "gabriel",
798 "hd_path": _old_path(0),
799 "fingerprint": "a" * 64,
800 "algorithm": "ed25519",
801 }
802 }
803 seed = mnemonic_to_seed(FAKE_MNEMONIC)
804 run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False)
805 fp = identity_map["localhost:1337"]["fingerprint"]
806 assert fp.startswith("sha256:")
807 assert len(fp) == 71
808
809 def test_hd_path_updated_even_when_registration_fails(self) -> None:
810 """identity_map hd_path is corrected to the new value even if hub_register_fn raises."""
811 from muse.core.domain_migration import run_migration
812 from muse.core.bip39 import mnemonic_to_seed
813
814 identity_map = {
815 "localhost:1337": {
816 "type": "human",
817 "handle": "gabriel",
818 "hd_path": _old_path(0),
819 "fingerprint": "a" * 64,
820 "algorithm": "ed25519",
821 }
822 }
823 seed = mnemonic_to_seed(FAKE_MNEMONIC)
824 run_migration(
825 identity_map=identity_map,
826 seed=seed,
827 hub_register_fn=MagicMock(side_effect=RuntimeError("hub unreachable")),
828 dry_run=False,
829 )
830 assert identity_map["localhost:1337"]["hd_path"] == muse_path(DOMAIN_IDENTITY)
831
832 def test_dry_run_twice_same_results(self) -> None:
833 """Running dry-run twice on the same identity_map produces identical results."""
834 from muse.core.domain_migration import run_migration
835 from muse.core.bip39 import mnemonic_to_seed
836
837 identity_map = {
838 "localhost:1337": {
839 "type": "human",
840 "hd_path": _old_path(0),
841 "fingerprint": "a" * 64,
842 }
843 }
844 seed = mnemonic_to_seed(FAKE_MNEMONIC)
845 r1 = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True)
846 r2 = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True)
847 assert r1[0].new_fingerprint == r2[0].new_fingerprint
848 assert r1[0].new_hd_path == r2[0].new_hd_path
849
850
851 # ============================================================================
852 # 12. Security: adversarial inputs and boundary conditions
853 # ============================================================================
854
855
856 class TestSecurity:
857 def test_non_muse_purpose_with_legacy_domain_not_flagged(self) -> None:
858 """Path with domain=0 but wrong purpose is not treated as legacy."""
859 from muse.core.domain_migration import is_legacy_hd_path
860 # BIP-44 path with domain integer 0 — should never be flagged as legacy
861 assert is_legacy_hd_path("m/44'/0'/0'/0'/0'/0'") is False
862
863 def test_domain_just_above_range_not_legacy(self) -> None:
864 """Domain integer 7 is outside the 0-6 legacy range and must not be flagged."""
865 from muse.core.domain_migration import is_legacy_hd_path
866 assert is_legacy_hd_path(_old_path(7)) is False
867
868 def test_large_domain_integer_not_legacy(self) -> None:
869 """A large hash-derived domain integer is never mistaken for a legacy path."""
870 from muse.core.domain_migration import is_legacy_hd_path
871 assert is_legacy_hd_path(muse_path(DOMAIN_IDENTITY)) is False
872 assert is_legacy_hd_path(muse_path(DOMAIN_BLOCKCHAIN)) is False
873
874 def test_malformed_path_missing_hardened_marker_not_legacy(self) -> None:
875 """A path without hardened markers (no apostrophes) is not treated as legacy."""
876 from muse.core.domain_migration import is_legacy_hd_path
877 from muse.core.slip010 import MUSE_PURPOSE
878 # Remove hardened markers — soft derivation path
879 assert is_legacy_hd_path(f"m/{MUSE_PURPOSE}/0/0/0/0/0") is False
880
881 def test_path_with_too_few_segments_not_legacy(self) -> None:
882 """A truncated path (fewer than 6 segments) never matches as legacy."""
883 from muse.core.domain_migration import is_legacy_hd_path
884 from muse.core.slip010 import MUSE_PURPOSE
885 assert is_legacy_hd_path(f"m/{MUSE_PURPOSE}'/0'/0'") is False
886
887 def test_path_with_extra_segments_not_legacy(self) -> None:
888 """A path with more than 6 segments is never flagged as legacy."""
889 from muse.core.domain_migration import is_legacy_hd_path
890 from muse.core.slip010 import MUSE_PURPOSE
891 assert is_legacy_hd_path(f"m/{MUSE_PURPOSE}'/0'/0'/0'/0'/0'/0'") is False
892
893 def test_empty_string_not_legacy(self) -> None:
894 """Empty string input does not raise and returns False."""
895 from muse.core.domain_migration import is_legacy_hd_path
896 assert is_legacy_hd_path("") is False
897
898 def test_whitespace_only_not_legacy(self) -> None:
899 """Whitespace-only string does not raise and returns False."""
900 from muse.core.domain_migration import is_legacy_hd_path
901 assert is_legacy_hd_path(" ") is False
902
903 def test_path_traversal_attempt_not_legacy(self) -> None:
904 """A path-traversal-style string is not treated as a legacy path."""
905 from muse.core.domain_migration import is_legacy_hd_path
906 assert is_legacy_hd_path("m/1075233755'/../../../etc/passwd") is False
907
908 def test_derive_fingerprint_invalid_path_raises(self) -> None:
909 """derive_fingerprint_at_path raises ValueError for a non-parseable path."""
910 from muse.core.domain_migration import derive_fingerprint_at_path
911 from muse.core.bip39 import mnemonic_to_seed
912 seed = mnemonic_to_seed(FAKE_MNEMONIC)
913 with pytest.raises(ValueError, match="Cannot parse"):
914 derive_fingerprint_at_path(seed, "not-a-valid-path")
915
916
917 # ============================================================================
918 # 13. Performance: timing assertions
919 # ============================================================================
920
921
922 class TestPerformance:
923 def test_run_migration_twenty_entries_under_ten_seconds(self) -> None:
924 """run_migration with 20 entries (dry-run) completes within 10 seconds."""
925 import time
926 from muse.core.domain_migration import run_migration
927 from muse.core.bip39 import mnemonic_to_seed
928
929 identity_map = {
930 f"hub{i}.example.com:1337": {
931 "type": "human",
932 "hd_path": _old_path(i % 7),
933 "fingerprint": "a" * 64,
934 }
935 for i in range(20)
936 }
937 seed = mnemonic_to_seed(FAKE_MNEMONIC)
938
939 t0 = time.monotonic()
940 results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True)
941 elapsed = time.monotonic() - t0
942
943 assert len(results) == 20
944 assert elapsed < 10.0, f"run_migration took {elapsed:.2f}s — expected < 10s"
945
946 def test_is_legacy_hd_path_fast_for_thousand_calls(self) -> None:
947 """is_legacy_hd_path classifies 1000 paths in under 1 second."""
948 import time
949 from muse.core.domain_migration import is_legacy_hd_path
950
951 paths = [_old_path(i % 7) for i in range(500)] + [muse_path(DOMAIN_IDENTITY)] * 500
952
953 t0 = time.monotonic()
954 results = [is_legacy_hd_path(p) for p in paths]
955 elapsed = time.monotonic() - t0
956
957 assert sum(results) == 500 # exactly the 500 legacy paths
958 assert elapsed < 1.0, f"is_legacy_hd_path took {elapsed:.2f}s for 1000 calls"
File History 1 commit
sha256:2a703f78341332ef0beb9856d2267de6aec89b3883c31519b6900b667d026e62 chore: delete muse/prose domain — hallucinated, never existed Sonnet 4.6 minor 3 days ago