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

Code before:
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()
        assert(wallet0.getbalances()['mine']['trusted'] > 0)

        # Now create a fork in node1. This will be used to replace node0's chain later.
        self.nodes[1].createwallet(wallet_name="w1", load_on_startup=True)
        wallet1 = self.nodes[1].get_wallet_rpc("w1")
        self.generatetoaddress(self.nodes[1], 1, wallet1.getnewaddress(), sync_fun=self.no_op)
        wallet1.syncwithvalidationinterfacequeue()

        # Verify both nodes are on a different chain
        block0_best_hash, block1_best_hash = wallet0.getbestblockhash(), wallet1.getbestblockhash()

Review comment (suggestion):
Might be in the nit-land but I find reading `25` here distracting. I had to figure out why it's 25 that is because in regtest the subsidy halving interval is 150 and initially a 199 blocks long chain is created in the regtest. The block count at this stage is 215 which is because few blocks were created in the preceding test in this file, thereby making this test dependent on the changes of that test as well.
IMO, a simple balance greater than 0 check is sufficient.

Also generally, I am inclined to creating and using a constant, for such usages of 25, called `INITIAL_ACTIVE_BLOCK_REWARD` in the test framework.

Code after:
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
        wallet0.syncwithvalidationinterfacequeue()
        assert(wallet0.getbalances()['mine']['trusted'] > 0)

        # Now create a fork in node1. This will be used to replace node0's chain later.
        self.nodes[1].createwallet(wallet_name="w1", load_on_startup=True)
        wallet1 = self.nodes[1].get_wallet_rpc("w1")
        self.generatetoaddress(self.nodes[1], 1, wallet1.getnewaddress(), sync_fun=self.no_op)
        wallet1.syncwithvalidationinterfacequeue()

        # Verify both nodes are on a different chain