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

Code before:
        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
        wallet.sendall([def_wallet.getnewaddress()])
        def_wallet.sendall(recipients=[def_wallet.getnewaddress()], inputs=imported_utxos)
        self.generate(self.master_node, 1)
        balances = wallet.getbalances()
        assert_equal(balances["mine"]["trusted"], 0)
        assert_equal(balances["watchonly"]["trusted"], 0)

        # Reuse the addresses
        def_wallet.sendtoaddress(reused_addr, 1)
        def_wallet.sendtoaddress(reused_imported_addr, 1)
        self.generate(self.master_node, 1)
        balances = wallet.getbalances()
        assert_equal(balances["mine"]["used"], 1)
        # Reused watchonly will not show up in balances
        assert_equal(balances["watchonly"]["trusted"], 0)
        assert_equal(balances["watchonly"]["untrusted_pending"], 0)
        assert_equal(balances["watchonly"]["immature"], 0)

        utxos = wallet.listunspent()
        assert_equal(len(utxos), 2)
        for utxo in utxos:
            assert_equal(utxo["reused"], True)

        # Migrate
        _, wallet = self.migrate_and_get_rpc("avoidreuse")
        watchonly_wallet = self.master_node.get_wallet_rpc("avoidreuse_watchonly")

        # One utxo in each wallet, marked used
        utxos = wallet.listunspent()
        assert_equal(len(utxos), 1)
        assert_equal(utxos[0]["reused"], True)
        watchonly_utxos = watchonly_wallet.listunspent()
        assert_equal(len(watchonly_utxos), 1)
        assert_equal(watchonly_utxos[0]["reused"], True)


Review comment (refactor):
**Minor / Nit:** A separate helper function could be introduced to validate it's BDB for better readability.
Example:
```
def is_bdb_wallet(filepath):
    with open(filepath, "rb") as f:
        data = f.read(16)
        _, _, magic = struct.unpack("QII", data)
        return magic == BTREE_MAGIC
```

and then:
```
assert is_bdb_wallet(datfile)
```


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

        assert not (self.master_node.wallets_path / f"{wallet_name}_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