Inside Bitcoin

1501        assert "taproot_watchonly" in self.master_node.listwallets()
1502        watchonly = self.master_node.get_wallet_rpc("taproot_watchonly")
1503        assert_equal(watchonly.getbalances()["mine"], {"trusted": 5, "untrusted_pending": 0, "immature": 0, "nonmempool": 0})
1504        assert_equal(watchonly.getaddressinfo(rawtr_addr)["ismine"], False)
1505        assert_equal(watchonly.getaddressinfo(tr_addr)["ismine"], True)
1506        assert_equal(watchonly.getaddressinfo(tr_script_addr)["ismine"], True)
1507
1508    def test_no_load_after_migration(self):
1509        self.log.info("Test migration with load_wallet disabled")
1510        default = self.master_node.get_wallet_rpc(self.default_wallet_name)
1511
1512        wallet_name = "no_load_after_migration"
1513        wallet = self.create_legacy_wallet(wallet_name)
1514        addr = wallet.getnewaddress()
1515        txid = default.sendtoaddress(addr, 1)
1516        self.generate(self.master_node, 1)
1517        bals = wallet.getbalances()
1518
1519        # Copy wallet from old node to master node
1520        self.old_node.unloadwallet(wallet_name)
1521        shutil.copytree(
1522            self.old_node.wallets_path / wallet_name,
1523            self.master_node.wallets_path / wallet_name,
1524            dirs_exist_ok=True,
1525        )
1526
1527        migrate_info = self.master_node.migratewallet(wallet_name=wallet_name, load_wallet=False)
1528
1529        # The wallet should NOT be loaded after migration
1530        assert wallet_name not in self.master_node.listwallets()
1531
1532        # The returned wallet_name should be populated correctly
1533        assert_equal(migrate_info["wallet_name"], wallet_name)
1534
1535        # The backup path should be reported and exist on disk
1536        assert os.path.exists(migrate_info["backup_path"])
1537
1538        # The migrated wallet files should be on disk in SQLite format
1539        self.assert_is_sqlite(wallet_name, wallet_loaded = False)
1540
1541        # Load the wallet and verify its state is correct after migration
1542        self.master_node.loadwallet(wallet_name)
1543        loaded_wallet = self.master_node.get_wallet_rpc(wallet_name)
1544        info = loaded_wallet.getwalletinfo()
1545        assert_equal(info["descriptors"], True)
1546        assert_equal(info["format"], "sqlite")
1547        loaded_wallet.gettransaction(txid)
1548        assert_equal(loaded_wallet.getbalance(), bals["mine"]["trusted"])
1549        loaded_wallet.unloadwallet()
1550
1551    def test_no_load_unnamed_wallet_does_not_autoload(self):
1552        self.log.info("Test no-load migration of unnamed wallet suppresses default autoload")
1553        self.master_node = self.nodes[0]
1554        self.old_node = self.nodes[1]
1555
1556        wallet = self.create_legacy_wallet("", load_on_startup=False)
1557        wallet.unloadwallet()
1558
1559        self.stop_node(0)
1560
1561        def remove_wallet_setting(node):
1562            settings_path = node.chain_path / "settings.json"
1563            if not settings_path.exists():
1564                return
1565            with settings_path.open(encoding="utf8") as settings_file:
1566                settings = json.load(settings_file)
1567            settings.pop("wallet", None)
1568            with settings_path.open("w", encoding="utf8") as settings_file:
1569                json.dump(settings, settings_file, indent=4)
1570                settings_file.write("\n")
1571
1572        remove_wallet_setting(self.master_node)
1573        (self.master_node.wallets_path / "wallet.dat").unlink(missing_ok=True)
1574        shutil.copyfile(self.old_node.wallets_path / "wallet.dat", self.master_node.wallets_path / "wallet.dat")
1575        self.start_node(0)
1576
1577        assert_equal(self.master_node.listwallets(), [])
1578
1579        migrate_info = self.master_node.migratewallet(wallet_name="", load_wallet=False)
1580        assert_equal(migrate_info["wallet_name"], "")
1581        backup_path = Path(migrate_info["backup_path"])
1582        assert "" not in self.master_node.listwallets()
1583
1584        with (self.master_node.chain_path / "settings.json").open(encoding="utf8") as settings_file:
1585            wallet_setting_after_migration = json.load(settings_file).get("wallet")
1586
1587        self.restart_node(0)
1588        wallets_after_restart = self.master_node.listwallets()
1589
1590        assert "" not in wallets_after_restart
1591
1592        self.clear_default_wallet(backup_path)
1593        self.master_node.loadwallet(self.default_wallet_name, load_on_startup=True)
1594
1595        assert_equal(wallet_setting_after_migration, [])
1596
1597    def test_no_load_reports_auxiliary_wallet_names(self):
1598        self.log.info("Test no-load migration reports auxiliary wallet names")
1599        wallet_name = "no_load_auxiliary_names"
1600        wallet = self.create_legacy_wallet(wallet_name)
1601
1602        wallet.importaddress(address=self.master_node.get_wallet_rpc(self.default_wallet_name).getnewaddress(), rescan=False)
1603        _, pubkey = generate_keypair(compressed=True, wif=True)
1604        wallet.addmultisigaddress(nrequired=1, keys=[pubkey.hex()])
1605
1606        # Simulate that the wallet was created by the master node with an old version
1607        # so it should already know the wallet prior migration. Set load_on_startup true,
1608        # to simulate that the wallet was automatically being loaded on each restart.
1609        self.master_node.createwallet(wallet_name=wallet_name, load_on_startup=True)
1610        self.master_node.unloadwallet(wallet_name)
1611        self.cleanup_folder(self.master_node.wallets_path / wallet_name)
1612        with (self.master_node.chain_path / "settings.json").open(encoding="utf8") as settings_file:
1613            assert wallet_name in json.load(settings_file).get("wallet", [])
1614
1615        self.old_node.unloadwallet(wallet_name)
1616        shutil.copytree(self.old_node.wallets_path / wallet_name, self.master_node.wallets_path / wallet_name)
1617
1618        migrate_info = self.master_node.migratewallet(wallet_name=wallet_name, load_wallet=False)
1619
1620        assert_equal(migrate_info["wallet_name"], wallet_name)
1621        assert_equal(migrate_info["watchonly_name"], f"{wallet_name}_watchonly")
1622        assert_equal(migrate_info["solvables_name"], f"{wallet_name}_solvables")
1623        assert wallet_name not in self.master_node.listwallets()
1624        assert f"{wallet_name}_watchonly" not in self.master_node.listwallets()
1625        assert f"{wallet_name}_solvables" not in self.master_node.listwallets()
1626
1627        # Migrate wallet with load_wallet=False should have overwritten the wallet settings
1628        # and the load_on_startup setting must removed.
1629        with (self.master_node.chain_path / "settings.json").open(encoding="utf8") as settings_file:
1630            startup_wallets = json.load(settings_file).get("wallet", [])
1631        assert wallet_name not in startup_wallets
1632        assert f"{wallet_name}_watchonly" not in startup_wallets
1633        assert f"{wallet_name}_solvables" not in startup_wallets
1634
1635
1636    def test_solvable_no_privs(self):
1637        self.log.info("Test migrating a multisig that we do not have any private keys for")
1638        wallet = self.create_legacy_wallet("multisig_noprivs")
1639
1640        _, pubkey = generate_keypair(compressed=True, wif=True)
1641
1642        add_ms_res = wallet.addmultisigaddress(nrequired=1, keys=[pubkey.hex()])
1643        addr = add_ms_res["address"]
1644
1645        # The multisig address should be ISMINE_NO but we should have the script info
1646        addr_info = wallet.getaddressinfo(addr)
1647        assert_equal(addr_info["ismine"], False)
1648        assert "hex" in addr_info
1649
1650        migrate_res, wallet = self.migrate_and_get_rpc("multisig_noprivs")
1651        assert_equal(migrate_res["solvables_name"], "multisig_noprivs_solvables")
1652        solvables = self.master_node.get_wallet_rpc(migrate_res["solvables_name"])
1653
1654        # The multisig should not be in the spendable wallet
1655        addr_info = wallet.getaddressinfo(addr)
1656        assert_equal(addr_info["ismine"], False)
1657        assert "hex" not in addr_info
1658
1659        # The multisig address should be in the solvables wallet
1660        addr_info = solvables.getaddressinfo(addr)
1661        assert_equal(addr_info["ismine"], True)
1662        assert_equal(addr_info["solvable"], True)
1663        assert "hex" in addr_info
1664
1665    def test_loading_failure_after_migration(self):
1666        self.log.info("Test that a failed loading of the wallet at the end of migration restores the backup")
1667        self.stop_node(self.old_node.index)
1668        self.old_node.chain = "signet"
1669        self.old_node.replace_in_config([("regtest=", "signet="), ("[regtest]", "[signet]")])
1670        # Disable network sync and prevent disk space warning on small (tmp)fs
1671        self.start_node(self.old_node.index, extra_args=self.old_node.extra_args + ["-maxconnections=0", "-prune=550"])
1672
1673        wallet_name = "failed_load_after_migrate"
1674        self.create_legacy_wallet(wallet_name)
1675        assert_raises_rpc_error(-4, "Wallet loading failed. Wallet files should not be reused across chains.", lambda: self.migrate_and_get_rpc(wallet_name))
1676
1677        # Check the wallet we tried to migrate is still BDB
1678        self.assert_is_bdb(wallet_name)
1679
1680        self.stop_node(self.old_node.index)
1681        self.old_node.chain = "regtest"
1682        self.old_node.replace_in_config([("signet=", "regtest="), ("[signet]", "[regtest]")])
1683        self.start_node(self.old_node.index)
1684        self.connect_nodes(1, 0)
1685
1686    def unsynced_wallet_on_pruned_node_fails(self):
1687        self.log.info("Test migration of an unsynced wallet on a pruned node fails gracefully if loadwallet is set")
1688        wallet = self.create_legacy_wallet("", load_on_startup=False)
1689        last_wallet_synced_block = wallet.getwalletinfo()['lastprocessedblock']['height']
1690        wallet.unloadwallet()
1691
1692        shutil.copyfile(self.old_node.wallets_path / "wallet.dat", self.master_node.wallets_path / "wallet.dat")
1693
1694        # Generate blocks just so the wallet best block is pruned
1695        self.restart_node(0, ["-fastprune", "-prune=1", "-nowallet"])
1696        self.connect_nodes(0, 1)
1697        self.generate(self.master_node, 450, sync_fun=self.no_op)
1698        self.master_node.pruneblockchain(250)
1699        # Ensure next block to sync is unavailable
1700        assert_raises_rpc_error(-1, "Block not available (pruned data)", self.master_node.getblock, self.master_node.getblockhash(last_wallet_synced_block + 1))
1701
1702        # Check migration failure
1703        mocked_time = int(time.time())
1704        self.master_node.setmocktime(mocked_time)
1705        assert_raises_rpc_error(-4, "last wallet synchronisation goes beyond pruned data. You need to -reindex (download the whole blockchain again in case of a pruned node)", self.master_node.migratewallet, wallet_name="")
1706        self.master_node.setmocktime(0)
1707
1708        # Verify the /wallets/ path exists, the wallet is still BDB and the backup file is there.
1709        assert self.master_node.wallets_path.exists()
1710        self.assert_is_bdb("")
1711        backup_path = self.master_node.wallets_path / f"default_wallet_{mocked_time}.legacy.bak"
1712        assert backup_path.exists()
1713
1714        self.log.info("Test migration of an unsynced wallet on a pruned node does not fails if loadwallet is not set")
1715        self.master_node.migratewallet("", load_wallet=False)
1716        # The wallet should NOT be loaded after migration
1717        assert "" not in self.master_node.listwallets()
1718        # Load the wallet should fail
1719        assert_raises_rpc_error(-4, "last wallet synchronisation goes beyond pruned data. You need to -reindex (download the whole blockchain again in case of a pruned node)", self.master_node.loadwallet, filename="")
1720
1721        self.clear_default_wallet(backup_path)
1722
1723    @staticmethod
1724    def erase_bdb_record(wallet_dat_path, key):
1725        data = bytearray(wallet_dat_path.read_bytes())
1726        idx = data.find(key)
1727        assert idx != -1, f"{key!r} not found in wallet.dat"
1728
1729        for i in range(idx, idx + len(key)):
1730            data[i] = 0
1731
1732        wallet_dat_path.write_bytes(data)
1733
1734    def test_missing_bestblock(self):
1735        self.log.info("Test migrating legacy BDB wallet without bestblock record")
1736        wallet_name = "nobestblock"
1737        wallet = self.create_legacy_wallet(wallet_name)
1738        wallet.unloadwallet()
1739
1740        # Erase block locator records like if this would be a pre-#152 wallet
1741        self.erase_bdb_record(self.old_node.wallets_path / wallet_name / "wallet.dat", b"bestblock_nomerkle")
1742        self.erase_bdb_record(self.old_node.wallets_path / wallet_name / "wallet.dat", b"bestblock")
1743
1744        shutil.copytree(self.old_node.wallets_path / wallet_name, self.master_node.wallets_path / wallet_name, dirs_exist_ok=True)
1745        # Migrate, checking that rescan occurs
1746        with self.master_node.assert_debug_log(expected_msgs=["Rescanning"], unexpected_msgs=[]):
1747            self.master_node.migratewallet(wallet_name)
1748
1749        wallet = self.master_node.get_wallet_rpc(wallet_name)
1750        info = wallet.getwalletinfo()