Chaincode test

test_chaincode.py

"""Test mining a block.

"""

from collections import defaultdict

from test_framework.test_framework import BitcoinTestFramework

from test_framework.blocktools import (create_block, create_coinbase)
from test_framework.messages import (
    CBlock,
    CBlockHeader,
    BLOCK_HEADER_SIZE,
    ser_uint256,
)

from test_framework.blocktools import (
    create_coinbase,
    get_witness_script,
    NORMAL_GBT_REQUEST_PARAMS,
    TIME_GENESIS_BLOCK,
)

from test_framework.p2p import (
    P2PInterface,
    msg_block,
    msg_getdata,
    p2p_lock,
)

from test_framework.util import (
    assert_equal,
)

class BaseNode(P2PInterface):
    def __init__(self):
        """Initialize the P2PInterface

        Used to initialize custom properties for the Node that aren't
        included by default in the base class. Be aware that the P2PInterface
        base class already stores a counter for each P2P message type and the
        last received message of each type, which should be sufficient for the
        needs of most tests.

        Call super().__init__() first for standard initialization and then
        initialize custom properties."""
        super().__init__()
        # Stores a dictionary of all blocks received
        self.block_receive_map = defaultdict(int)

    def on_block(self, message):
        """Override the standard on_block callback

        Store the hash of a received block in the dictionary."""
        message.block.calc_sha256()
        self.block_receive_map[message.block.sha256] += 1

    def on_inv(self, message):
        """Override the standard on_inv callback"""
        pass

class ChainCodeTest(BitcoinTestFramework):


    def set_test_params(self):
        self.num_nodes = 2
        self.setup_clean_chain = True
        self.extra_args = [[], ["-logips"]]

    def skip_test_if_missing_module(self):
        self.skip_if_no_wallet()

    def setup_network(self):
        self.setup_nodes()
        self.connect_nodes(0, 1)
        self.sync_all()


    def run_test(self):
        self.log.info("Starting test!")

        peer_messaging = self.nodes[0].add_p2p_connection(BaseNode())

        blocks = [int(self.generate(self.nodes[0], sync_fun=lambda: self.sync_all(), nblocks=1)[0], 16)]

        self.log.info("Create some blocks")
        self.tip = int(self.nodes[0].getbestblockhash(), 16)
        self.block_time = self.nodes[0].getblock(self.nodes[0].getbestblockhash())['time'] + 1

        height = self.nodes[0].getblockcount()


        # Generating a block on one of the nodes will get us out of IBD
        self.log.info("Generating block!")
        block = create_block(self.tip, create_coinbase(height+1), self.block_time)
        block.solve()
        block_message = msg_block(block)
        peer_messaging.send_message(block_message)
        self.tip = block.sha256
        blocks.append(self.tip)
        self.block_time += 1
        height += 1

        self.log.info("Wait for node1 to reach current tip (height 2) using RPC")
        self.nodes[1].waitforblockheight(1)

        self.connect_nodes(0, 1)
        self.sync_all()

        self.nodes[0].disconnect_p2ps()
        peer_receiving = self.nodes[1].add_p2p_connection(BaseNode())

        with p2p_lock:
            for block in peer_receiving.block_receive_map.values():
                assert_equal(block, 1)

if __name__ == '__main__':
    ChainCodeTest().main()