Repository: bitcoin/bitcoin
File: test/functional/wallet_migration.py (Python)

Code before:

        # Pre-migration verification
        check_comments()
        # Migrate
        _, wallet = self.migrate_and_get_rpc("persist_comments")
        # Post-migration verification
        check_comments()

        wallet.unloadwallet()

    def test_migrate_simple_watch_only(self):
        self.log.info("Test migrating a watch-only p2pk script")
        wallet = self.create_legacy_wallet("bare_p2pk", blank=True)
        _, pubkey = generate_keypair()
        p2pk_script = key_to_p2pk_script(pubkey)
        wallet.importaddress(address=p2pk_script.hex())
        # Migrate wallet in the latest node
        res, _ = self.migrate_and_get_rpc("bare_p2pk")
        wo_wallet = self.master_node.get_wallet_rpc(res['wallet_name'])
        assert_equal(wo_wallet.listdescriptors()['descriptors'][0]['desc'], descsum_create(f'pk({pubkey.hex()})'))
        assert_equal(wo_wallet.getwalletinfo()["private_keys_enabled"], False)

        # Ensure that migrating a wallet with watch-only scripts does not create a spendable wallet.
        assert_equal('bare_p2pk_watchonly', res['wallet_name'])
        assert "bare_p2pk" not in self.master_node.listwallets()
        assert "bare_p2pk" not in [w["name"] for w in self.master_node.listwalletdir()["wallets"]]

        wo_wallet.unloadwallet()

    def test_manual_keys_import(self):
        self.log.info("Test migrating standalone private keys")
        wallet = self.create_legacy_wallet("import_privkeys", blank=True)
        privkey, pubkey = generate_keypair(wif=True)
        wallet.importprivkey(privkey=privkey, label="hi", rescan=False)

        # Migrate and verify
        res, wallet = self.migrate_and_get_rpc("import_privkeys")

        # There should be descriptors containing the imported key for: pk(), pkh(), sh(wpkh()), wpkh()
        key_origin = hash160(pubkey)[:4].hex()
        pubkey_hex = pubkey.hex()
        combo_desc = descsum_create(f"combo([{key_origin}]{pubkey_hex})")

        # Verify all expected descriptors were migrated
        migrated_desc = [item['desc'] for item in wallet.listdescriptors()['descriptors'] if pubkey.hex() in item['desc']]
        assert_equal([combo_desc], migrated_desc)
        wallet.unloadwallet()

        ######################################################
        self.log.info("Test migrating standalone public keys")
        wallet = self.create_legacy_wallet("import_pubkeys", blank=True)

Review comment (bug):
In e7fec273593d825e0e7dbec46a026aacc58c479f "test: Migration fail recovery w/ `../` in path"

I think `failed_watchonly` is the wrong wallet name. Probably best to do `f"{wallet_name}_watchonly"`

Code after:
        original_shasum = sha256sum_file(solvables_path / "wallet.dat")

        self.old_node.loadwallet("failed")

        # Add a multisig so that a solvables wallet is created
        wallet.addmultisigaddress(2, [wallet.getnewaddress(), get_generate_key().pubkey])
        wallet.importaddress(get_generate_key().p2pkh_addr)

        self.old_node.unloadwallet("failed")
        shutil.copytree(self.old_node.wallets_path / "failed", self.master_node.wallets_path / "failed")
        assert_raises_rpc_error(-4, "Failed to create database", self.master_node.migratewallet, "failed")

        assert all(wallet not in self.master_node.listwallets() for wallet in ["failed", "failed_watchonly", "failed_solvables"])

        assert not (self.master_node.wallets_path / "failed_watchonly").exists()
        # Since the file in failed_solvables is one that we put there, migration shouldn't touch it
        assert solvables_path.exists()
        new_shasum = sha256sum_file(solvables_path / "wallet.dat")
        assert_equal(original_shasum, new_shasum)

        # Check the wallet we tried to migrate is still BDB
        with open(self.master_node.wallets_path / "failed" / "wallet.dat", "rb") as f:
            data = f.read(16)
            _, _, magic = struct.unpack("QII", data)
            assert_equal(magic, BTREE_MAGIC)

    def test_failed_migration_cleanup_relative_path(self):
        self.log.info("Test that a failed migration with a relative path is cleaned up")

        # Get the nearest common path of both nodes' wallet paths.
        common_parent = os.path.commonpath([self.master_node.wallets_path, self.old_node.wallets_path])

        # This test assumes that the relative path from each wallet directory to the common path is identical.
        assert_equal(os.path.relpath(common_parent, start=self.master_node.wallets_path), os.path.relpath(common_parent, start=self.old_node.wallets_path))

        wallet_name = "relativefailure"
        absolute_path = os.path.abspath(os.path.join(common_parent, wallet_name))
        relative_name = os.path.relpath(absolute_path, start=self.master_node.wallets_path)

        wallet = self.create_legacy_wallet(relative_name)

        # Make a copy of the wallet with the solvables wallet name so that we are unable
        # to create the solvables wallet when migrating, thus failing to migrate
        wallet.unloadwallet()
        solvables_path = os.path.join(common_parent, f"{wallet_name}_solvables")

        shutil.copytree(self.old_node.wallets_path / relative_name, solvables_path)
        original_shasum = sha256sum_file(os.path.join(solvables_path, "wallet.dat"))

        self.old_node.loadwallet(relative_name)