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