mirror of https://github.com/bitcoin/bitcoin.git
1026 lines
48 KiB
Python
Executable File
1026 lines
48 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# Copyright (c) 2020-2022 The Bitcoin Core developers
|
|
# Distributed under the MIT software license, see the accompanying
|
|
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
|
|
"""Test Migrating a wallet from legacy to descriptor."""
|
|
|
|
import random
|
|
import shutil
|
|
import struct
|
|
import time
|
|
|
|
from test_framework.address import (
|
|
script_to_p2sh,
|
|
key_to_p2pkh,
|
|
key_to_p2wpkh,
|
|
)
|
|
from test_framework.bdb import BTREE_MAGIC
|
|
from test_framework.descriptors import descsum_create
|
|
from test_framework.key import ECPubKey
|
|
from test_framework.test_framework import BitcoinTestFramework
|
|
from test_framework.messages import COIN, CTransaction, CTxOut
|
|
from test_framework.script_util import key_to_p2pkh_script, script_to_p2sh_script, script_to_p2wsh_script
|
|
from test_framework.util import (
|
|
assert_equal,
|
|
assert_raises_rpc_error,
|
|
sha256sum_file,
|
|
)
|
|
from test_framework.wallet_util import (
|
|
get_generate_key,
|
|
)
|
|
|
|
|
|
class WalletMigrationTest(BitcoinTestFramework):
|
|
def add_options(self, parser):
|
|
self.add_wallet_options(parser)
|
|
|
|
def set_test_params(self):
|
|
self.setup_clean_chain = True
|
|
self.num_nodes = 1
|
|
self.extra_args = [[]]
|
|
self.supports_cli = False
|
|
|
|
def skip_test_if_missing_module(self):
|
|
self.skip_if_no_wallet()
|
|
self.skip_if_no_sqlite()
|
|
self.skip_if_no_bdb()
|
|
|
|
def assert_is_sqlite(self, wallet_name):
|
|
wallet_file_path = self.nodes[0].wallets_path / wallet_name / self.wallet_data_filename
|
|
with open(wallet_file_path, 'rb') as f:
|
|
file_magic = f.read(16)
|
|
assert_equal(file_magic, b'SQLite format 3\x00')
|
|
assert_equal(self.nodes[0].get_wallet_rpc(wallet_name).getwalletinfo()["format"], "sqlite")
|
|
|
|
def create_legacy_wallet(self, wallet_name, **kwargs):
|
|
self.nodes[0].createwallet(wallet_name=wallet_name, descriptors=False, **kwargs)
|
|
wallet = self.nodes[0].get_wallet_rpc(wallet_name)
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], False)
|
|
assert_equal(info["format"], "bdb")
|
|
return wallet
|
|
|
|
def migrate_wallet(self, wallet_rpc, *args, **kwargs):
|
|
# Helper to ensure that only migration happens
|
|
# Since we may rescan on loading of a wallet, make sure that the best block
|
|
# is written before beginning migration
|
|
# Reload to force write that record
|
|
wallet_name = wallet_rpc.getwalletinfo()["walletname"]
|
|
wallet_rpc.unloadwallet()
|
|
self.nodes[0].loadwallet(wallet_name)
|
|
# Migrate, checking that rescan does not occur
|
|
with self.nodes[0].assert_debug_log(expected_msgs=[], unexpected_msgs=["Rescanning"]):
|
|
return wallet_rpc.migratewallet(*args, **kwargs)
|
|
|
|
def assert_addr_info_equal(self, addr_info, addr_info_old):
|
|
assert_equal(addr_info["address"], addr_info_old["address"])
|
|
assert_equal(addr_info["scriptPubKey"], addr_info_old["scriptPubKey"])
|
|
assert_equal(addr_info["ismine"], addr_info_old["ismine"])
|
|
assert_equal(addr_info["hdkeypath"], addr_info_old["hdkeypath"].replace("'","h"))
|
|
assert_equal(addr_info["solvable"], addr_info_old["solvable"])
|
|
assert_equal(addr_info["ischange"], addr_info_old["ischange"])
|
|
assert_equal(addr_info["hdmasterfingerprint"], addr_info_old["hdmasterfingerprint"])
|
|
|
|
def assert_list_txs_equal(self, received_list_txs, expected_list_txs):
|
|
for d in received_list_txs:
|
|
if "parent_descs" in d:
|
|
del d["parent_descs"]
|
|
for d in expected_list_txs:
|
|
if "parent_descs" in d:
|
|
del d["parent_descs"]
|
|
assert_equal(received_list_txs, expected_list_txs)
|
|
|
|
def check_address(self, wallet, addr, is_mine, is_change, label):
|
|
addr_info = wallet.getaddressinfo(addr)
|
|
assert_equal(addr_info['ismine'], is_mine)
|
|
assert_equal(addr_info['ischange'], is_change)
|
|
if label is not None:
|
|
assert_equal(addr_info['labels'], [label]),
|
|
else:
|
|
assert_equal(addr_info['labels'], []),
|
|
|
|
def test_basic(self):
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
self.log.info("Test migration of a basic keys only wallet without balance")
|
|
basic0 = self.create_legacy_wallet("basic0")
|
|
|
|
addr = basic0.getnewaddress()
|
|
change = basic0.getrawchangeaddress()
|
|
|
|
old_addr_info = basic0.getaddressinfo(addr)
|
|
old_change_addr_info = basic0.getaddressinfo(change)
|
|
assert_equal(old_addr_info["ismine"], True)
|
|
assert_equal(old_addr_info["hdkeypath"], "m/0'/0'/0'")
|
|
assert_equal(old_change_addr_info["ismine"], True)
|
|
assert_equal(old_change_addr_info["hdkeypath"], "m/0'/1'/0'")
|
|
|
|
# Note: migration could take a while.
|
|
self.migrate_wallet(basic0)
|
|
|
|
# Verify created descriptors
|
|
assert_equal(basic0.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("basic0")
|
|
|
|
# The wallet should create the following descriptors:
|
|
# * BIP32 descriptors in the form of "0h/0h/*" and "0h/1h/*" (2 descriptors)
|
|
# * BIP44 descriptors in the form of "44h/1h/0h/0/*" and "44h/1h/0h/1/*" (2 descriptors)
|
|
# * BIP49 descriptors, P2SH(P2WPKH), in the form of "86h/1h/0h/0/*" and "86h/1h/0h/1/*" (2 descriptors)
|
|
# * BIP84 descriptors, P2WPKH, in the form of "84h/1h/0h/1/*" and "84h/1h/0h/1/*" (2 descriptors)
|
|
# * BIP86 descriptors, P2TR, in the form of "86h/1h/0h/0/*" and "86h/1h/0h/1/*" (2 descriptors)
|
|
# * A combo(PK) descriptor for the wallet master key.
|
|
# So, should have a total of 11 descriptors on it.
|
|
assert_equal(len(basic0.listdescriptors()["descriptors"]), 11)
|
|
|
|
# Compare addresses info
|
|
addr_info = basic0.getaddressinfo(addr)
|
|
change_addr_info = basic0.getaddressinfo(change)
|
|
self.assert_addr_info_equal(addr_info, old_addr_info)
|
|
self.assert_addr_info_equal(change_addr_info, old_change_addr_info)
|
|
|
|
addr_info = basic0.getaddressinfo(basic0.getnewaddress("", "bech32"))
|
|
assert_equal(addr_info["hdkeypath"], "m/84h/1h/0h/0/0")
|
|
|
|
self.log.info("Test migration of a basic keys only wallet with a balance")
|
|
basic1 = self.create_legacy_wallet("basic1")
|
|
|
|
for _ in range(0, 10):
|
|
default.sendtoaddress(basic1.getnewaddress(), 1)
|
|
|
|
self.generate(self.nodes[0], 1)
|
|
|
|
for _ in range(0, 5):
|
|
basic1.sendtoaddress(default.getnewaddress(), 0.5)
|
|
|
|
self.generate(self.nodes[0], 1)
|
|
bal = basic1.getbalance()
|
|
txs = basic1.listtransactions()
|
|
addr_gps = basic1.listaddressgroupings()
|
|
|
|
basic1_migrate = self.migrate_wallet(basic1)
|
|
assert_equal(basic1.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("basic1")
|
|
assert_equal(basic1.getbalance(), bal)
|
|
self.assert_list_txs_equal(basic1.listtransactions(), txs)
|
|
|
|
self.log.info("Test backup file can be successfully restored")
|
|
self.nodes[0].restorewallet("basic1_restored", basic1_migrate['backup_path'])
|
|
basic1_restored = self.nodes[0].get_wallet_rpc("basic1_restored")
|
|
basic1_restored_wi = basic1_restored.getwalletinfo()
|
|
assert_equal(basic1_restored_wi['balance'], bal)
|
|
assert_equal(basic1_restored.listaddressgroupings(), addr_gps)
|
|
self.assert_list_txs_equal(basic1_restored.listtransactions(), txs)
|
|
|
|
# restart node and verify that everything is still there
|
|
self.restart_node(0)
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
self.nodes[0].loadwallet("basic1")
|
|
basic1 = self.nodes[0].get_wallet_rpc("basic1")
|
|
assert_equal(basic1.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("basic1")
|
|
assert_equal(basic1.getbalance(), bal)
|
|
self.assert_list_txs_equal(basic1.listtransactions(), txs)
|
|
|
|
self.log.info("Test migration of a wallet with balance received on the seed")
|
|
basic2 = self.create_legacy_wallet("basic2")
|
|
basic2_seed = get_generate_key()
|
|
basic2.sethdseed(True, basic2_seed.privkey)
|
|
assert_equal(basic2.getbalance(), 0)
|
|
|
|
# Receive coins on different output types for the same seed
|
|
basic2_balance = 0
|
|
for addr in [basic2_seed.p2pkh_addr, basic2_seed.p2wpkh_addr, basic2_seed.p2sh_p2wpkh_addr]:
|
|
send_value = random.randint(1, 4)
|
|
default.sendtoaddress(addr, send_value)
|
|
basic2_balance += send_value
|
|
self.generate(self.nodes[0], 1)
|
|
assert_equal(basic2.getbalance(), basic2_balance)
|
|
basic2_txs = basic2.listtransactions()
|
|
|
|
# Now migrate and test that we still see have the same balance/transactions
|
|
self.migrate_wallet(basic2)
|
|
assert_equal(basic2.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("basic2")
|
|
assert_equal(basic2.getbalance(), basic2_balance)
|
|
self.assert_list_txs_equal(basic2.listtransactions(), basic2_txs)
|
|
|
|
# Now test migration on a descriptor wallet
|
|
self.log.info("Test \"nothing to migrate\" when the user tries to migrate a wallet with no legacy data")
|
|
assert_raises_rpc_error(-4, "Error: This wallet is already a descriptor wallet", basic2.migratewallet)
|
|
|
|
def test_multisig(self):
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
# Contrived case where all the multisig keys are in a single wallet
|
|
self.log.info("Test migration of a wallet with all keys for a multisig")
|
|
multisig0 = self.create_legacy_wallet("multisig0")
|
|
addr1 = multisig0.getnewaddress()
|
|
addr2 = multisig0.getnewaddress()
|
|
addr3 = multisig0.getnewaddress()
|
|
|
|
ms_info = multisig0.addmultisigaddress(2, [addr1, addr2, addr3])
|
|
|
|
self.migrate_wallet(multisig0)
|
|
assert_equal(multisig0.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("multisig0")
|
|
ms_addr_info = multisig0.getaddressinfo(ms_info["address"])
|
|
assert_equal(ms_addr_info["ismine"], True)
|
|
assert_equal(ms_addr_info["desc"], ms_info["descriptor"])
|
|
assert_equal("multisig0_watchonly" in self.nodes[0].listwallets(), False)
|
|
assert_equal("multisig0_solvables" in self.nodes[0].listwallets(), False)
|
|
|
|
pub1 = multisig0.getaddressinfo(addr1)["pubkey"]
|
|
pub2 = multisig0.getaddressinfo(addr2)["pubkey"]
|
|
|
|
# Some keys in multisig do not belong to this wallet
|
|
self.log.info("Test migration of a wallet that has some keys in a multisig")
|
|
multisig1 = self.create_legacy_wallet("multisig1")
|
|
ms_info = multisig1.addmultisigaddress(2, [multisig1.getnewaddress(), pub1, pub2])
|
|
ms_info2 = multisig1.addmultisigaddress(2, [multisig1.getnewaddress(), pub1, pub2])
|
|
|
|
addr1 = ms_info["address"]
|
|
addr2 = ms_info2["address"]
|
|
txid = default.sendtoaddress(addr1, 10)
|
|
multisig1.importaddress(addr1)
|
|
assert_equal(multisig1.getaddressinfo(addr1)["ismine"], False)
|
|
assert_equal(multisig1.getaddressinfo(addr1)["iswatchonly"], True)
|
|
assert_equal(multisig1.getaddressinfo(addr1)["solvable"], True)
|
|
self.generate(self.nodes[0], 1)
|
|
multisig1.gettransaction(txid)
|
|
assert_equal(multisig1.getbalances()["watchonly"]["trusted"], 10)
|
|
assert_equal(multisig1.getaddressinfo(addr2)["ismine"], False)
|
|
assert_equal(multisig1.getaddressinfo(addr2)["iswatchonly"], False)
|
|
assert_equal(multisig1.getaddressinfo(addr2)["solvable"], True)
|
|
|
|
# Migrating multisig1 should see the multisig is no longer part of multisig1
|
|
# A new wallet multisig1_watchonly is created which has the multisig address
|
|
# Transaction to multisig is in multisig1_watchonly and not multisig1
|
|
self.migrate_wallet(multisig1)
|
|
assert_equal(multisig1.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("multisig1")
|
|
assert_equal(multisig1.getaddressinfo(addr1)["ismine"], False)
|
|
assert_equal(multisig1.getaddressinfo(addr1)["iswatchonly"], False)
|
|
assert_equal(multisig1.getaddressinfo(addr1)["solvable"], False)
|
|
assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", multisig1.gettransaction, txid)
|
|
assert_equal(multisig1.getbalance(), 0)
|
|
assert_equal(multisig1.listtransactions(), [])
|
|
|
|
assert_equal("multisig1_watchonly" in self.nodes[0].listwallets(), True)
|
|
ms1_watchonly = self.nodes[0].get_wallet_rpc("multisig1_watchonly")
|
|
ms1_wallet_info = ms1_watchonly.getwalletinfo()
|
|
assert_equal(ms1_wallet_info['descriptors'], True)
|
|
assert_equal(ms1_wallet_info['private_keys_enabled'], False)
|
|
self.assert_is_sqlite("multisig1_watchonly")
|
|
assert_equal(ms1_watchonly.getaddressinfo(addr1)["ismine"], True)
|
|
assert_equal(ms1_watchonly.getaddressinfo(addr1)["solvable"], True)
|
|
# Because addr2 was not being watched, it isn't in multisig1_watchonly but rather multisig1_solvables
|
|
assert_equal(ms1_watchonly.getaddressinfo(addr2)["ismine"], False)
|
|
assert_equal(ms1_watchonly.getaddressinfo(addr2)["solvable"], False)
|
|
ms1_watchonly.gettransaction(txid)
|
|
assert_equal(ms1_watchonly.getbalance(), 10)
|
|
|
|
# Migrating multisig1 should see the second multisig is no longer part of multisig1
|
|
# A new wallet multisig1_solvables is created which has the second address
|
|
# This should have no transactions
|
|
assert_equal("multisig1_solvables" in self.nodes[0].listwallets(), True)
|
|
ms1_solvable = self.nodes[0].get_wallet_rpc("multisig1_solvables")
|
|
ms1_wallet_info = ms1_solvable.getwalletinfo()
|
|
assert_equal(ms1_wallet_info['descriptors'], True)
|
|
assert_equal(ms1_wallet_info['private_keys_enabled'], False)
|
|
self.assert_is_sqlite("multisig1_solvables")
|
|
assert_equal(ms1_solvable.getaddressinfo(addr1)["ismine"], False)
|
|
assert_equal(ms1_solvable.getaddressinfo(addr1)["solvable"], False)
|
|
assert_equal(ms1_solvable.getaddressinfo(addr2)["ismine"], True)
|
|
assert_equal(ms1_solvable.getaddressinfo(addr2)["solvable"], True)
|
|
assert_equal(ms1_solvable.getbalance(), 0)
|
|
assert_equal(ms1_solvable.listtransactions(), [])
|
|
|
|
|
|
def test_other_watchonly(self):
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
# Wallet with an imported address. Should be the same thing as the multisig test
|
|
self.log.info("Test migration of a wallet with watchonly imports")
|
|
imports0 = self.create_legacy_wallet("imports0")
|
|
|
|
# External address label
|
|
imports0.setlabel(default.getnewaddress(), "external")
|
|
|
|
# Normal non-watchonly tx
|
|
received_addr = imports0.getnewaddress()
|
|
imports0.setlabel(received_addr, "Receiving")
|
|
received_txid = default.sendtoaddress(received_addr, 10)
|
|
|
|
# Watchonly tx
|
|
import_addr = default.getnewaddress()
|
|
imports0.importaddress(import_addr)
|
|
imports0.setlabel(import_addr, "imported")
|
|
received_watchonly_txid = default.sendtoaddress(import_addr, 10)
|
|
|
|
# Received watchonly tx that is then spent
|
|
import_sent_addr = default.getnewaddress()
|
|
imports0.importaddress(import_sent_addr)
|
|
received_sent_watchonly_utxo = self.create_outpoints(node=default, outputs=[{import_sent_addr: 10}])[0]
|
|
|
|
send = default.sendall(recipients=[default.getnewaddress()], inputs=[received_sent_watchonly_utxo])
|
|
sent_watchonly_txid = send["txid"]
|
|
|
|
# Tx that has both a watchonly and spendable output
|
|
watchonly_spendable_txid = default.send(outputs=[{received_addr: 1}, {import_addr:1}])["txid"]
|
|
|
|
self.generate(self.nodes[0], 2)
|
|
received_watchonly_tx_info = imports0.gettransaction(received_watchonly_txid, True)
|
|
received_sent_watchonly_tx_info = imports0.gettransaction(received_sent_watchonly_utxo["txid"], True)
|
|
|
|
balances = imports0.getbalances()
|
|
spendable_bal = balances["mine"]["trusted"]
|
|
watchonly_bal = balances["watchonly"]["trusted"]
|
|
assert_equal(len(imports0.listtransactions(include_watchonly=True)), 6)
|
|
|
|
# Mock time forward a bit so we can check that tx metadata is preserved
|
|
self.nodes[0].setmocktime(int(time.time()) + 100)
|
|
|
|
# Migrate
|
|
self.migrate_wallet(imports0)
|
|
assert_equal(imports0.getwalletinfo()["descriptors"], True)
|
|
self.assert_is_sqlite("imports0")
|
|
assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", imports0.gettransaction, received_watchonly_txid)
|
|
assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", imports0.gettransaction, received_sent_watchonly_utxo['txid'])
|
|
assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", imports0.gettransaction, sent_watchonly_txid)
|
|
assert_equal(len(imports0.listtransactions(include_watchonly=True)), 2)
|
|
imports0.gettransaction(received_txid)
|
|
imports0.gettransaction(watchonly_spendable_txid)
|
|
assert_equal(imports0.getbalance(), spendable_bal)
|
|
|
|
assert_equal("imports0_watchonly" in self.nodes[0].listwallets(), True)
|
|
watchonly = self.nodes[0].get_wallet_rpc("imports0_watchonly")
|
|
watchonly_info = watchonly.getwalletinfo()
|
|
assert_equal(watchonly_info["descriptors"], True)
|
|
self.assert_is_sqlite("imports0_watchonly")
|
|
assert_equal(watchonly_info["private_keys_enabled"], False)
|
|
received_migrated_watchonly_tx_info = watchonly.gettransaction(received_watchonly_txid)
|
|
assert_equal(received_watchonly_tx_info["time"], received_migrated_watchonly_tx_info["time"])
|
|
assert_equal(received_watchonly_tx_info["timereceived"], received_migrated_watchonly_tx_info["timereceived"])
|
|
received_sent_migrated_watchonly_tx_info = watchonly.gettransaction(received_sent_watchonly_utxo["txid"])
|
|
assert_equal(received_sent_watchonly_tx_info["time"], received_sent_migrated_watchonly_tx_info["time"])
|
|
assert_equal(received_sent_watchonly_tx_info["timereceived"], received_sent_migrated_watchonly_tx_info["timereceived"])
|
|
watchonly.gettransaction(sent_watchonly_txid)
|
|
watchonly.gettransaction(watchonly_spendable_txid)
|
|
assert_equal(watchonly.getbalance(), watchonly_bal)
|
|
assert_raises_rpc_error(-5, "Invalid or non-wallet transaction id", watchonly.gettransaction, received_txid)
|
|
assert_equal(len(watchonly.listtransactions(include_watchonly=True)), 4)
|
|
|
|
# Check that labels were migrated and persisted to watchonly wallet
|
|
self.nodes[0].unloadwallet("imports0_watchonly")
|
|
self.nodes[0].loadwallet("imports0_watchonly")
|
|
labels = watchonly.listlabels()
|
|
assert "external" in labels
|
|
assert "imported" in labels
|
|
|
|
def test_no_privkeys(self):
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
# Migrating an actual watchonly wallet should not create a new watchonly wallet
|
|
self.log.info("Test migration of a pure watchonly wallet")
|
|
watchonly0 = self.create_legacy_wallet("watchonly0", disable_private_keys=True)
|
|
|
|
addr = default.getnewaddress()
|
|
desc = default.getaddressinfo(addr)["desc"]
|
|
res = watchonly0.importmulti([
|
|
{
|
|
"desc": desc,
|
|
"watchonly": True,
|
|
"timestamp": "now",
|
|
}])
|
|
assert_equal(res[0]['success'], True)
|
|
default.sendtoaddress(addr, 10)
|
|
self.generate(self.nodes[0], 1)
|
|
|
|
self.migrate_wallet(watchonly0)
|
|
assert_equal("watchonly0_watchonly" in self.nodes[0].listwallets(), False)
|
|
info = watchonly0.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["private_keys_enabled"], False)
|
|
self.assert_is_sqlite("watchonly0")
|
|
|
|
# Migrating a wallet with pubkeys added to the keypool
|
|
self.log.info("Test migration of a pure watchonly wallet with pubkeys in keypool")
|
|
watchonly1 = self.create_legacy_wallet("watchonly1", disable_private_keys=True)
|
|
|
|
addr1 = default.getnewaddress(address_type="bech32")
|
|
addr2 = default.getnewaddress(address_type="bech32")
|
|
desc1 = default.getaddressinfo(addr1)["desc"]
|
|
desc2 = default.getaddressinfo(addr2)["desc"]
|
|
res = watchonly1.importmulti([
|
|
{
|
|
"desc": desc1,
|
|
"keypool": True,
|
|
"timestamp": "now",
|
|
},
|
|
{
|
|
"desc": desc2,
|
|
"keypool": True,
|
|
"timestamp": "now",
|
|
}
|
|
])
|
|
assert_equal(res[0]["success"], True)
|
|
assert_equal(res[1]["success"], True)
|
|
# Before migrating, we can fetch addr1 from the keypool
|
|
assert_equal(watchonly1.getnewaddress(address_type="bech32"), addr1)
|
|
|
|
self.migrate_wallet(watchonly1)
|
|
info = watchonly1.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["private_keys_enabled"], False)
|
|
self.assert_is_sqlite("watchonly1")
|
|
# After migrating, the "keypool" is empty
|
|
assert_raises_rpc_error(-4, "Error: This wallet has no available keys", watchonly1.getnewaddress)
|
|
|
|
def test_pk_coinbases(self):
|
|
self.log.info("Test migration of a wallet using old pk() coinbases")
|
|
wallet = self.create_legacy_wallet("pkcb")
|
|
|
|
addr = wallet.getnewaddress()
|
|
addr_info = wallet.getaddressinfo(addr)
|
|
desc = descsum_create("pk(" + addr_info["pubkey"] + ")")
|
|
|
|
self.nodes[0].generatetodescriptor(1, desc, invalid_call=False)
|
|
|
|
bals = wallet.getbalances()
|
|
|
|
self.migrate_wallet(wallet)
|
|
|
|
assert_equal(bals, wallet.getbalances())
|
|
|
|
def test_encrypted(self):
|
|
self.log.info("Test migration of an encrypted wallet")
|
|
wallet = self.create_legacy_wallet("encrypted")
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
wallet.encryptwallet("pass")
|
|
addr = wallet.getnewaddress()
|
|
txid = default.sendtoaddress(addr, 1)
|
|
self.generate(self.nodes[0], 1)
|
|
bals = wallet.getbalances()
|
|
|
|
assert_raises_rpc_error(-4, "Error: Wallet decryption failed, the wallet passphrase was not provided or was incorrect", wallet.migratewallet)
|
|
assert_raises_rpc_error(-4, "Error: Wallet decryption failed, the wallet passphrase was not provided or was incorrect", wallet.migratewallet, None, "badpass")
|
|
assert_raises_rpc_error(-4, "The passphrase contains a null character", wallet.migratewallet, None, "pass\0with\0null")
|
|
|
|
self.migrate_wallet(wallet, passphrase="pass")
|
|
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["format"], "sqlite")
|
|
assert_equal(info["unlocked_until"], 0)
|
|
wallet.gettransaction(txid)
|
|
|
|
assert_equal(bals, wallet.getbalances())
|
|
|
|
def test_unloaded(self):
|
|
self.log.info("Test migration of a wallet that isn't loaded")
|
|
wallet = self.create_legacy_wallet("notloaded")
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
addr = wallet.getnewaddress()
|
|
txid = default.sendtoaddress(addr, 1)
|
|
self.generate(self.nodes[0], 1)
|
|
bals = wallet.getbalances()
|
|
|
|
wallet.unloadwallet()
|
|
|
|
assert_raises_rpc_error(-8, "RPC endpoint wallet and wallet_name parameter specify different wallets", wallet.migratewallet, "someotherwallet")
|
|
assert_raises_rpc_error(-8, "Either RPC endpoint wallet or wallet_name parameter must be provided", self.nodes[0].migratewallet)
|
|
self.nodes[0].migratewallet("notloaded")
|
|
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["format"], "sqlite")
|
|
wallet.gettransaction(txid)
|
|
|
|
assert_equal(bals, wallet.getbalances())
|
|
|
|
def test_unloaded_by_path(self):
|
|
self.log.info("Test migration of a wallet that isn't loaded, specified by path")
|
|
wallet = self.create_legacy_wallet("notloaded2")
|
|
default = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
addr = wallet.getnewaddress()
|
|
txid = default.sendtoaddress(addr, 1)
|
|
self.generate(self.nodes[0], 1)
|
|
bals = wallet.getbalances()
|
|
|
|
wallet.unloadwallet()
|
|
|
|
wallet_file_path = self.nodes[0].wallets_path / "notloaded2"
|
|
self.nodes[0].migratewallet(wallet_file_path)
|
|
|
|
# Because we gave the name by full path, the loaded wallet's name is that path too.
|
|
wallet = self.nodes[0].get_wallet_rpc(str(wallet_file_path))
|
|
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["format"], "sqlite")
|
|
wallet.gettransaction(txid)
|
|
|
|
assert_equal(bals, wallet.getbalances())
|
|
|
|
def test_default_wallet(self):
|
|
self.log.info("Test migration of the wallet named as the empty string")
|
|
wallet = self.create_legacy_wallet("")
|
|
|
|
# Set time to verify backup existence later
|
|
curr_time = int(time.time())
|
|
wallet.setmocktime(curr_time)
|
|
|
|
res = self.migrate_wallet(wallet)
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["format"], "sqlite")
|
|
|
|
# Check backup existence and its non-empty wallet filename
|
|
backup_path = self.nodes[0].wallets_path / f'default_wallet_{curr_time}.legacy.bak'
|
|
assert backup_path.exists()
|
|
assert_equal(str(backup_path), res['backup_path'])
|
|
|
|
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()
|
|
|
|
wallets_dir = self.nodes[0].wallets_path
|
|
wallet_path = wallets_dir / "plainfile"
|
|
wallet_dat_path = wallet_path / "wallet.dat"
|
|
shutil.copyfile(wallet_dat_path, wallets_dir / "plainfile.bak")
|
|
shutil.rmtree(wallet_path)
|
|
shutil.move(wallets_dir / "plainfile.bak", wallet_path)
|
|
|
|
self.nodes[0].loadwallet("plainfile")
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], False)
|
|
assert_equal(info["format"], "bdb")
|
|
|
|
self.migrate_wallet(wallet)
|
|
info = wallet.getwalletinfo()
|
|
assert_equal(info["descriptors"], True)
|
|
assert_equal(info["format"], "sqlite")
|
|
|
|
assert wallet_path.is_dir()
|
|
assert wallet_dat_path.is_file()
|
|
|
|
def test_addressbook(self):
|
|
df_wallet = self.nodes[0].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()
|
|
multi_addr2 = wallet.getnewaddress()
|
|
multi_addr3 = df_wallet.getnewaddress()
|
|
wallet.importpubkey(df_wallet.getaddressinfo(multi_addr3)["pubkey"])
|
|
ms_addr_info = wallet.addmultisigaddress(2, [multi_addr1, multi_addr2, multi_addr3])
|
|
|
|
self.generate(self.nodes[0], 1)
|
|
|
|
# Test vectors
|
|
addr_external = {
|
|
"addr": df_wallet.getnewaddress(),
|
|
"is_mine": False,
|
|
"is_change": False,
|
|
"label": ""
|
|
}
|
|
addr_external_with_label = {
|
|
"addr": df_wallet.getnewaddress(),
|
|
"is_mine": False,
|
|
"is_change": False,
|
|
"label": "external"
|
|
}
|
|
addr_internal = {
|
|
"addr": wallet.getnewaddress(),
|
|
"is_mine": True,
|
|
"is_change": False,
|
|
"label": ""
|
|
}
|
|
addr_internal_with_label = {
|
|
"addr": wallet.getnewaddress(),
|
|
"is_mine": True,
|
|
"is_change": False,
|
|
"label": "internal"
|
|
}
|
|
change_address = {
|
|
"addr": wallet.getrawchangeaddress(),
|
|
"is_mine": True,
|
|
"is_change": True,
|
|
"label": None
|
|
}
|
|
watch_only_addr = {
|
|
"addr": watch_addr,
|
|
"is_mine": False,
|
|
"is_change": False,
|
|
"label": "imported"
|
|
}
|
|
ms_addr = {
|
|
"addr": ms_addr_info['address'],
|
|
"is_mine": False,
|
|
"is_change": False,
|
|
"label": "multisig"
|
|
}
|
|
|
|
# To store the change address in the addressbook need to send coins to it
|
|
wallet.send(outputs=[{wallet.getnewaddress(): 2}], options={"change_address": change_address['addr']})
|
|
self.generate(self.nodes[0], 1)
|
|
|
|
# Util wrapper func for 'addr_info'
|
|
def check(info, node):
|
|
self.check_address(node, info['addr'], info['is_mine'], info['is_change'], info["label"])
|
|
|
|
# Pre-migration: set label and perform initial checks
|
|
for addr_info in [addr_external, addr_external_with_label, addr_internal, addr_internal_with_label, change_address, watch_only_addr, ms_addr]:
|
|
if not addr_info['is_change']:
|
|
wallet.setlabel(addr_info['addr'], addr_info["label"])
|
|
check(addr_info, wallet)
|
|
|
|
# Migrate wallet
|
|
info_migration = self.migrate_wallet(wallet)
|
|
wallet_wo = self.nodes[0].get_wallet_rpc(info_migration["watchonly_name"])
|
|
wallet_solvables = self.nodes[0].get_wallet_rpc(info_migration["solvables_name"])
|
|
|
|
#########################
|
|
# Post migration checks #
|
|
#########################
|
|
|
|
# First check the main wallet
|
|
for addr_info in [addr_external, addr_external_with_label, addr_internal, addr_internal_with_label, change_address, ms_addr]:
|
|
check(addr_info, wallet)
|
|
|
|
# Watch-only wallet will contain the watch-only entry (with 'is_mine=True') and all external addresses ('send')
|
|
self.check_address(wallet_wo, watch_only_addr['addr'], is_mine=True, is_change=watch_only_addr['is_change'], label=watch_only_addr["label"])
|
|
for addr_info in [addr_external, addr_external_with_label, ms_addr]:
|
|
check(addr_info, wallet_wo)
|
|
|
|
# Solvables wallet will contain the multisig entry (with 'is_mine=True') and all external addresses ('send')
|
|
self.check_address(wallet_solvables, ms_addr['addr'], is_mine=True, is_change=ms_addr['is_change'], label=ms_addr["label"])
|
|
for addr_info in [addr_external, addr_external_with_label]:
|
|
check(addr_info, wallet_solvables)
|
|
|
|
########################################################################################
|
|
# Now restart migrated wallets and verify that the addressbook entries are still there #
|
|
########################################################################################
|
|
|
|
# First the main wallet
|
|
self.nodes[0].unloadwallet("legacy_addrbook")
|
|
self.nodes[0].loadwallet("legacy_addrbook")
|
|
for addr_info in [addr_external, addr_external_with_label, addr_internal, addr_internal_with_label, change_address, ms_addr]:
|
|
check(addr_info, wallet)
|
|
|
|
# Watch-only wallet
|
|
self.nodes[0].unloadwallet(info_migration["watchonly_name"])
|
|
self.nodes[0].loadwallet(info_migration["watchonly_name"])
|
|
self.check_address(wallet_wo, watch_only_addr['addr'], is_mine=True, is_change=watch_only_addr['is_change'], label=watch_only_addr["label"])
|
|
for addr_info in [addr_external, addr_external_with_label, ms_addr]:
|
|
check(addr_info, wallet_wo)
|
|
|
|
# Solvables wallet
|
|
self.nodes[0].unloadwallet(info_migration["solvables_name"])
|
|
self.nodes[0].loadwallet(info_migration["solvables_name"])
|
|
self.check_address(wallet_solvables, ms_addr['addr'], is_mine=True, is_change=ms_addr['is_change'], label=ms_addr["label"])
|
|
for addr_info in [addr_external, addr_external_with_label]:
|
|
check(addr_info, wallet_solvables)
|
|
|
|
def test_migrate_raw_p2sh(self):
|
|
self.log.info("Test migration of watch-only raw p2sh script")
|
|
df_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
wallet = self.create_legacy_wallet("raw_p2sh")
|
|
|
|
def send_to_script(script, amount):
|
|
tx = CTransaction()
|
|
tx.vout.append(CTxOut(nValue=amount*COIN, scriptPubKey=script))
|
|
|
|
hex_tx = df_wallet.fundrawtransaction(tx.serialize().hex())['hex']
|
|
signed_tx = df_wallet.signrawtransactionwithwallet(hex_tx)
|
|
df_wallet.sendrawtransaction(signed_tx['hex'])
|
|
self.generate(self.nodes[0], 1)
|
|
|
|
# Craft sh(pkh(key)) script and send coins to it
|
|
pubkey = df_wallet.getaddressinfo(df_wallet.getnewaddress())["pubkey"]
|
|
script_pkh = key_to_p2pkh_script(pubkey)
|
|
script_sh_pkh = script_to_p2sh_script(script_pkh)
|
|
send_to_script(script=script_sh_pkh, amount=2)
|
|
|
|
# Import script and check balance
|
|
wallet.rpc.importaddress(address=script_pkh.hex(), label="raw_spk", rescan=True, p2sh=True)
|
|
assert_equal(wallet.getbalances()['watchonly']['trusted'], 2)
|
|
|
|
# Craft wsh(pkh(key)) and send coins to it
|
|
pubkey = df_wallet.getaddressinfo(df_wallet.getnewaddress())["pubkey"]
|
|
script_wsh_pkh = script_to_p2wsh_script(key_to_p2pkh_script(pubkey))
|
|
send_to_script(script=script_wsh_pkh, amount=3)
|
|
|
|
# Import script and check balance
|
|
wallet.rpc.importaddress(address=script_wsh_pkh.hex(), label="raw_spk2", rescan=True, p2sh=False)
|
|
assert_equal(wallet.getbalances()['watchonly']['trusted'], 5)
|
|
|
|
# Import sh(pkh()) script, by using importaddress(), with the p2sh flag enabled.
|
|
# This will wrap the script under another sh level, which is invalid!, and store it inside the wallet.
|
|
# The migration process must skip the invalid scripts and the addressbook records linked to them.
|
|
# They are not being watched by the current wallet, nor should be watched by the migrated one.
|
|
label_sh_pkh = "raw_sh_pkh"
|
|
script_pkh = key_to_p2pkh_script(df_wallet.getaddressinfo(df_wallet.getnewaddress())["pubkey"])
|
|
script_sh_pkh = script_to_p2sh_script(script_pkh)
|
|
addy_script_sh_pkh = script_to_p2sh(script_pkh) # valid script address
|
|
addy_script_double_sh_pkh = script_to_p2sh(script_sh_pkh) # invalid script address
|
|
|
|
# Note: 'importaddress()' will add two scripts, a valid one sh(pkh()) and an invalid one 'sh(sh(pkh()))'.
|
|
# Both of them will be stored with the same addressbook label. And only the latter one should
|
|
# be discarded during migration. The first one must be migrated.
|
|
wallet.rpc.importaddress(address=script_sh_pkh.hex(), label=label_sh_pkh, rescan=False, p2sh=True)
|
|
|
|
# Migrate wallet and re-check balance
|
|
info_migration = self.migrate_wallet(wallet)
|
|
wallet_wo = self.nodes[0].get_wallet_rpc(info_migration["watchonly_name"])
|
|
|
|
# Watch-only balance is under "mine".
|
|
assert_equal(wallet_wo.getbalances()['mine']['trusted'], 5)
|
|
# The watch-only scripts are no longer part of the main wallet
|
|
assert_equal(wallet.getbalances()['mine']['trusted'], 0)
|
|
|
|
# The invalid sh(sh(pk())) script label must not be part of the main wallet anymore
|
|
assert label_sh_pkh not in wallet.listlabels()
|
|
# But, the standard sh(pkh()) script should be part of the watch-only wallet.
|
|
addrs_by_label = wallet_wo.getaddressesbylabel(label_sh_pkh)
|
|
assert addy_script_sh_pkh in addrs_by_label
|
|
assert addy_script_double_sh_pkh not in addrs_by_label
|
|
|
|
# Also, the watch-only wallet should have the descriptor for the standard sh(pkh())
|
|
desc = descsum_create(f"addr({addy_script_sh_pkh})")
|
|
assert next(it['desc'] for it in wallet_wo.listdescriptors()['descriptors'] if it['desc'] == desc)
|
|
# And doesn't have a descriptor for the invalid one
|
|
desc_invalid = descsum_create(f"addr({addy_script_double_sh_pkh})")
|
|
assert_equal(next((it['desc'] for it in wallet_wo.listdescriptors()['descriptors'] if it['desc'] == desc_invalid), None), None)
|
|
|
|
# Just in case, also verify wallet restart
|
|
self.nodes[0].unloadwallet(info_migration["watchonly_name"])
|
|
self.nodes[0].loadwallet(info_migration["watchonly_name"])
|
|
assert_equal(wallet_wo.getbalances()['mine']['trusted'], 5)
|
|
|
|
def test_conflict_txs(self):
|
|
self.log.info("Test migration when wallet contains conflicting transactions")
|
|
def_wallet = self.nodes[0].get_wallet_rpc(self.default_wallet_name)
|
|
|
|
wallet = self.create_legacy_wallet("conflicts")
|
|
def_wallet.sendtoaddress(wallet.getnewaddress(), 10)
|
|
self.generate(self.nodes[0], 1)
|
|
|
|
# parent tx
|
|
parent_txid = wallet.sendtoaddress(wallet.getnewaddress(), 9)
|
|
parent_txid_bytes = bytes.fromhex(parent_txid)[::-1]
|
|
conflict_utxo = wallet.gettransaction(txid=parent_txid, verbose=True)["decoded"]["vin"][0]
|
|
|
|
# The specific assertion in MarkConflicted being tested requires that the parent tx is already loaded
|
|
# by the time the child tx is loaded. Since transactions end up being loaded in txid order due to how both
|
|
# and sqlite store things, we can just grind the child tx until it has a txid that is greater than the parent's.
|
|
locktime = 500000000 # Use locktime as nonce, starting at unix timestamp minimum
|
|
addr = wallet.getnewaddress()
|
|
while True:
|
|
child_send_res = wallet.send(outputs=[{addr: 8}], add_to_wallet=False, locktime=locktime)
|
|
child_txid = child_send_res["txid"]
|
|
child_txid_bytes = bytes.fromhex(child_txid)[::-1]
|
|
if (child_txid_bytes > parent_txid_bytes):
|
|
wallet.sendrawtransaction(child_send_res["hex"])
|
|
break
|
|
locktime += 1
|
|
|
|
# conflict with parent
|
|
conflict_unsigned = self.nodes[0].createrawtransaction(inputs=[conflict_utxo], outputs=[{wallet.getnewaddress(): 9.9999}])
|
|
conflict_signed = wallet.signrawtransactionwithwallet(conflict_unsigned)["hex"]
|
|
conflict_txid = self.nodes[0].sendrawtransaction(conflict_signed)
|
|
self.generate(self.nodes[0], 1)
|
|
assert_equal(wallet.gettransaction(txid=parent_txid)["confirmations"], -1)
|
|
assert_equal(wallet.gettransaction(txid=child_txid)["confirmations"], -1)
|
|
assert_equal(wallet.gettransaction(txid=conflict_txid)["confirmations"], 1)
|
|
|
|
self.migrate_wallet(wallet)
|
|
assert_equal(wallet.gettransaction(txid=parent_txid)["confirmations"], -1)
|
|
assert_equal(wallet.gettransaction(txid=child_txid)["confirmations"], -1)
|
|
assert_equal(wallet.gettransaction(txid=conflict_txid)["confirmations"], 1)
|
|
|
|
wallet.unloadwallet()
|
|
|
|
def test_hybrid_pubkey(self):
|
|
self.log.info("Test migration when wallet contains a hybrid pubkey")
|
|
|
|
wallet = self.create_legacy_wallet("hybrid_keys")
|
|
|
|
# Get the hybrid pubkey for one of the keys in the wallet
|
|
normal_pubkey = wallet.getaddressinfo(wallet.getnewaddress())["pubkey"]
|
|
first_byte = bytes.fromhex(normal_pubkey)[0] + 4 # Get the hybrid pubkey first byte
|
|
parsed_pubkey = ECPubKey()
|
|
parsed_pubkey.set(bytes.fromhex(normal_pubkey))
|
|
parsed_pubkey.compressed = False
|
|
hybrid_pubkey_bytes = bytearray(parsed_pubkey.get_bytes())
|
|
hybrid_pubkey_bytes[0] = first_byte # Make it hybrid
|
|
hybrid_pubkey = hybrid_pubkey_bytes.hex()
|
|
|
|
# Import the hybrid pubkey
|
|
wallet.importpubkey(hybrid_pubkey)
|
|
p2pkh_addr = key_to_p2pkh(hybrid_pubkey)
|
|
p2pkh_addr_info = wallet.getaddressinfo(p2pkh_addr)
|
|
assert_equal(p2pkh_addr_info["iswatchonly"], True)
|
|
assert_equal(p2pkh_addr_info["ismine"], False) # Things involving hybrid pubkeys are not spendable
|
|
|
|
# Also import the p2wpkh for the pubkey to make sure we don't migrate it
|
|
p2wpkh_addr = key_to_p2wpkh(hybrid_pubkey)
|
|
wallet.importaddress(p2wpkh_addr)
|
|
|
|
migrate_info = self.migrate_wallet(wallet)
|
|
|
|
# Both addresses should only appear in the watchonly wallet
|
|
p2pkh_addr_info = wallet.getaddressinfo(p2pkh_addr)
|
|
assert_equal(p2pkh_addr_info["iswatchonly"], False)
|
|
assert_equal(p2pkh_addr_info["ismine"], False)
|
|
p2wpkh_addr_info = wallet.getaddressinfo(p2wpkh_addr)
|
|
assert_equal(p2wpkh_addr_info["iswatchonly"], False)
|
|
assert_equal(p2wpkh_addr_info["ismine"], False)
|
|
|
|
watchonly_wallet = self.nodes[0].get_wallet_rpc(migrate_info["watchonly_name"])
|
|
watchonly_p2pkh_addr_info = watchonly_wallet.getaddressinfo(p2pkh_addr)
|
|
assert_equal(watchonly_p2pkh_addr_info["iswatchonly"], False)
|
|
assert_equal(watchonly_p2pkh_addr_info["ismine"], True)
|
|
watchonly_p2wpkh_addr_info = watchonly_wallet.getaddressinfo(p2wpkh_addr)
|
|
assert_equal(watchonly_p2wpkh_addr_info["iswatchonly"], False)
|
|
assert_equal(watchonly_p2wpkh_addr_info["ismine"], True)
|
|
|
|
# There should only be raw or addr descriptors
|
|
for desc in watchonly_wallet.listdescriptors()["descriptors"]:
|
|
if desc["desc"].startswith("raw(") or desc["desc"].startswith("addr("):
|
|
continue
|
|
assert False, "Hybrid pubkey watchonly wallet has more than just raw() and addr()"
|
|
|
|
wallet.unloadwallet()
|
|
|
|
def test_failed_migration_cleanup(self):
|
|
self.log.info("Test that a failed migration is cleaned up")
|
|
wallet = self.create_legacy_wallet("failed")
|
|
|
|
# Make a copy of the wallet with the solvables wallet name so that we are unable
|
|
# to create the solvables wallet when migrating, thus failing to migrate
|
|
wallet.unloadwallet()
|
|
solvables_path = self.nodes[0].wallets_path / "failed_solvables"
|
|
shutil.copytree(self.nodes[0].wallets_path / "failed", solvables_path)
|
|
original_shasum = sha256sum_file(solvables_path / "wallet.dat")
|
|
|
|
self.nodes[0].loadwallet("failed")
|
|
|
|
# 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)
|
|
|
|
assert_raises_rpc_error(-4, "Failed to create database", wallet.migratewallet)
|
|
|
|
assert "failed" in self.nodes[0].listwallets()
|
|
assert "failed_watchonly" not in self.nodes[0].listwallets()
|
|
assert "failed_solvables" not in self.nodes[0].listwallets()
|
|
|
|
assert not (self.nodes[0].wallets_path / "failed_watchonly").exists()
|
|
# Since the file in failed_solvables is one that we put there, migration shouldn't touch it
|
|
assert solvables_path.exists()
|
|
new_shasum = sha256sum_file(solvables_path / "wallet.dat")
|
|
assert_equal(original_shasum, new_shasum)
|
|
|
|
wallet.unloadwallet()
|
|
# Check the wallet we tried to migrate is still BDB
|
|
with open(self.nodes[0].wallets_path / "failed" / "wallet.dat", "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.migratewallet()
|
|
assert_equal(wallet.getwalletinfo()["blank"], True)
|
|
assert_equal(wallet.getwalletinfo()["descriptors"], True)
|
|
|
|
def test_avoidreuse(self):
|
|
self.log.info("Test that avoidreuse persists after migration")
|
|
def_wallet = self.nodes[0].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
|
|
reused_addr = wallet.getnewaddress()
|
|
def_wallet.sendtoaddress(reused_addr, 2)
|
|
|
|
self.generate(self.nodes[0], 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.nodes[0], 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.nodes[0], 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
|
|
migrate_res = wallet.migratewallet()
|
|
watchonly_wallet = self.nodes[0].get_wallet_rpc(migrate_res["watchonly_name"])
|
|
|
|
# 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)
|
|
|
|
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 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()
|
|
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()
|
|
self.test_avoidreuse()
|
|
self.test_preserve_tx_extra_info()
|
|
self.test_blank()
|
|
|
|
if __name__ == '__main__':
|
|
WalletMigrationTest().main()
|