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