Inside Bitcoin

501            {
502                "desc": desc2,
503                "keypool": True,
504                "timestamp": "now",
505            }
506        ])
507        assert_equal(res[0]["success"], True)
508        assert_equal(res[1]["success"], True)
509        # Before migrating, we can fetch addr1 from the keypool
510        assert_equal(watchonly1.getnewaddress(address_type="bech32"), addr1)
511
512        _, watchonly1 = self.migrate_and_get_rpc("watchonly1")
513        info = watchonly1.getwalletinfo()
514        assert_equal(info["descriptors"], True)
515        assert_equal(info["private_keys_enabled"], False)
516        self.assert_is_sqlite("watchonly1")
517        # After migrating, the "keypool" is empty
518        assert_raises_rpc_error(-4, "Error: This wallet has no available keys", watchonly1.getnewaddress)
519
520        self.log.info("Test migration of a watch-only empty wallet")
521        for idx, is_blank in enumerate([True, False], start=1):
522            wallet_name = f"watchonly_empty{idx}"
523            self.create_legacy_wallet(wallet_name, disable_private_keys=True, blank=is_blank)
524            _, watchonly_empty = self.migrate_and_get_rpc(wallet_name)
525            info = watchonly_empty.getwalletinfo()
526            assert_equal(info["private_keys_enabled"], False)
527            assert_equal(info["blank"], is_blank)
528
529    def test_pk_coinbases(self):
530        self.log.info("Test migration of a wallet using old pk() coinbases")
531        wallet = self.create_legacy_wallet("pkcb")
532
533        addr = wallet.getnewaddress()
534        addr_info = wallet.getaddressinfo(addr)
535        desc = descsum_create("pk(" + addr_info["pubkey"] + ")")
536
537        self.generatetodescriptor(self.master_node, 1, desc)
538
539        bals = wallet.getbalances()
540        bals["mine"]["nonmempool"] = Decimal('0.0')
541
542        _, wallet = self.migrate_and_get_rpc("pkcb")
543
544        assert_equal(bals, wallet.getbalances())
545
546    def test_encrypted(self):
547        self.log.info("Test migration of an encrypted wallet")
548        wallet = self.create_legacy_wallet("encrypted")
549        default = self.master_node.get_wallet_rpc(self.default_wallet_name)
550
551        wallet.encryptwallet("pass")
552        addr = wallet.getnewaddress()
553        txid = default.sendtoaddress(addr, 1)
554        self.generate(self.master_node, 1)
555        bals = wallet.getbalances()
556        bals["mine"]["nonmempool"] = Decimal('0.0')
557
558        # Use self.migrate_and_get_rpc to test this error to get everything copied over to the master node
559        assert_raises_rpc_error(-4, "Error: Wallet decryption failed, the wallet passphrase was not provided or was incorrect", self.migrate_and_get_rpc, "encrypted")
560
561        # Use the RPC directly on the master node for the rest of these checks
562        self.master_node.bumpmocktime(1) # Prevents filename duplication on wallet backups which is a problem on Windows
563        assert_raises_rpc_error(-4, "Error: Wallet decryption failed, the wallet passphrase was not provided or was incorrect", self.master_node.migratewallet, "encrypted", "badpass")
564
565        self.master_node.bumpmocktime(1) # Prevents filename duplication on wallet backups which is a problem on Windows
566        assert_raises_rpc_error(-4, "The passphrase contains a null character", self.master_node.migratewallet, "encrypted", "pass\0with\0null")
567
568        # Verify we can properly migrate the encrypted wallet
569        self.master_node.bumpmocktime(1) # Prevents filename duplication on wallet backups which is a problem on Windows
570        self.master_node.migratewallet("encrypted", passphrase="pass")
571        wallet = self.master_node.get_wallet_rpc("encrypted")
572
573        info = wallet.getwalletinfo()
574        assert_equal(info["descriptors"], True)
575        assert_equal(info["format"], "sqlite")
576        assert_equal(info["unlocked_until"], 0)
577        wallet.gettransaction(txid)
578
579        assert_equal(bals, wallet.getbalances())
580
581    def test_nonexistent(self):
582        self.log.info("Check migratewallet errors for nonexistent wallets")
583        default = self.master_node.get_wallet_rpc(self.default_wallet_name)
584        assert_raises_rpc_error(-8, "The RPC endpoint wallet and the wallet name parameter specify different wallets", default.migratewallet, "someotherwallet")
585        assert_raises_rpc_error(-8, "Either the RPC endpoint wallet or the wallet name parameter must be provided", self.master_node.migratewallet)
586        assert_raises_rpc_error(-4, "Error: Wallet does not exist", self.master_node.migratewallet, "notawallet")
587
588    def test_unloaded_by_path(self):
589        self.log.info("Test migration of a wallet that isn't loaded, specified by path")
590        wallet = self.create_legacy_wallet("notloaded2")
591        default = self.master_node.get_wallet_rpc(self.default_wallet_name)
592
593        addr = wallet.getnewaddress()
594        txid = default.sendtoaddress(addr, 1)
595        self.generate(self.master_node, 1)
596        bals = wallet.getbalances()
597        bals["mine"]["nonmempool"] = Decimal('0.0')
598
599        wallet.unloadwallet()
600
601        wallet_file_path = self.old_node.wallets_path / "notloaded2"
602        self.master_node.migratewallet(wallet_file_path)
603
604        # Because we gave the name by full path, the loaded wallet's name is that path too.
605        wallet = self.master_node.get_wallet_rpc(str(wallet_file_path))
606
607        info = wallet.getwalletinfo()
608        assert_equal(info["descriptors"], True)
609        assert_equal(info["format"], "sqlite")
610        wallet.gettransaction(txid)
611
612        assert_equal(bals, wallet.getbalances())
613
614    def test_wallet_with_path(self, wallet_path):
615        self.log.info("Test migrating a wallet with the following path/name: %s", wallet_path)
616        # the wallet data is actually inside of path/that/ends/
617        wallet = self.create_legacy_wallet(wallet_path)
618        default = self.master_node.get_wallet_rpc(self.default_wallet_name)
619
620        addr = wallet.getnewaddress()
621        txid = default.sendtoaddress(addr, 1)
622        self.generate(self.master_node, 1)
623        bals = wallet.getbalances()
624        bals["mine"]["nonmempool"] = Decimal('0.0')
625
626        _, wallet = self.migrate_and_get_rpc(wallet_path)
627
628        assert wallet.gettransaction(txid)
629
630        assert_equal(bals, wallet.getbalances())
631
632    def clear_default_wallet(self, backup_file):
633        # Test cleanup: Clear unnamed default wallet for subsequent tests
634        (self.old_node.wallets_path / "wallet.dat").unlink()
635        (self.master_node.wallets_path / "wallet.dat").unlink(missing_ok=True)
636        with suppress(FileNotFoundError):
637            (self.master_node.wallets_path / "default_wallet_watchonly" / "wallet.dat").unlink()
638            (self.master_node.wallets_path / "default_wallet_watchonly").rmdir()
639
640            (self.master_node.wallets_path / "default_wallet_solvables" / "wallet.dat").unlink()
641            (self.master_node.wallets_path / "default_wallet_solvables").rmdir()
642        backup_file.unlink()
643
644    def test_default_wallet(self):
645        self.log.info("Test migration of the wallet named as the empty string")
646        wallet = self.create_legacy_wallet("")
647
648        res, wallet = self.migrate_and_get_rpc("")
649        info = wallet.getwalletinfo()
650        assert_equal(info["descriptors"], True)
651        assert_equal(info["format"], "sqlite")
652
653        walletdir_list = wallet.listwalletdir()
654        assert {"name": info["walletname"]} in [{"name": w["name"]} for w in walletdir_list["wallets"]]
655
656        # Make sure the backup uses a non-empty filename
657        # migrate_and_get_rpc already checks for backup file existence
658        assert os.path.basename(res["backup_path"]).startswith("default_wallet")
659
660        wallet.unloadwallet()
661        self.clear_default_wallet(backup_file=Path(res["backup_path"]))
662
663    def test_default_wallet_watch_only(self):
664        self.log.info("Test unnamed (default) watch-only wallet migration")
665        master_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)
666        wallet = self.create_legacy_wallet("", blank=True)
667        wallet.importaddress(master_wallet.getnewaddress(address_type="legacy"))
668
669        res, wallet = self.migrate_and_get_rpc("")
670
671        info = wallet.getwalletinfo()
672        assert_equal(info["descriptors"], True)
673        assert_equal(info["format"], "sqlite")
674        assert_equal(info["private_keys_enabled"], False)
675        assert_equal(info["walletname"], "default_wallet_watchonly")
676        # Check the default wallet is not available anymore
677        assert not (self.master_node.wallets_path / "wallet.dat").exists()
678
679        wallet.unloadwallet()
680        self.clear_default_wallet(backup_file=Path(res["backup_path"]))
681
682    def test_migration_failure(self, wallet_name):
683        is_default = wallet_name == ""
684        wallet_pretty_name = "unnamed (default)" if is_default else f'"{wallet_name}"'
685        self.log.info(f"Test failure during migration of wallet named: {wallet_pretty_name}")
686        # Preface, set up legacy wallet and unload it
687        master_wallet = self.master_node.get_wallet_rpc(self.default_wallet_name)
688        wallet = self.create_legacy_wallet(wallet_name, blank=True)
689        wallet.importaddress(master_wallet.getnewaddress(address_type="legacy"))
690        wallet.unloadwallet()
691
692        if os.path.isabs(wallet_name):
693            old_path = master_path = Path(wallet_name)
694        else:
695            old_path = self.old_node.wallets_path / wallet_name
696            master_path = self.master_node.wallets_path / wallet_name
697            os.makedirs(master_path, exist_ok=True)
698            shutil.copyfile(old_path / "wallet.dat", master_path / "wallet.dat")
699
700        # This will be the watch-only directory the migration tries to create,
701        # we make migration fail by placing a wallet.dat file there.
702        wo_prefix = wallet_name or "default_wallet"
703        # wo_prefix might have path characters in it, this corresponds with
704        # DoMigration().
705        wo_dirname = f"{wo_prefix}_watchonly"
706        watch_only_dir = self.master_node.wallets_path / wo_dirname
707        os.mkdir(watch_only_dir)
708        shutil.copyfile(old_path / "wallet.dat", watch_only_dir / "wallet.dat")
709
710        mocked_time = int(time.time())
711        self.master_node.setmocktime(mocked_time)
712        assert_raises_rpc_error(-4, "Failed to create database", self.master_node.migratewallet, wallet_name)
713        self.master_node.setmocktime(0)
714
715        # Verify the /wallets/ path exists.
716        assert self.master_node.wallets_path.exists()
717
718        # Verify both wallet paths exist.
719        assert Path(old_path / "wallet.dat").exists()
720        assert Path(master_path / "wallet.dat").exists()
721
722        backup_prefix = "default_wallet" if is_default else os.path.basename(os.path.abspath(master_path))
723        backup_path = self.master_node.wallets_path / f"{backup_prefix}_{mocked_time}.legacy.bak"
724        assert backup_path.exists()
725
726        self.assert_is_bdb(wallet_name)
727
728        # Cleanup
729        if is_default:
730            self.clear_default_wallet(backup_path)
731        else:
732            backup_path.unlink()
733            Path(watch_only_dir / "wallet.dat").unlink()
734            Path(watch_only_dir).rmdir()
735            Path(master_path / "wallet.dat").unlink()
736            Path(old_path / "wallet.dat").unlink(missing_ok=True)
737
738    def test_direct_file(self):
739        self.log.info("Test migration of a wallet that is not in a wallet directory")
740        wallet = self.create_legacy_wallet("plainfile")
741        wallet.unloadwallet()
742
743        shutil.copyfile(
744            self.old_node.wallets_path / "plainfile" / "wallet.dat" ,
745            self.master_node.wallets_path / "plainfile"
746        )
747        assert (self.master_node.wallets_path / "plainfile").is_file()
748
749        mocked_time = int(time.time())
750        self.master_node.setmocktime(mocked_time)