chaincode_test

chaincode_test.py

#!/usr/bin/env python3
# Copyright (c) 2017-2021 The Bitcoin Core developers
# Distributed under the MIT software license, see the accompanying
# file COPYING or http://www.opensource.org/licenses/mit-license.php.
"""An example functional test

getting node 1 to mine another block, send it to node 2, and check that node 2 received it. 
"""

from collections import defaultdict
import time

from test_framework.blocktools import (create_block, create_coinbase)
from test_framework.messages import CInv, MSG_BLOCK
from test_framework.p2p import (
    P2PInterface,
    msg_block,
    msg_getdata,
    p2p_lock,
)
from test_framework.test_framework import BitcoinTestFramework
from test_framework.util import (
    assert_equal,
)

# P2PInterface is a class containing callbacks to be executed when a P2P
# message is received from the node-under-test. Subclass P2PInterface and
# override the on_*() methods if you need custom behaviour.
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

def nop():
    ''' No Operation '''
    pass

class BlockPublishTest(BitcoinTestFramework):

    # Override the set_test_params(), skip_test_if_missing_module(), add_options(), setup_chain(), setup_network()
    # and setup_nodes() methods to customize the test setup as required.

    def set_test_params(self):
        
        self.setup_clean_chain = False
        self.num_nodes = 2
        # Use self.extra_args to change command-line arguments for the nodes
        self.extra_args = [[], []]


    def setup_network(self):
        ''' Override in order to not connect initially '''
        self.setup_nodes()



    def run_test(self):
        """Main test logic"""

        # Create P2P connections will wait for a verack to make sure the connection is fully up
        peer_messaging = self.nodes[0].add_p2p_connection(BaseNode())
        time.sleep(1)



        self.log.info("Initial state:")
        self.log.info(f"Current block for node 0: {self.nodes[0].getbestblockhash()}")
        self.log.info(f"Current block for node 1: {self.nodes[1].getbestblockhash()}")
        initial_block = self.nodes[0].getbestblockhash()
        
        block_hash = self.generate(self.nodes[0], nblocks=1, sync_fun=lambda: nop() )[0] # somehow easier to not sync at all?!
        self.log.info(f"nodes are not yet connected but let's mine a block on node 0: {block_hash}")
        self.log.info(f"Current block for node 0: {self.nodes[0].getbestblockhash()}")
        self.log.info(f"Current block for node 1: {self.nodes[1].getbestblockhash()}")
        self.log.info("Check whether block did not get propagated")
        assert self.nodes[0].getbestblockhash() != initial_block
        assert self.nodes[0].getbestblockhash() != self.nodes[1].getbestblockhash()
        self.log.info("connect the nodes")
        self.connect_nodes(0, 1)
        
        self.log.info("sync them")
        self.sync_all()

        self.log.info("Check blockhashes")
        self.log.info(f"Current block for node 0: {self.nodes[0].getbestblockhash()}")
        self.log.info(f"Current block for node 1: {self.nodes[1].getbestblockhash()}")

        assert self.nodes[1].getbestblockhash() == block_hash
        assert self.nodes[0].getbestblockhash() == self.nodes[1].getbestblockhash()

        self.log.info("peer_messaging.block_receive_map:")
        self.log.info(peer_messaging.block_receive_map)
        self.log.info("Listing last_messages:")
        for key, value in peer_messaging.last_message.items():
            self.log.info(f"  {key}: {value}")


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