Created
July 2, 2022 12:04
-
-
Save k9ert/d3d7433be7dade6c605521751c86eca1 to your computer and use it in GitHub Desktop.
chaincode_test
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment