-
-
Save amovfx/a76378090d9ed4c6ff561e82f6e09156 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
"""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() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment