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

Code before:
block updates are not going to be received. At wallet loading, we
check against chain if confirmed txn are still in chain and change
their status if block in which they have been included has been
disconnected.
"""

from decimal import Decimal
import shutil

from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
        assert_equal,
        assert_raises_rpc_error
)

class ReorgsRestoreTest(BitcoinTestFramework):
    def add_options(self, parser):
        self.add_wallet_options(parser)

    def set_test_params(self):
        self.num_nodes = 3

    def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()

    def test_coinbase_automatic_abandon_during_startup(self):
        ##########################################################################################################
        # Verify the wallet marks coinbase transactions, and their descendants, as abandoned during startup when #
        # the block is no longer part of the best chain.                                                         #
        ##########################################################################################################
        self.log.info("Test automatic coinbase abandonment during startup")
        # Test setup: Sync nodes for the coming test, ensuring both are at the same block, then disconnect them to
        # generate two competing chains. After disconnection, verify no other peer connection exists.
        self.connect_nodes(1, 0)
        self.sync_blocks(self.nodes[:2])
        self.disconnect_nodes(1, 0)
        assert all(len(node.getpeerinfo()) == 0 for node in self.nodes[:2])

        # Create a new block in node0, coinbase going to wallet0
        self.nodes[0].createwallet(wallet_name="w0", load_on_startup=True)
        wallet0 = self.nodes[0].get_wallet_rpc("w0")
        self.generatetoaddress(self.nodes[0], 1, wallet0.getnewaddress(), sync_fun=self.no_op)
        node0_coinbase_tx_hash = wallet0.getblock(wallet0.getbestblockhash(), verbose=1)['tx'][0]

        # Mine 100 blocks on top to mature the coinbase and create a descendant
        self.generate(self.nodes[0], 101, sync_fun=self.no_op)
        # Make descendant, send-to-self
        descendant_tx_id = wallet0.sendtoaddress(wallet0.getnewaddress(), 1)

        # Verify balance
        wallet0.syncwithvalidationinterfacequeue()

Review comment (bug):
`Test that wallet doesn't crash due to a duplicate block disconnection event after an unclean shutdown`

Code after:
block updates are not going to be received. At wallet loading, we
check against chain if confirmed txn are still in chain and change
their status if block in which they have been included has been
disconnected.
"""

from decimal import Decimal
import shutil

from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
        assert_equal,
        assert_greater_than,
        assert_raises_rpc_error
)

class ReorgsRestoreTest(BitcoinTestFramework):
    def add_options(self, parser):
        self.add_wallet_options(parser)

    def set_test_params(self):
        self.num_nodes = 3

    def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()

    def test_coinbase_automatic_abandon_during_startup(self):
        ##########################################################################################################
        # Verify the wallet marks coinbase transactions, and their descendants, as abandoned during startup when #
        # the block is no longer part of the best chain.                                                         #
        ##########################################################################################################
        self.log.info("Test automatic coinbase abandonment during startup")
        # Test setup: Sync nodes for the coming test, ensuring both are at the same block, then disconnect them to
        # generate two competing chains. After disconnection, verify no other peer connection exists.
        self.connect_nodes(1, 0)
        self.sync_blocks(self.nodes[:2])
        self.disconnect_nodes(1, 0)
        assert all(len(node.getpeerinfo()) == 0 for node in self.nodes[:2])

        # Create a new block in node0, coinbase going to wallet0
        self.nodes[0].createwallet(wallet_name="w0", load_on_startup=True)
        wallet0 = self.nodes[0].get_wallet_rpc("w0")
        self.generatetoaddress(self.nodes[0], 1, wallet0.getnewaddress(), sync_fun=self.no_op)
        node0_coinbase_tx_hash = wallet0.getblock(wallet0.getbestblockhash(), verbose=1)['tx'][0]

        # Mine 100 blocks on top to mature the coinbase and create a descendant
        self.generate(self.nodes[0], 101, sync_fun=self.no_op)
        # Make descendant, send-to-self
        descendant_tx_id = wallet0.sendtoaddress(wallet0.getnewaddress(), 1)

        # Verify balance