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

Code before:

        # 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(relative_name)
        assert_raises_rpc_error(-4, "Failed to create database", self.master_node.migratewallet, relative_name)

        assert all(wallet not in self.master_node.listwallets() for wallet in [f"{wallet_name}", f"{wallet_name}_watchonly", f"{wallet_name}_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 os.path.exists(solvables_path)
        new_shasum = sha256sum_file(os.path.join(solvables_path , "wallet.dat"))
        assert_equal(original_shasum, new_shasum)

        # Check the wallet we tried to migrate is still BDB
        datfile = os.path.join(absolute_path, "wallet.dat")
        with open(datfile, "rb") as f:
            data = f.read(16)
            _, _, magic = struct.unpack("QII", data)
            assert_equal(magic, BTREE_MAGIC)


    def test_blank(self):
        self.log.info("Test that a blank wallet is migrated")
        wallet = self.create_legacy_wallet("blank", blank=True)
        assert_equal(wallet.getwalletinfo()["blank"], True)
        _, wallet = self.migrate_and_get_rpc("blank")
        assert_equal(wallet.getwalletinfo()["blank"], True)

    def test_avoidreuse(self):
        self.log.info("Test that avoidreuse persists after migration")
        def_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)

        wallet = self.create_legacy_wallet("avoidreuse")
        wallet.setwalletflag("avoid_reuse", True)

        # Import a pubkey to the test wallet and send some funds to it
        reused_imported_addr = def_wallet.getnewaddress()
        wallet.importpubkey(def_wallet.getaddressinfo(reused_imported_addr)["pubkey"])
        imported_utxos = self.create_outpoints(def_wallet, outputs=[{reused_imported_addr: 2}])
        def_wallet.lockunspent(False, imported_utxos)

        # Send funds to the test wallet
        reused_addr = wallet.getnewaddress()
        def_wallet.sendtoaddress(reused_addr, 2)

        self.generate(self.master_node, 1)

        # Send funds from the test wallet with both its own and the imported

Review comment (bug):
Similar to previous test and all other places, can we fail gracefully with more informative message:
```
assert_equal(..., ..., "Test requires that both wallet paths have the same relative path to the common parent")
```

Code after:
        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)

        # 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(relative_name)
        assert_raises_rpc_error(-4, "Failed to create database", self.master_node.migratewallet, relative_name)

        assert all(wallet not in self.master_node.listwallets() for wallet in [f"{wallet_name}", f"{wallet_name}_watchonly", f"{wallet_name}_solvables"])