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

Code before:
        wallet.migratewallet()
        # Post-migration verification
        check_comments()

        wallet.unloadwallet()


    def unsynced_wallet_on_pruned_node_fails(self):
        self.log.info("Test migration of an unsynced wallet on a pruned node fails gracefully")
        shutil.rmtree(self.nodes[0].wallets_path / "database", ignore_errors=True)
        wallet = self.create_legacy_wallet("", load_on_startup=False)
        last_wallet_synced_block = wallet.getwalletinfo()['lastprocessedblock']['height']

        # Generate blocks just so the wallet best block is pruned
        self.restart_node(0, ["-fastprune", "-prune=1", "-nowallet"])
        self.generate(self.nodes[0], 450, sync_fun=self.no_op)
        self.nodes[0].pruneblockchain(250)
        # Ensure next block to sync is unavailable
        assert_raises_rpc_error(-1, "Block not available (pruned data)", self.nodes[0].getblock, self.nodes[0].getblockhash(last_wallet_synced_block + 1))

        # Check migration failure
        mocked_time = int(time.time())
        self.nodes[0].setmocktime(mocked_time)
        assert_raises_rpc_error(-4, "last wallet synchronisation goes beyond pruned data. You need to -reindex (download the whole blockchain again in case of pruned node)\nUnable to restore backup of wallet.", self.nodes[0].migratewallet, wallet_name="")
        self.nodes[0].setmocktime(0)

        # The wallet will not have been restored, and the backup will be in the wallets directory.
        backup_path = self.nodes[0].wallets_path / f"default_wallet_{mocked_time}.legacy.bak"
        assert backup_path.exists()

    def run_test(self):
        self.generate(self.nodes[0], 101)

        # TODO: Test the actual records in the wallet for these tests too. The behavior may be correct, but the data written may not be what we actually want
        self.test_basic()
        self.test_multisig()
        self.test_other_watchonly()
        self.test_no_privkeys()
        self.test_pk_coinbases()
        self.test_encrypted()
        self.test_unloaded()
        self.test_unloaded_by_path()
        self.test_default_wallet_failure()
        self.test_default_wallet()
        self.test_default_wallet_watch_only()
        self.test_direct_file()
        self.test_addressbook()
        self.test_migrate_raw_p2sh()
        self.test_conflict_txs()
        self.test_hybrid_pubkey()
        self.test_failed_migration_cleanup()

Review comment (bug):
probably only if retouching:

```suggestion
        # Verify the /wallets/ path exists, the wallet is still BDB and the backup file is there.
        assert self.nodes[0].wallets_path.exists()
        with open(self.nodes[0].wallets_path / "wallet.dat", "rb") as f:
            data = f.read(16)
            _, _, magic = struct.unpack("QII", data)
            assert_equal(magic, BTREE_MAGIC)
```

`assert self.nodes[0].wallets_path.exists()` is redundant with the backup path check below, but would make the reason for test failure more obvious and makes the rebase cleaner

Code after:
        assert_equal(len(watchonly_utxos), 1)
        assert_equal(watchonly_utxos[0]["reused"], True)

    def test_preserve_tx_extra_info(self):
        self.log.info("Test that tx extra data is preserved after migration")
        def_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)

        # Create and fund wallet
        wallet = self.create_legacy_wallet("persist_comments")
        def_wallet.sendtoaddress(wallet.getnewaddress(), 2)

        self.generate(self.nodes[0], 6)

        # Create tx and bump it to store 'replaced_by_txid' and 'replaces_txid' data within the transactions.
        # Additionally, store an extra comment within the original tx.
        extra_comment = "don't discard me"
        original_tx_id = wallet.sendtoaddress(address=wallet.getnewaddress(), amount=1, comment=extra_comment)
        bumped_tx = wallet.bumpfee(txid=original_tx_id)

        def check_comments():
            for record in wallet.listtransactions():
                if record["txid"] == original_tx_id:
                    assert_equal(record["replaced_by_txid"], bumped_tx["txid"])
                    assert_equal(record['comment'], extra_comment)
                elif record["txid"] == bumped_tx["txid"]:
                    assert_equal(record["replaces_txid"], original_tx_id)

        # Pre-migration verification
        check_comments()
        # Migrate
        wallet.migratewallet()
        # Post-migration verification
        check_comments()

        wallet.unloadwallet()


    def unsynced_wallet_on_pruned_node_fails(self):
        self.log.info("Test migration of an unsynced wallet on a pruned node fails gracefully")
        shutil.rmtree(self.nodes[0].wallets_path / "database", ignore_errors=True)
        wallet = self.create_legacy_wallet("", load_on_startup=False)
        last_wallet_synced_block = wallet.getwalletinfo()['lastprocessedblock']['height']

        # Generate blocks just so the wallet best block is pruned
        self.restart_node(0, ["-fastprune", "-prune=1", "-nowallet"])
        self.generate(self.nodes[0], 450, sync_fun=self.no_op)
        self.nodes[0].pruneblockchain(250)
        # Ensure next block to sync is unavailable
        assert_raises_rpc_error(-1, "Block not available (pruned data)", self.nodes[0].getblock, self.nodes[0].getblockhash(last_wallet_synced_block + 1))

        # Check migration failure