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

Code before:

        wallet_path = self.master_node.wallets_path / wallet_name / "wallet.dat"
        shutil.move(self.old_node.wallets_path / wallet_name / "wallet.dat", wallet_path)

        # Create wallet directory with the watch-only name and a wallet file.
        # Because the wallet dir exists, this will cause migration to fail.
        watch_only_dir = self.master_node.wallets_path / ".." / "_watchonly"
        os.mkdir(watch_only_dir)
        shutil.copyfile(wallet_path, watch_only_dir / "wallet.dat")

        mocked_time = int(time.time())
        self.master_node.setmocktime(mocked_time)
        assert_raises_rpc_error(-4, "Failed to create database", self.master_node.migratewallet, wallet_name)
        self.master_node.setmocktime(0)

        # Verify the original wallet was not deleted
        assert os.path.exists(wallet_path)
        # And verify it is still a BDB wallet
        self.assert_is_bdb(wallet_name)
        # Verify the watchonly wallet was not deleted
        assert os.path.exists(watch_only_dir / "wallet.dat")

        # Check backup file exists.
        backup_prefix = os.path.basename(os.path.abspath(wallet_path.parent))
        backup_path = os.path.join(self.master_node.wallets_path, f"{backup_prefix}_{mocked_time}.legacy.bak")
        assert os.path.exists(backup_path)

        # Cleanup for next test
        os.unlink(backup_path)
        os.unlink(wallet_path)
        shutil.rmtree(watch_only_dir)

    def test_absolute_path_wallet_failure(self):
        self.log.info("Test failure during migration of absolute path wallet")

        wallet_name = "absolutely"
        absolute_path = os.path.abspath(self.master_node.datadir_path / wallet_name)

        master_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)
        wallet = self.create_legacy_wallet(absolute_path, blank=True)
        wallet.importaddress(master_wallet.getnewaddress(address_type="legacy"))

        # Unload legacy wallet, see migrate_and_get_rpc
        self.old_node.unloadwallet(absolute_path)
        self.old_node.loadwallet(absolute_path)
        self.old_node.unloadwallet(absolute_path)

        # Create wallet directory with the watch-only name and a wallet file.
        # Because the wallet dir exists, this will cause migration to fail.
        watch_only_dir = self.master_node.datadir_path / f"{wallet_name}_watchonly"
        os.mkdir(watch_only_dir)

Review comment (bug):
There are commonalities in what we check when the migration either succeeds or fails. I suggested in PR #34221 to use `migrate_and_get_rpc` more in case of errors as well here in this comment: https://github.com/bitcoin/bitcoin/pull/34221#pullrequestreview-3639390846

We might be able to avoid setting mock time in the tests and checking for backup (& original) files in every test case.

Code after:
        assert backup_path.exists()

        self.assert_is_bdb(wallet_name)

        # Cleanup
        if is_default:
            self.clear_default_wallet(backup_path)
        else:
            backup_path.unlink()
            Path(watch_only_dir / "wallet.dat").unlink()
            Path(watch_only_dir).rmdir()
            Path(master_path / "wallet.dat").unlink()
            Path(old_path / "wallet.dat").unlink(missing_ok=True)

    def test_direct_file(self):
        self.log.info("Test migration of a wallet that is not in a wallet directory")
        wallet = self.create_legacy_wallet("plainfile")
        wallet.unloadwallet()

        shutil.copyfile(
            self.old_node.wallets_path / "plainfile" / "wallet.dat" ,
            self.master_node.wallets_path / "plainfile"
        )
        assert (self.master_node.wallets_path / "plainfile").is_file()

        mocked_time = int(time.time())
        self.master_node.setmocktime(mocked_time)
        migrate_res = self.master_node.migratewallet("plainfile")
        assert_equal(f"plainfile_{mocked_time}.legacy.bak", os.path.basename(migrate_res["backup_path"]))
        wallet = self.master_node.get_wallet_rpc("plainfile")
        info = wallet.getwalletinfo()
        assert_equal(info["descriptors"], True)
        assert_equal(info["format"], "sqlite")

        assert (self.master_node.wallets_path / "plainfile").is_dir()
        assert (self.master_node.wallets_path / "plainfile" / "wallet.dat").is_file()

    def test_addressbook(self):
        df_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)

        self.log.info("Test migration of address book data")
        wallet = self.create_legacy_wallet("legacy_addrbook")
        df_wallet.sendtoaddress(wallet.getnewaddress(), 3)

        # Import watch-only script to create a watch-only wallet after migration
        watch_addr = df_wallet.getnewaddress()
        wallet.importaddress(watch_addr)
        df_wallet.sendtoaddress(watch_addr, 2)

        # Import solvable script
        multi_addr1 = wallet.getnewaddress()