test/functional/wallet_ancient_migration.py - master - githubmirror/bitcoin

test/functional/wallet_ancient_migration.py
229 строк9 KB
w0xlt
test: cover v0.14.3 wallet migration
05 авг 2026, 22:50
05 авг 2026, 22:50ea59f17
#!/usr/bin/env python3
# Copyright (c) 2026-present 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 wallet migration from ancient versions (v0.14.3) to descriptor wallets.

Previous releases are required by this test, see test/README.md.
"""
from decimal import Decimal
from pathlib import Path
import shutil

from test_framework.blocktools import COINBASE_MATURITY
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
    assert_equal,
    assert_greater_than,
    assert_is_hash_string,
    assert_raises_rpc_error,
    dumb_sync_blocks,
)

# Wallet version constants from ancient Bitcoin Core (pre-0.15)
FEATURE_COMPRPUBKEY = 60000  # Non-HD wallet version
FEATURE_HD = 130000          # Single-chain HD wallet version (pre-v0.15)
WALLET_PASSPHRASE = "test_passphrase"


def get_vout_addresses(decoded_tx):
    for vout in decoded_tx["vout"]:
        script_pub_key = vout["scriptPubKey"]
        if "address" in script_pub_key:
            yield script_pub_key["address"]
        else:
            yield from script_pub_key.get("addresses", [])


def normalize_listtransactions(txs):
    """Return cross-version stable fields from listtransactions output."""
    return sorted(
        (tx["txid"], tx["amount"], tx["category"], tx["confirmations"])
        for tx in txs
    )


class WalletAncientMigrationTest(BitcoinTestFramework):
    def set_test_params(self):
        self.setup_clean_chain = True
        self.num_nodes = 8

    def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()
        self.skip_if_no_previous_releases()

    def setup_network(self):
        # Use dedicated old/new node pairs for each migration scenario to avoid
        # resetting node datadirs between sub-tests.
        self.add_nodes(self.num_nodes, versions=[140300, None] * 4)

    def run_migration_test(
        self,
        wallet_type,
        old_node_idx,
        new_node_idx,
        extra_args_old,
        expected_version,
        expect_hd,
        passphrase=None,
    ):
        """Test migration of a v0.14.3 wallet to descriptor wallet."""
        old_node = self.nodes[old_node_idx]
        new_node = self.nodes[new_node_idx]

        self.log.info(f"Testing {wallet_type} wallet migration")
        self.start_node(old_node_idx, extra_args=extra_args_old)

        # Create addresses and verify wallet version
        old_addresses = [old_node.getnewaddress() for _ in range(4)]
        unfunded_address = old_node.getnewaddress()
        old_wallet_info = old_node.getwalletinfo()
        assert_equal(old_wallet_info['walletversion'], expected_version)
        # v0.14.3 predates HD split keypool (v0.15+).
        assert_equal('keypoolsize_hd_internal' in old_wallet_info, False)
        assert_equal('hdmasterkeyid' in old_wallet_info, expect_hd)

        # Generate blocks and create transaction history
        self.generatetoaddress(old_node, COINBASE_MATURITY + 1, old_addresses[0], sync_fun=self.no_op)
        send_txs = []
        for i, amount in enumerate([0.1, 0.2, 0.3], start=1):
            send_txs.append((old_node.sendtoaddress(old_addresses[i], amount), old_addresses[i]))
        self.generatetoaddress(old_node, 1, old_addresses[0], sync_fun=self.no_op)

        old_balance = old_node.getbalance()
        old_txs = old_node.listtransactions("*", 1000)
        old_wallet_state = old_node.getwalletinfo()
        assert_equal(old_balance, Decimal("99.99986440"))
        assert_equal(len(old_txs), 108)

        # Collect all addresses including change; listtransactions omits change
        # addresses for self-sends, so decode the sent transactions explicitly.
        change_addresses = set()
        for txid, destination in send_txs:
            decoded_tx = old_node.decoderawtransaction(old_node.gettransaction(txid)["hex"])
            for address in get_vout_addresses(decoded_tx):
                if address != destination:
                    change_addresses.add(address)

        all_addresses = {tx['address'] for tx in old_txs if 'address' in tx} | change_addresses
        assert_equal(unfunded_address in all_addresses, False)
        assert_greater_than(len(change_addresses), 0)
        assert_greater_than(len(all_addresses), 0)

        # Sync blocks to modern node via RPC (avoids filesystem copy and reindex)
        self.start_node(new_node_idx)
        dumb_sync_blocks(src=old_node, dst=new_node)

        if passphrase is None:
            self.stop_node(old_node_idx)
        else:
            # v0.14.3 shuts down after encrypting the wallet.
            old_node.encryptwallet(passphrase)
            old_node.wait_until_stopped()

        # Copy and migrate wallet
        self.log.info("Migrating wallet to descriptor format")
        old_wallet_path = old_node.chain_path / "wallet.dat"
        migrated_wallet_dir = new_node.wallets_path / "migrated_wallet"
        migrated_wallet_dir.mkdir(parents=True)
        shutil.copy2(old_wallet_path, migrated_wallet_dir / "wallet.dat")

        with new_node.assert_debug_log(expected_msgs=[], unexpected_msgs=["Rescanning"]):
            migration_result = new_node.migratewallet("migrated_wallet", passphrase=passphrase)
        assert_equal(migration_result['wallet_name'], 'migrated_wallet')
        assert_equal('watchonly_name' in migration_result, False)
        assert_equal('solvables_name' in migration_result, False)
        assert Path(migration_result['backup_path']).is_file()
        new_wallet = new_node.get_wallet_rpc(migration_result['wallet_name'])

        # Verify migration results
        self.log.info("Verifying migration")
        new_wallet_info = new_wallet.getwalletinfo()
        assert_equal(new_wallet_info['format'], 'sqlite')
        assert_equal(new_wallet_info['descriptors'], True)

        # v0.14.3 HD wallets use a single external chain. Encrypting an HD
        # wallet rotates its seed, leaving the original chain inactive.
        single_chain_hd_descriptors = [
            descriptor for descriptor in new_wallet.listdescriptors()['descriptors']
            if '/0h/0h/*h)' in descriptor['desc']
        ]
        expected_single_chain_hd_descriptors = 0
        if expect_hd:
            expected_single_chain_hd_descriptors = 1
            if passphrase is not None:
                expected_single_chain_hd_descriptors += 1
        assert_equal(len(single_chain_hd_descriptors), expected_single_chain_hd_descriptors)

        assert_equal(new_wallet_info['txcount'], 105)
        if passphrase is not None:
            assert_equal(new_wallet_info['unlocked_until'], 0)
        assert_equal(new_wallet.getbalance(), old_balance)
        new_balances = new_wallet.getbalances()['mine']
        assert_equal(new_balances['trusted'], old_wallet_state['balance'])
        assert_equal(new_balances['untrusted_pending'], old_wallet_state['unconfirmed_balance'])
        assert_equal(new_balances['immature'], old_wallet_state['immature_balance'])
        assert_equal(new_wallet_info['txcount'], old_wallet_state['txcount'])
        new_txs = new_wallet.listtransactions("*", 1000)
        assert_equal(normalize_listtransactions(new_txs), normalize_listtransactions(old_txs))

        # Verify all addresses are still owned
        for addr in all_addresses:
            assert_equal(new_wallet.getaddressinfo(addr)['ismine'], True)
        assert_equal(new_wallet.getaddressinfo(unfunded_address)['ismine'], True)

        # Test post-migration functionality
        new_addr = new_wallet.getnewaddress()
        if passphrase is not None:
            assert_raises_rpc_error(-13, "Please enter the wallet passphrase", new_wallet.sendtoaddress, new_addr, 0.1)
            new_wallet.walletpassphrase(passphrase, 60)
        txid = new_wallet.sendtoaddress(new_addr, 0.1)
        assert_is_hash_string(txid)

        self.stop_node(new_node_idx)
        self.log.info(f"{wallet_type} wallet migration successful")

    def run_test(self):
        self.log.info("Testing wallet migration from v0.14.3")

        self.run_migration_test(
            wallet_type="non-HD",
            old_node_idx=0,
            new_node_idx=1,
            extra_args_old=["-usehd=0", "-keypool=10"],
            expected_version=FEATURE_COMPRPUBKEY,
            expect_hd=False,
        )

        self.run_migration_test(
            wallet_type="encrypted non-HD",
            old_node_idx=2,
            new_node_idx=3,
            extra_args_old=["-usehd=0", "-keypool=10"],
            expected_version=FEATURE_COMPRPUBKEY,
            expect_hd=False,
            passphrase=WALLET_PASSPHRASE,
        )

        self.run_migration_test(
            wallet_type="HD (VERSION_HD_BASE)",
            old_node_idx=4,
            new_node_idx=5,
            extra_args_old=["-keypool=10"],
            expected_version=FEATURE_HD,
            expect_hd=True,
        )

        self.run_migration_test(
            wallet_type="encrypted HD (VERSION_HD_BASE)",
            old_node_idx=6,
            new_node_idx=7,
            extra_args_old=["-keypool=10"],
            expected_version=FEATURE_HD,
            expect_hd=True,
            passphrase=WALLET_PASSPHRASE,
        )


if __name__ == '__main__':
    WalletAncientMigrationTest(__file__).main()