"""TDD tests for ``muse migrate domain-integers``. Phase 4 of issue #3. Covers three layers: 1. Core library (muse.core.domain_migration) - Legacy path detection - Old → new path mapping for all 7 pre-Phase-1 domains - Key re-derivation produces a different (correct) fingerprint - Identity file scanning 2. Dry-run plan (no writes, no hub calls) 3. CLI smoke (muse migrate domain-integers --dry-run) Background ---------- Pre-Phase-1 Muse used sequential domain integers (0–6). Phase 1 switched to hash-derived integers (e.g. DOMAIN_IDENTITY 0 → 1660078172). Users with keys derived at the old paths must re-derive at the new paths and re-register with the hub. This command automates that migration. Old → new mapping:: 0 muse/identity → 1660078172 1 muse/payments → 284229149 2 muse/code → 678195575 3 muse/music → 1755707987 4 muse/midi → 1444628350 6 muse/blockchain → 1556829714 """ from __future__ import annotations import json import pathlib from collections.abc import Mapping from unittest.mock import MagicMock, patch import pytest from muse.core.hdkeys import ( DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, DOMAIN_MUSIC, DOMAIN_MIDI, DOMAIN_BLOCKCHAIN, ENTITY_HUMAN, ENTITY_AGENT, ROLE_SIGN, ROLE_ATTEST, muse_path, ) from muse.core.paths import muse_dir # --------------------------------------------------------------------------- # Test mnemonic — BIP39 all-zeros vector; deterministic, never touches keychain # --------------------------------------------------------------------------- FAKE_MNEMONIC = ( "abandon abandon abandon abandon abandon abandon " "abandon abandon abandon abandon abandon about" ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _old_path(domain: int, entity_type: int = 0, entity_id: int = 0, role: int = 0, index: int = 0) -> str: """Build a pre-Phase-1 HD path with a sequential domain integer.""" from muse.core.slip010 import MUSE_PURPOSE return f"m/{MUSE_PURPOSE}'/{domain}'/{entity_type}'/{entity_id}'/{role}'/{index}'" # ============================================================================ # 1. Core: legacy path detection # ============================================================================ class TestIsLegacyHdPath: def test_domain_0_identity_is_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path(_old_path(0)) is True def test_all_old_sequential_domains_are_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path for d in range(7): assert is_legacy_hd_path(_old_path(d)) is True, f"domain {d} should be legacy" def test_new_identity_domain_is_not_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path(muse_path(DOMAIN_IDENTITY)) is False def test_all_new_hash_derived_domains_are_not_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path for domain in (DOMAIN_IDENTITY, DOMAIN_PAYMENTS, DOMAIN_CODE, DOMAIN_MUSIC, DOMAIN_MIDI, DOMAIN_BLOCKCHAIN): assert is_legacy_hd_path(muse_path(domain)) is False, f"domain {domain} should not be legacy" def test_empty_string_is_not_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path("") is False def test_non_muse_purpose_is_not_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path("m/44'/0'/0'/0'/0'/0'") is False def test_agent_old_path_is_legacy(self) -> None: from muse.core.domain_migration import is_legacy_hd_path # entity_type=1 (agent), old domain=0 assert is_legacy_hd_path(_old_path(0, entity_type=1, entity_id=3)) is True def test_integer_7_is_not_legacy(self) -> None: # 7 was never a valid old domain — only 0–6 are known legacy integers from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path(_old_path(7)) is False # ============================================================================ # 2. Core: old → new path mapping # ============================================================================ class TestNewPathForLegacy: @pytest.mark.parametrize("old_int,expected_new", [ (0, DOMAIN_IDENTITY), (1, DOMAIN_PAYMENTS), (2, DOMAIN_CODE), (3, DOMAIN_MUSIC), (4, DOMAIN_MIDI), (6, DOMAIN_BLOCKCHAIN), ]) def test_all_old_domains_map_to_correct_new_index(self, old_int: int, expected_new: int) -> None: from muse.core.domain_migration import new_path_for_legacy new = new_path_for_legacy(_old_path(old_int)) assert f"/{expected_new}'" in new def test_preserves_entity_type(self) -> None: from muse.core.domain_migration import new_path_for_legacy old = _old_path(0, entity_type=1) new = new_path_for_legacy(old) assert "/1'/" in new # entity_type=1 preserved def test_preserves_entity_id(self) -> None: from muse.core.domain_migration import new_path_for_legacy old = _old_path(0, entity_id=5) new = new_path_for_legacy(old) assert "/5'/" in new # entity_id=5 preserved def test_preserves_role(self) -> None: from muse.core.domain_migration import new_path_for_legacy old = _old_path(0, role=ROLE_ATTEST) new = new_path_for_legacy(old) assert f"/{ROLE_ATTEST}'" in new def test_preserves_index(self) -> None: from muse.core.domain_migration import new_path_for_legacy old = _old_path(0, index=2) new = new_path_for_legacy(old) assert "/2'" in new # index=2 preserved def test_unknown_old_domain_raises(self) -> None: from muse.core.domain_migration import new_path_for_legacy with pytest.raises(ValueError, match="Unknown legacy domain integer"): new_path_for_legacy(_old_path(7)) def test_already_new_path_raises(self) -> None: from muse.core.domain_migration import new_path_for_legacy with pytest.raises(ValueError): new_path_for_legacy(muse_path(DOMAIN_IDENTITY)) def test_output_is_valid_muse_path_format(self) -> None: from muse.core.domain_migration import new_path_for_legacy from muse.core.slip010 import MUSE_PURPOSE new = new_path_for_legacy(_old_path(0)) assert new.startswith(f"m/{MUSE_PURPOSE}'") parts = new.lstrip("m/").split("/") assert len(parts) == 6 assert all(p.endswith("'") for p in parts) # ============================================================================ # 3. Core: key re-derivation # ============================================================================ class TestDeriveNewFingerprint: def test_old_and_new_fingerprints_differ(self) -> None: """Key at new path is different from key at old path — migration has effect.""" from muse.core.domain_migration import derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) old = derive_fingerprint_at_path(seed, _old_path(0)) new = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) assert old != new def test_new_fingerprint_is_deterministic(self) -> None: """Same mnemonic + new path always produces the same fingerprint.""" from muse.core.domain_migration import derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) fp1 = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) fp2 = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) assert fp1 == fp2 def test_new_fingerprint_matches_direct_derive(self) -> None: """derive_fingerprint_at_path agrees with derive_hd_public_info for identity path.""" from muse.core.domain_migration import derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed from muse.core.keypair import derive_hd_public_info seed = mnemonic_to_seed(FAKE_MNEMONIC) _, expected_fp = derive_hd_public_info(seed) actual_fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) assert actual_fp == expected_fp def test_fingerprint_is_sha256_prefixed(self) -> None: from muse.core.domain_migration import derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) assert fp.startswith("sha256:") assert len(fp) == 71 # "sha256:" (7) + 64 hex chars # ============================================================================ # 4. Core: scanning identity map for legacy entries # ============================================================================ class TestScanForLegacy: def test_finds_legacy_human_entry(self) -> None: from muse.core.domain_migration import scan_for_legacy identity_map = { "localhost:1337": { "type": "human", "hd_path": _old_path(0), "fingerprint": "a" * 64, } } keys = scan_for_legacy(identity_map) assert "localhost:1337" in keys def test_ignores_already_migrated_entry(self) -> None: from muse.core.domain_migration import scan_for_legacy identity_map = { "localhost:1337": { "type": "human", "hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "b" * 64, } } assert scan_for_legacy(identity_map) == [] def test_ignores_entry_with_no_hd_path(self) -> None: from muse.core.domain_migration import scan_for_legacy identity_map = { "localhost:1337": { "type": "human", "fingerprint": "c" * 64, } } assert scan_for_legacy(identity_map) == [] def test_finds_multiple_legacy_hubs(self) -> None: from muse.core.domain_migration import scan_for_legacy identity_map = { "localhost:1337": {"hd_path": _old_path(0), "fingerprint": "a" * 64}, "staging.musehub.ai": {"hd_path": _old_path(2), "fingerprint": "b" * 64}, } keys = scan_for_legacy(identity_map) assert set(keys) == {"localhost:1337", "staging.musehub.ai"} def test_mixed_returns_only_legacy(self) -> None: from muse.core.domain_migration import scan_for_legacy identity_map = { "localhost:1337": {"hd_path": _old_path(0), "fingerprint": "a" * 64}, "staging.musehub.ai": {"hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "b" * 64}, } keys = scan_for_legacy(identity_map) assert keys == ["localhost:1337"] def test_empty_map_returns_empty(self) -> None: from muse.core.domain_migration import scan_for_legacy assert scan_for_legacy({}) == [] # ============================================================================ # 5. Core: MigrationPlan dataclass # ============================================================================ class TestMigrationPlan: def test_plan_has_expected_fields(self) -> None: from muse.core.domain_migration import MigrationPlan plan = MigrationPlan( hub_key="localhost:1337", old_hd_path=_old_path(0), new_hd_path=muse_path(DOMAIN_IDENTITY), old_domain_int=0, new_domain_int=DOMAIN_IDENTITY, domain_name="muse/identity", ) assert plan.hub_key == "localhost:1337" assert plan.old_domain_int == 0 assert plan.new_domain_int == DOMAIN_IDENTITY assert plan.domain_name == "muse/identity" def test_build_plans_from_identity_map(self) -> None: from muse.core.domain_migration import build_plans identity_map = { "localhost:1337": {"hd_path": _old_path(0), "fingerprint": "a" * 64, "type": "human"}, "staging.musehub.ai": {"hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "b" * 64}, } plans = build_plans(identity_map) assert len(plans) == 1 assert plans[0].hub_key == "localhost:1337" assert plans[0].domain_name == "muse/identity" assert plans[0].old_domain_int == 0 assert plans[0].new_domain_int == DOMAIN_IDENTITY def test_build_plans_all_seven_legacy_domains(self) -> None: from muse.core.domain_migration import build_plans, OLD_DOMAIN_NAMES identity_map = { f"hub{i}": {"hd_path": _old_path(i), "fingerprint": "a" * 64} for i in range(7) } plans = build_plans(identity_map) assert len(plans) == 7 plan_by_key = {p.hub_key: p for p in plans} for i, name in OLD_DOMAIN_NAMES.items(): assert plan_by_key[f"hub{i}"].domain_name == name # ============================================================================ # 6. Dry-run: no writes, returns plan # ============================================================================ class TestDryRun: def test_dry_run_returns_plans_without_registering(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) # hub_register_fn should NOT be called in dry_run mode hub_register = MagicMock() result = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=True, ) hub_register.assert_not_called() assert len(result) == 1 assert result[0].old_hd_path == _old_path(0) assert result[0].new_hd_path == muse_path(DOMAIN_IDENTITY) assert result[0].hub_registered is False assert result[0].new_fingerprint != result[0].old_fingerprint def test_dry_run_does_not_mutate_identity_map(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "hd_path": _old_path(0), "fingerprint": "a" * 64, } } original_fp = identity_map["localhost:1337"]["fingerprint"] seed = mnemonic_to_seed(FAKE_MNEMONIC) run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True) assert identity_map["localhost:1337"]["fingerprint"] == original_fp assert identity_map["localhost:1337"]["hd_path"] == _old_path(0) # ============================================================================ # 7. Live run: hub called, identity map mutated # ============================================================================ class TestLiveMigration: def test_live_run_calls_hub_register(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock(return_value=True) run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) hub_register.assert_called_once() call_kwargs = hub_register.call_args assert "localhost:1337" in str(call_kwargs) def test_live_run_updates_fingerprint_in_identity_map(self) -> None: from muse.core.domain_migration import run_migration, derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock(return_value=True) run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) expected_fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) assert identity_map["localhost:1337"]["fingerprint"] == expected_fp assert identity_map["localhost:1337"]["hd_path"] == muse_path(DOMAIN_IDENTITY) def test_live_run_skips_already_migrated_entries(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "b" * 64, } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock() result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) hub_register.assert_not_called() assert result == [] def test_live_run_result_hub_registered_true_on_success(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) result = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False, ) assert result[0].hub_registered is True def test_live_run_result_hub_registered_false_on_failure(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock(side_effect=RuntimeError("hub unreachable")) result = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False, ) assert result[0].hub_registered is False def test_multi_hub_migrates_all_legacy(self) -> None: from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", }, "staging.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", }, } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock(return_value=True) result = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) assert len(result) == 2 assert hub_register.call_count == 2 # ============================================================================ # 8. CLI smoke: muse migrate domain-integers --dry-run # ============================================================================ def _write_identity_toml(path: pathlib.Path, data: Mapping[str, Mapping[str, object]]) -> None: """Write a minimal identity.toml without tomli_w dependency.""" lines = [] for section, fields in data.items(): lines.append(f'["{section}"]') for k, v in fields.items(): lines.append(f'{k} = "{v}"') lines.append("") path.write_text("\n".join(lines), encoding="utf-8") path.chmod(0o600) class TestCliDryRun: def test_cli_dry_run_exits_0_with_json(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: """CLI smoke: dry-run against a fake identity.toml with one legacy entry.""" dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "localhost:1337": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _old_path(0), } }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke(None, ["migrate", "domain-integers", "--dry-run", "--json"]) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["dry_run"] is True assert data["entries_found"] == 1 assert data["entries_migrated"] == 0 def test_cli_no_legacy_entries_exits_0(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: """CLI returns 0 and reports 0 entries when identity is already migrated.""" dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "localhost:1337": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "b" * 64, "hd_path": muse_path(DOMAIN_IDENTITY), } }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) import muse.core.keychain as kc_module monkeypatch.setattr(kc_module, "load", lambda: FAKE_MNEMONIC) from tests.cli_test_helper import CliRunner runner = CliRunner() result = runner.invoke(None, ["migrate", "domain-integers", "--dry-run", "--json"]) assert result.exit_code == 0, result.output data = json.loads(result.output) assert data["entries_found"] == 0 assert data["entries_migrated"] == 0 # ============================================================================ # 9. Integration: real TOML file on disk # ============================================================================ class TestIntegration: def test_migration_against_real_toml_file(self, tmp_path: pathlib.Path, monkeypatch: pytest.MonkeyPatch) -> None: """run_migration correctly processes an identity_map loaded from a real TOML file.""" from muse.core.bip39 import mnemonic_to_seed from muse.core.domain_migration import run_migration, derive_fingerprint_at_path dot_muse = muse_dir(tmp_path) dot_muse.mkdir() identity_file = dot_muse / "identity.toml" _write_identity_toml(identity_file, { "localhost:1337": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "a" * 64, "hd_path": _old_path(0), }, "staging.musehub.ai": { "type": "human", "handle": "gabriel", "algorithm": "ed25519", "fingerprint": "b" * 64, "hd_path": muse_path(DOMAIN_IDENTITY), }, }) import muse.core.identity as id_module monkeypatch.setattr(id_module, "_IDENTITY_DIR", dot_muse) monkeypatch.setattr(id_module, "_IDENTITY_FILE", identity_file) from muse.core.identity import list_all_identities identity_map = dict(list_all_identities()) seed = mnemonic_to_seed(FAKE_MNEMONIC) results = run_migration( identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False, ) assert len(results) == 1 assert results[0].hub_key == "localhost:1337" assert results[0].hub_registered is True expected_fp = derive_fingerprint_at_path(seed, muse_path(DOMAIN_IDENTITY)) assert identity_map["localhost:1337"]["fingerprint"] == expected_fp # ============================================================================ # 10. Stress: large numbers of hubs # ============================================================================ class TestStress: def test_dry_run_fifty_hubs_no_writes(self) -> None: """Dry-run with 50 hubs returns all results without calling hub_register_fn.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed NUM_HUBS = 50 identity_map = { f"hub{i}.example.com:1337": { "type": "human", "hd_path": _old_path(i % 7), "fingerprint": "a" * 64, } for i in range(NUM_HUBS) } original_paths = {k: v["hd_path"] for k, v in identity_map.items()} seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock() results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=True) hub_register.assert_not_called() assert len(results) == NUM_HUBS for k, v in identity_map.items(): assert v["hd_path"] == original_paths[k] def test_live_run_fifty_hubs_all_registered(self) -> None: """Live run with 50 hubs successfully migrates and registers each entry.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed NUM_HUBS = 50 identity_map = { f"hub{i}.example.com:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(i % 7), "fingerprint": "a" * 64, "algorithm": "ed25519", } for i in range(NUM_HUBS) } seed = mnemonic_to_seed(FAKE_MNEMONIC) hub_register = MagicMock(return_value=True) results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=hub_register, dry_run=False) assert len(results) == NUM_HUBS assert hub_register.call_count == NUM_HUBS assert all(r.hub_registered for r in results) # ============================================================================ # 11. Data integrity: idempotency and partial failures # ============================================================================ class TestDataIntegrity: def test_already_migrated_map_returns_empty(self) -> None: """Running migration on an already-migrated identity_map returns no results.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "hd_path": muse_path(DOMAIN_IDENTITY), "fingerprint": "z" * 64, } } seed = mnemonic_to_seed(FAKE_MNEMONIC) results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=False) assert results == [] assert identity_map["localhost:1337"]["fingerprint"] == "z" * 64 def test_partial_failure_updates_successful_entries(self) -> None: """If hub registration fails for one entry, successful entries are still updated.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "ok.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", }, "bad.musehub.ai": { "type": "human", "handle": "gabriel", "hd_path": _old_path(1), "fingerprint": "b" * 64, "algorithm": "ed25519", }, } seed = mnemonic_to_seed(FAKE_MNEMONIC) def _flaky_register(hub_key: str, new_fingerprint: str, new_hd_path: str, entry: Mapping[str, object]) -> bool: if "bad" in hub_key: raise RuntimeError("network error") return True results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=_flaky_register, dry_run=False) assert len(results) == 2 ok_result = next(r for r in results if r.hub_key == "ok.musehub.ai") bad_result = next(r for r in results if r.hub_key == "bad.musehub.ai") assert ok_result.hub_registered is True assert bad_result.hub_registered is False # Both entries get their hd_path updated even when hub registration fails assert identity_map["ok.musehub.ai"]["hd_path"] == muse_path(DOMAIN_IDENTITY) assert identity_map["bad.musehub.ai"]["hd_path"] == muse_path(DOMAIN_PAYMENTS) def test_fingerprint_format_stored_in_identity_map(self) -> None: """Fingerprint written to identity_map after live migration is sha256: prefixed.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(return_value=True), dry_run=False) fp = identity_map["localhost:1337"]["fingerprint"] assert fp.startswith("sha256:") assert len(fp) == 71 def test_hd_path_updated_even_when_registration_fails(self) -> None: """identity_map hd_path is corrected to the new value even if hub_register_fn raises.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "handle": "gabriel", "hd_path": _old_path(0), "fingerprint": "a" * 64, "algorithm": "ed25519", } } seed = mnemonic_to_seed(FAKE_MNEMONIC) run_migration( identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(side_effect=RuntimeError("hub unreachable")), dry_run=False, ) assert identity_map["localhost:1337"]["hd_path"] == muse_path(DOMAIN_IDENTITY) def test_dry_run_twice_same_results(self) -> None: """Running dry-run twice on the same identity_map produces identical results.""" from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { "localhost:1337": { "type": "human", "hd_path": _old_path(0), "fingerprint": "a" * 64, } } seed = mnemonic_to_seed(FAKE_MNEMONIC) r1 = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True) r2 = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True) assert r1[0].new_fingerprint == r2[0].new_fingerprint assert r1[0].new_hd_path == r2[0].new_hd_path # ============================================================================ # 12. Security: adversarial inputs and boundary conditions # ============================================================================ class TestSecurity: def test_non_muse_purpose_with_legacy_domain_not_flagged(self) -> None: """Path with domain=0 but wrong purpose is not treated as legacy.""" from muse.core.domain_migration import is_legacy_hd_path # BIP-44 path with domain integer 0 — should never be flagged as legacy assert is_legacy_hd_path("m/44'/0'/0'/0'/0'/0'") is False def test_domain_just_above_range_not_legacy(self) -> None: """Domain integer 7 is outside the 0-6 legacy range and must not be flagged.""" from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path(_old_path(7)) is False def test_large_domain_integer_not_legacy(self) -> None: """A large hash-derived domain integer is never mistaken for a legacy path.""" from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path(muse_path(DOMAIN_IDENTITY)) is False assert is_legacy_hd_path(muse_path(DOMAIN_BLOCKCHAIN)) is False def test_malformed_path_missing_hardened_marker_not_legacy(self) -> None: """A path without hardened markers (no apostrophes) is not treated as legacy.""" from muse.core.domain_migration import is_legacy_hd_path from muse.core.slip010 import MUSE_PURPOSE # Remove hardened markers — soft derivation path assert is_legacy_hd_path(f"m/{MUSE_PURPOSE}/0/0/0/0/0") is False def test_path_with_too_few_segments_not_legacy(self) -> None: """A truncated path (fewer than 6 segments) never matches as legacy.""" from muse.core.domain_migration import is_legacy_hd_path from muse.core.slip010 import MUSE_PURPOSE assert is_legacy_hd_path(f"m/{MUSE_PURPOSE}'/0'/0'") is False def test_path_with_extra_segments_not_legacy(self) -> None: """A path with more than 6 segments is never flagged as legacy.""" from muse.core.domain_migration import is_legacy_hd_path from muse.core.slip010 import MUSE_PURPOSE assert is_legacy_hd_path(f"m/{MUSE_PURPOSE}'/0'/0'/0'/0'/0'/0'") is False def test_empty_string_not_legacy(self) -> None: """Empty string input does not raise and returns False.""" from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path("") is False def test_whitespace_only_not_legacy(self) -> None: """Whitespace-only string does not raise and returns False.""" from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path(" ") is False def test_path_traversal_attempt_not_legacy(self) -> None: """A path-traversal-style string is not treated as a legacy path.""" from muse.core.domain_migration import is_legacy_hd_path assert is_legacy_hd_path("m/1075233755'/../../../etc/passwd") is False def test_derive_fingerprint_invalid_path_raises(self) -> None: """derive_fingerprint_at_path raises ValueError for a non-parseable path.""" from muse.core.domain_migration import derive_fingerprint_at_path from muse.core.bip39 import mnemonic_to_seed seed = mnemonic_to_seed(FAKE_MNEMONIC) with pytest.raises(ValueError, match="Cannot parse"): derive_fingerprint_at_path(seed, "not-a-valid-path") # ============================================================================ # 13. Performance: timing assertions # ============================================================================ class TestPerformance: def test_run_migration_twenty_entries_under_ten_seconds(self) -> None: """run_migration with 20 entries (dry-run) completes within 10 seconds.""" import time from muse.core.domain_migration import run_migration from muse.core.bip39 import mnemonic_to_seed identity_map = { f"hub{i}.example.com:1337": { "type": "human", "hd_path": _old_path(i % 7), "fingerprint": "a" * 64, } for i in range(20) } seed = mnemonic_to_seed(FAKE_MNEMONIC) t0 = time.monotonic() results = run_migration(identity_map=identity_map, seed=seed, hub_register_fn=MagicMock(), dry_run=True) elapsed = time.monotonic() - t0 assert len(results) == 20 assert elapsed < 10.0, f"run_migration took {elapsed:.2f}s — expected < 10s" def test_is_legacy_hd_path_fast_for_thousand_calls(self) -> None: """is_legacy_hd_path classifies 1000 paths in under 1 second.""" import time from muse.core.domain_migration import is_legacy_hd_path paths = [_old_path(i % 7) for i in range(500)] + [muse_path(DOMAIN_IDENTITY)] * 500 t0 = time.monotonic() results = [is_legacy_hd_path(p) for p in paths] elapsed = time.monotonic() - t0 assert sum(results) == 500 # exactly the 500 legacy paths assert elapsed < 1.0, f"is_legacy_hd_path took {elapsed:.2f}s for 1000 calls"