diff --git a/test/functional/abc-p2p-compactblocks.py b/test/functional/abc-p2p-compactblocks.py index 1bd2dc660..d832b904c 100755 --- a/test/functional/abc-p2p-compactblocks.py +++ b/test/functional/abc-p2p-compactblocks.py @@ -1,358 +1,358 @@ #!/usr/bin/env python3 # Copyright (c) 2015-2016 The Bitcoin Core developers # Copyright (c) 2017 The Bitcoin developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """ This test checks simple acceptance of bigger blocks via p2p. It is derived from the much more complex p2p-fullblocktest. The intention is that small tests can be derived from this one, or this one can be extended, to cover the checks done for bigger blocks (e.g. sigops limits). """ from test_framework.test_framework import ComparisonTestFramework from test_framework.util import * from test_framework.comptool import TestManager, TestInstance, RejectResult from test_framework.blocktools import * import time from test_framework.script import * from test_framework.cdefs import (ONE_MEGABYTE, LEGACY_MAX_BLOCK_SIZE, MAX_BLOCK_SIGOPS_PER_MB, MAX_TX_SIGOPS_COUNT) from collections import deque class PreviousSpendableOutput(): def __init__(self, tx=CTransaction(), n=-1): self.tx = tx self.n = n # the output we're spending # TestNode: A peer we use to send messages to bitcoind, and store responses. class TestNode(NodeConnCB): def __init__(self): self.last_sendcmpct = None self.last_cmpctblock = None self.last_getheaders = None self.last_headers = None super().__init__() def on_sendcmpct(self, conn, message): self.last_sendcmpct = message def on_cmpctblock(self, conn, message): self.last_cmpctblock = message self.last_cmpctblock.header_and_shortids.header.calc_sha256() def on_getheaders(self, conn, message): self.last_getheaders = message def on_headers(self, conn, message): self.last_headers = message for x in self.last_headers.headers: x.calc_sha256() def clear_block_data(self): with mininode_lock: self.last_sendcmpct = None self.last_cmpctblock = None class FullBlockTest(ComparisonTestFramework): # Can either run this test as 1 node with expected answers, or two and compare them. # Change the "outcome" variable from each TestInstance object to only do # the comparison. def set_test_params(self): self.num_nodes = 1 self.setup_clean_chain = True self.block_heights = {} self.tip = None self.blocks = {} self.excessive_block_size = 16 * ONE_MEGABYTE self.extra_args = [['-norelaypriority', '-whitelist=127.0.0.1', '-limitancestorcount=999999', '-limitancestorsize=999999', '-limitdescendantcount=999999', '-limitdescendantsize=999999', '-maxmempool=99999', "-excessiveblocksize=%d" % self.excessive_block_size]] def add_options(self, parser): super().add_options(parser) parser.add_argument( "--runbarelyexpensive", dest="runbarelyexpensive", default=True) def run_test(self): self.test = TestManager(self, self.options.tmpdir) self.test.add_all_connections(self.nodes) # Start up network handling in another thread NetworkThread().start() # Set the blocksize to 2MB as initial condition self.nodes[0].setexcessiveblock(self.excessive_block_size) self.test.run() def add_transactions_to_block(self, block, tx_list): [tx.rehash() for tx in tx_list] block.vtx.extend(tx_list) # this is a little handier to use than the version in blocktools.py def create_tx(self, spend_tx, n, value, script=CScript([OP_TRUE])): tx = create_transaction(spend_tx, n, b"", value, script) return tx def next_block(self, number, spend=None, script=CScript([OP_TRUE]), block_size=0, extra_txns=0): if self.tip == None: base_block_hash = self.genesis_hash block_time = int(time.time()) + 1 else: base_block_hash = self.tip.sha256 block_time = self.tip.nTime + 1 # First create the coinbase height = self.block_heights[base_block_hash] + 1 coinbase = create_coinbase(height) coinbase.rehash() if spend == None: # We need to have something to spend to fill the block. assert_equal(block_size, 0) block = create_block(base_block_hash, coinbase, block_time) else: # all but one satoshi to fees coinbase.vout[0].nValue += spend.tx.vout[spend.n].nValue - 1 coinbase.rehash() block = create_block(base_block_hash, coinbase, block_time) # Make sure we have plenty enough to spend going forward. spendable_outputs = deque([spend]) def get_base_transaction(): # Create the new transaction tx = CTransaction() # Spend from one of the spendable outputs spend = spendable_outputs.popleft() tx.vin.append(CTxIn(COutPoint(spend.tx.sha256, spend.n))) # Add spendable outputs for i in range(4): tx.vout.append(CTxOut(0, CScript([OP_TRUE]))) spendable_outputs.append(PreviousSpendableOutput(tx, i)) pad_tx(tx) return tx tx = get_base_transaction() # Make it the same format as transaction added for padding and save the size. # It's missing the padding output, so we add a constant to account for it. tx.rehash() base_tx_size = len(tx.serialize()) + 18 # If a specific script is required, add it. if script != None: tx.vout.append(CTxOut(1, script)) # Put some random data into the first transaction of the chain to randomize ids. tx.vout.append( CTxOut(0, CScript([random.randint(0, 256), OP_RETURN]))) # Add the transaction to the block self.add_transactions_to_block(block, [tx]) # Add transaction until we reach the expected transaction count for _ in range(extra_txns): self.add_transactions_to_block(block, [get_base_transaction()]) # If we have a block size requirement, just fill # the block until we get there current_block_size = len(block.serialize()) overage_bytes = 0 while current_block_size < block_size: # We will add a new transaction. That means the size of # the field enumerating how many transaction go in the block # may change. current_block_size -= len(ser_compact_size(len(block.vtx))) current_block_size += len(ser_compact_size(len(block.vtx) + 1)) # Add padding to fill the block. left_to_fill = block_size - current_block_size # Don't go over the 1 mb limit for a txn if left_to_fill > 500000: # Make sure we eat up non-divisible by 100 amounts quickly # Also keep transaction less than 1 MB left_to_fill = 500000 + left_to_fill % 100 # Create the new transaction tx = get_base_transaction() pad_tx(tx, left_to_fill - overage_bytes) if len(tx.serialize()) + current_block_size > block_size: # Our padding was too big try again overage_bytes += 1 continue # Add the tx to the list of transactions to be included # in the block. self.add_transactions_to_block(block, [tx]) current_block_size += len(tx.serialize()) # Now that we added a bunch of transaction, we need to recompute # the merkle root. make_conform_to_ctor(block) block.hashMerkleRoot = block.calc_merkle_root() # Check that the block size is what's expected if block_size > 0: assert_equal(len(block.serialize()), block_size) # Do PoW, which is cheap on regnet block.solve() self.tip = block self.block_heights[block.sha256] = height assert number not in self.blocks self.blocks[number] = block return block def get_tests(self): self.genesis_hash = int(self.nodes[0].getbestblockhash(), 16) self.block_heights[self.genesis_hash] = 0 spendable_outputs = [] # save the current tip so it can be spent by a later block def save_spendable_output(): spendable_outputs.append(self.tip) # get an output that we previously marked as spendable def get_spendable_output(): return PreviousSpendableOutput(spendable_outputs.pop(0).vtx[0], 0) # returns a test case that asserts that the current tip was accepted def accepted(): return TestInstance([[self.tip, True]]) # returns a test case that asserts that the current tip was rejected def rejected(reject=None): if reject is None: return TestInstance([[self.tip, False]]) else: return TestInstance([[self.tip, reject]]) # move the tip back to a previous block def tip(number): self.tip = self.blocks[number] # shorthand for functions block = self.next_block # Create a new block block(0) save_spendable_output() yield accepted() # Now we need that block to mature so we can spend the coinbase. test = TestInstance(sync_every_block=False) for i in range(99): block(5000 + i) test.blocks_and_transactions.append([self.tip, True]) save_spendable_output() # Get to one block of the May 15, 2018 HF activation for i in range(6): block(5100 + i) test.blocks_and_transactions.append([self.tip, True]) # Send it all to the node at once. yield test # collect spendable outputs now to avoid cluttering the code later on out = [] for i in range(100): out.append(get_spendable_output()) # Check that compact block also work for big blocks node = self.nodes[0] peer = TestNode() - peer.add_connection(NodeConn('127.0.0.1', p2p_port(0), node, peer)) + peer.add_connection(NodeConn('127.0.0.1', p2p_port(0), peer)) # Wait for connection to be etablished peer.wait_for_verack() # Wait for SENDCMPCT def received_sendcmpct(): return (peer.last_sendcmpct != None) wait_until(received_sendcmpct, timeout=30) sendcmpct = msg_sendcmpct() sendcmpct.version = 1 sendcmpct.announce = True peer.send_and_ping(sendcmpct) # Exchange headers def received_getheaders(): return (peer.last_getheaders != None) wait_until(received_getheaders, timeout=30) # Return the favor peer.send_message(peer.last_getheaders) # Wait for the header list def received_headers(): return (peer.last_headers != None) wait_until(received_headers, timeout=30) # It's like we know about the same headers ! peer.send_message(peer.last_headers) # Send a block b1 = block(1, spend=out[0], block_size=ONE_MEGABYTE + 1) yield accepted() # Checks the node to forward it via compact block def received_block(): return (peer.last_cmpctblock != None) wait_until(received_block, timeout=30) # Was it our block ? cmpctblk_header = peer.last_cmpctblock.header_and_shortids.header cmpctblk_header.calc_sha256() assert(cmpctblk_header.sha256 == b1.sha256) # Send a large block with numerous transactions. peer.clear_block_data() b2 = block(2, spend=out[1], extra_txns=70000, block_size=self.excessive_block_size - 1000) yield accepted() # Checks the node forwards it via compact block wait_until(received_block, timeout=30) # Was it our block ? cmpctblk_header = peer.last_cmpctblock.header_and_shortids.header cmpctblk_header.calc_sha256() assert(cmpctblk_header.sha256 == b2.sha256) # In order to avoid having to resend a ton of transactions, we invalidate # b2, which will send all its transactions in the mempool. node.invalidateblock(node.getbestblockhash()) # Let's send a compact block and see if the node accepts it. # Let's modify b2 and use it so that we can reuse the mempool. tx = b2.vtx[0] tx.vout.append(CTxOut(0, CScript([random.randint(0, 256), OP_RETURN]))) tx.rehash() b2.vtx[0] = tx b2.hashMerkleRoot = b2.calc_merkle_root() b2.solve() # Now we create the compact block and send it comp_block = HeaderAndShortIDs() comp_block.initialize_from_block(b2) peer.send_and_ping(msg_cmpctblock(comp_block.to_p2p())) # Check that compact block is received properly assert(int(node.getbestblockhash(), 16) == b2.sha256) if __name__ == '__main__': FullBlockTest().main() diff --git a/test/functional/abc-sync-chain.py b/test/functional/abc-sync-chain.py index bf0bd9e43..141971a8c 100755 --- a/test/functional/abc-sync-chain.py +++ b/test/functional/abc-sync-chain.py @@ -1,83 +1,83 @@ #!/usr/bin/env python3 # Copyright (c) 2018 The Bitcoin developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """ Test that a node receiving many (potentially out of order) blocks exits initial block download (IBD; this occurs once it has passed minimumchainwork) and continues to sync without seizing. """ import random from test_framework.blocktools import create_block, create_coinbase from test_framework.mininode import (CBlockHeader, NetworkThread, NodeConn, NodeConnCB, msg_block, msg_headers) from test_framework.test_framework import BitcoinTestFramework from test_framework.util import assert_equal, wait_until, p2p_port NUM_IBD_BLOCKS = 50 class BaseNode(NodeConnCB): def send_header(self, block): msg = msg_headers() msg.headers = [CBlockHeader(block)] self.send_message(msg) def send_block(self, block): self.send_message(msg_block(block)) class SyncChainTest(BitcoinTestFramework): def set_test_params(self): self.num_nodes = 1 # Setting minimumchainwork makes sure we test IBD as well as post-IBD self.extra_args = [ ["-minimumchainwork={:#x}".format(202 + 2 * NUM_IBD_BLOCKS)]] def run_test(self): node0conn = BaseNode() node0conn.add_connection( - NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0conn)) + NodeConn('127.0.0.1', p2p_port(0), node0conn)) NetworkThread().start() node0conn.wait_for_verack() node0 = self.nodes[0] tip = int(node0.getbestblockhash(), 16) height = node0.getblockcount() + 1 time = node0.getblock(node0.getbestblockhash())['time'] + 1 blocks = [] for i in range(NUM_IBD_BLOCKS * 2): block = create_block(tip, create_coinbase(height), time) block.solve() blocks.append(block) tip = block.sha256 height += 1 time += 1 # Headers need to be sent in-order for b in blocks: node0conn.send_header(b) # Send blocks in some random order for b in random.sample(blocks, len(blocks)): node0conn.send_block(b) # The node should eventually, completely sync without getting stuck def node_synced(): return node0.getbestblockhash() == blocks[-1].hash wait_until(node_synced) if __name__ == '__main__': SyncChainTest().main() diff --git a/test/functional/test_framework/comptool.py b/test/functional/test_framework/comptool.py index d6408c2b2..193a4b54e 100755 --- a/test/functional/test_framework/comptool.py +++ b/test/functional/test_framework/comptool.py @@ -1,438 +1,438 @@ #!/usr/bin/env python3 # Copyright (c) 2015-2016 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """Compare two or more bitcoinds to each other. To use, create a class that implements get_tests(), and pass it in as the test generator to TestManager. get_tests() should be a python generator that returns TestInstance objects. See below for definition. TestNode behaves as follows: Configure with a BlockStore and TxStore on_inv: log the message but don't request on_headers: log the chain tip on_pong: update ping response map (for synchronization) on_getheaders: provide headers via BlockStore on_getdata: provide blocks via BlockStore """ from .mininode import * from .blockstore import BlockStore, TxStore from .util import p2p_port, wait_until import logging logger = logging.getLogger("TestFramework.comptool") global mininode_lock class RejectResult(): """Outcome that expects rejection of a transaction or block.""" def __init__(self, code, reason=b''): self.code = code self.reason = reason def match(self, other): if self.code != other.code: return False return other.reason.startswith(self.reason) def __repr__(self): return '%i:%s' % (self.code, self.reason or '*') class TestNode(NodeConnCB): def __init__(self, block_store, tx_store): super().__init__() self.conn = None self.bestblockhash = None self.block_store = block_store self.block_request_map = {} self.tx_store = tx_store self.tx_request_map = {} self.block_reject_map = {} self.tx_reject_map = {} # When the pingmap is non-empty we're waiting for # a response self.pingMap = {} self.lastInv = [] self.closed = False def on_close(self, conn): self.closed = True def add_connection(self, conn): self.conn = conn def on_headers(self, conn, message): if len(message.headers) > 0: best_header = message.headers[-1] best_header.calc_sha256() self.bestblockhash = best_header.sha256 def on_getheaders(self, conn, message): response = self.block_store.headers_for( message.locator, message.hashstop) if response is not None: conn.send_message(response) def on_getdata(self, conn, message): [conn.send_message(r) for r in self.block_store.get_blocks(message.inv)] [conn.send_message(r) for r in self.tx_store.get_transactions(message.inv)] for i in message.inv: if i.type == 1: self.tx_request_map[i.hash] = True elif i.type == 2: self.block_request_map[i.hash] = True def on_inv(self, conn, message): self.lastInv = [x.hash for x in message.inv] def on_pong(self, conn, message): try: del self.pingMap[message.nonce] except KeyError: raise AssertionError( "Got pong for unknown ping [%s]" % repr(message)) def on_reject(self, conn, message): if message.message == b'tx': self.tx_reject_map[message.data] = RejectResult( message.code, message.reason) if message.message == b'block': self.block_reject_map[message.data] = RejectResult( message.code, message.reason) def send_inv(self, obj): mtype = 2 if isinstance(obj, CBlock) else 1 self.conn.send_message(msg_inv([CInv(mtype, obj.sha256)])) def send_getheaders(self): # We ask for headers from their last tip. m = msg_getheaders() m.locator = self.block_store.get_locator(self.bestblockhash) self.conn.send_message(m) def send_header(self, header): m = msg_headers() m.headers.append(header) self.conn.send_message(m) # This assumes BIP31 def send_ping(self, nonce): self.pingMap[nonce] = True self.conn.send_message(msg_ping(nonce)) def received_ping_response(self, nonce): return nonce not in self.pingMap def send_mempool(self): self.lastInv = [] self.conn.send_message(msg_mempool()) # TestInstance: # # Instances of these are generated by the test generator, and fed into the # comptool. # # "blocks_and_transactions" should be an array of # [obj, True/False/None, hash/None]: # - obj is either a CBlock, CBlockHeader, or a CTransaction, and # - the second value indicates whether the object should be accepted # into the blockchain or mempool (for tests where we expect a certain # answer), or "None" if we don't expect a certain answer and are just # comparing the behavior of the nodes being tested. # - the third value is the hash to test the tip against (if None or omitted, # use the hash of the block) # - NOTE: if a block header, no test is performed; instead the header is # just added to the block_store. This is to facilitate block delivery # when communicating with headers-first clients (when withholding an # intermediate block). # sync_every_block: if True, then each block will be inv'ed, synced, and # nodes will be tested based on the outcome for the block. If False, # then inv's accumulate until all blocks are processed (or max inv size # is reached) and then sent out in one inv message. Then the final block # will be synced across all connections, and the outcome of the final # block will be tested. # sync_every_tx: analogous to behavior for sync_every_block, except if outcome # on the final tx is None, then contents of entire mempool are compared # across all connections. (If outcome of final tx is specified as true # or false, then only the last tx is tested against outcome.) class TestInstance(): def __init__(self, objects=None, sync_every_block=True, sync_every_tx=False): self.blocks_and_transactions = objects if objects else [] self.sync_every_block = sync_every_block self.sync_every_tx = sync_every_tx class TestManager(): def __init__(self, testgen, datadir): self.test_generator = testgen self.connections = [] self.test_nodes = [] self.block_store = BlockStore(datadir) self.tx_store = TxStore(datadir) self.ping_counter = 1 def add_all_connections(self, nodes): for i in range(len(nodes)): # Create a p2p connection to each node test_node = TestNode(self.block_store, self.tx_store) self.test_nodes.append(test_node) self.connections.append( - NodeConn('127.0.0.1', p2p_port(i), nodes[i], test_node)) + NodeConn('127.0.0.1', p2p_port(i), test_node)) # Make sure the TestNode (callback class) has a reference to its # associated NodeConn test_node.add_connection(self.connections[-1]) def clear_all_connections(self): self.connections = [] self.test_nodes = [] def wait_for_disconnections(self): def disconnected(): return all(node.closed for node in self.test_nodes) wait_until(disconnected, timeout=10, lock=mininode_lock) def wait_for_verack(self): return all(node.wait_for_verack() for node in self.test_nodes) def wait_for_pings(self, counter): def received_pongs(): return all(node.received_ping_response(counter) for node in self.test_nodes) wait_until(received_pongs, lock=mininode_lock) # sync_blocks: Wait for all connections to request the blockhash given # then send get_headers to find out the tip of each node, and synchronize # the response by using a ping (and waiting for pong with same nonce). def sync_blocks(self, blockhash, num_blocks): def blocks_requested(): return all( blockhash in node.block_request_map and node.block_request_map[blockhash] for node in self.test_nodes ) # --> error if not requested wait_until(blocks_requested, attempts=20 * num_blocks, lock=mininode_lock) # Send getheaders message [c.cb.send_getheaders() for c in self.connections] # Send ping and wait for response -- synchronization hack [c.cb.send_ping(self.ping_counter) for c in self.connections] self.wait_for_pings(self.ping_counter) self.ping_counter += 1 # Analogous to sync_block (see above) def sync_transaction(self, txhash, num_events): # Wait for nodes to request transaction (50ms sleep * 20 tries * num_events) def transaction_requested(): return all( txhash in node.tx_request_map and node.tx_request_map[txhash] for node in self.test_nodes ) # --> error if not requested wait_until(transaction_requested, attempts=20 * num_events, lock=mininode_lock) # Get the mempool [c.cb.send_mempool() for c in self.connections] # Send ping and wait for response -- synchronization hack [c.cb.send_ping(self.ping_counter) for c in self.connections] self.wait_for_pings(self.ping_counter) self.ping_counter += 1 # Sort inv responses from each node with mininode_lock: [c.cb.lastInv.sort() for c in self.connections] # Verify that the tip of each connection all agree with each other, and # with the expected outcome (if given) def check_results(self, blockhash, outcome): with mininode_lock: for c in self.connections: if outcome is None: if c.cb.bestblockhash != self.connections[0].cb.bestblockhash: return False # Check that block was rejected w/ code elif isinstance(outcome, RejectResult): if c.cb.bestblockhash == blockhash: return False if blockhash not in c.cb.block_reject_map: logger.error( 'Block not in reject map: %064x' % (blockhash)) return False if not outcome.match(c.cb.block_reject_map[blockhash]): logger.error('Block rejected with %s instead of expected %s: %064x' % ( c.cb.block_reject_map[blockhash], outcome, blockhash)) return False elif ((c.cb.bestblockhash == blockhash) != outcome): return False return True # Either check that the mempools all agree with each other, or that # txhash's presence in the mempool matches the outcome specified. # This is somewhat of a strange comparison, in that we're either comparing # a particular tx to an outcome, or the entire mempools altogether; # perhaps it would be useful to add the ability to check explicitly that # a particular tx's existence in the mempool is the same across all nodes. def check_mempool(self, txhash, outcome): with mininode_lock: for c in self.connections: if outcome is None: # Make sure the mempools agree with each other if c.cb.lastInv != self.connections[0].cb.lastInv: return False # Check that tx was rejected w/ code elif isinstance(outcome, RejectResult): if txhash in c.cb.lastInv: return False if txhash not in c.cb.tx_reject_map: logger.error('Tx not in reject map: %064x' % (txhash)) return False if not outcome.match(c.cb.tx_reject_map[txhash]): logger.error('Tx rejected with %s instead of expected %s: %064x' % ( c.cb.tx_reject_map[txhash], outcome, txhash)) return False elif ((txhash in c.cb.lastInv) != outcome): return False return True def run(self): # Wait until verack is received self.wait_for_verack() test_number = 1 for test_instance in self.test_generator.get_tests(): # We use these variables to keep track of the last block # and last transaction in the tests, which are used # if we're not syncing on every block or every tx. [block, block_outcome, tip] = [None, None, None] [tx, tx_outcome] = [None, None] invqueue = [] for test_obj in test_instance.blocks_and_transactions: b_or_t = test_obj[0] outcome = test_obj[1] # Determine if we're dealing with a block or tx if isinstance(b_or_t, CBlock): # Block test runner block = b_or_t block_outcome = outcome tip = block.sha256 # each test_obj can have an optional third argument # to specify the tip we should compare with # (default is to use the block being tested) if len(test_obj) >= 3: tip = test_obj[2] # Add to shared block_store, set as current block # If there was an open getdata request for the block # previously, and we didn't have an entry in the # block_store, then immediately deliver, because the # node wouldn't send another getdata request while # the earlier one is outstanding. first_block_with_hash = True if self.block_store.get(block.sha256) is not None: first_block_with_hash = False with mininode_lock: self.block_store.add_block(block) for c in self.connections: if first_block_with_hash and block.sha256 in c.cb.block_request_map and c.cb.block_request_map[block.sha256] == True: # There was a previous request for this block hash # Most likely, we delivered a header for this block # but never had the block to respond to the getdata c.send_message(msg_block(block)) else: c.cb.block_request_map[block.sha256] = False # Either send inv's to each node and sync, or add # to invqueue for later inv'ing. if (test_instance.sync_every_block): # if we expect success, send inv and sync every block # if we expect failure, just push the block and see what happens. if outcome == True: [c.cb.send_inv(block) for c in self.connections] self.sync_blocks(block.sha256, 1) else: [c.send_message(msg_block(block)) for c in self.connections] [c.cb.send_ping(self.ping_counter) for c in self.connections] self.wait_for_pings(self.ping_counter) self.ping_counter += 1 if (not self.check_results(tip, outcome)): raise AssertionError( "Test failed at test %d" % test_number) else: invqueue.append(CInv(2, block.sha256)) elif isinstance(b_or_t, CBlockHeader): block_header = b_or_t self.block_store.add_header(block_header) [c.cb.send_header(block_header) for c in self.connections] else: # Tx test runner assert(isinstance(b_or_t, CTransaction)) tx = b_or_t tx_outcome = outcome # Add to shared tx store and clear map entry with mininode_lock: self.tx_store.add_transaction(tx) for c in self.connections: c.cb.tx_request_map[tx.sha256] = False # Again, either inv to all nodes or save for later if (test_instance.sync_every_tx): [c.cb.send_inv(tx) for c in self.connections] self.sync_transaction(tx.sha256, 1) if (not self.check_mempool(tx.sha256, outcome)): raise AssertionError( "Test failed at test %d" % test_number) else: invqueue.append(CInv(1, tx.sha256)) # Ensure we're not overflowing the inv queue if len(invqueue) == MAX_INV_SZ: [c.send_message(msg_inv(invqueue)) for c in self.connections] invqueue = [] # Do final sync if we weren't syncing on every block or every tx. if (not test_instance.sync_every_block and block is not None): if len(invqueue) > 0: [c.send_message(msg_inv(invqueue)) for c in self.connections] invqueue = [] self.sync_blocks(block.sha256, len( test_instance.blocks_and_transactions)) if (not self.check_results(tip, block_outcome)): raise AssertionError( "Block test failed at test %d" % test_number) if (not test_instance.sync_every_tx and tx is not None): if len(invqueue) > 0: [c.send_message(msg_inv(invqueue)) for c in self.connections] invqueue = [] self.sync_transaction(tx.sha256, len( test_instance.blocks_and_transactions)) if (not self.check_mempool(tx.sha256, tx_outcome)): raise AssertionError( "Mempool test failed at test %d" % test_number) logger.info("Test %d: PASS" % test_number) test_number += 1 [c.disconnect_node() for c in self.connections] self.wait_for_disconnections() self.block_store.close() self.tx_store.close() diff --git a/test/functional/test_framework/mininode.py b/test/functional/test_framework/mininode.py index a2f99240e..406ed2edf 100755 --- a/test/functional/test_framework/mininode.py +++ b/test/functional/test_framework/mininode.py @@ -1,439 +1,435 @@ #!/usr/bin/env python3 # Copyright (c) 2010 ArtForz -- public domain half-a-node # Copyright (c) 2012 Jeff Garzik # Copyright (c) 2010-2016 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """Bitcoin P2P network half-a-node. This python code was modified from ArtForz' public domain half-a-node, as found in the mini-node branch of http://github.com/jgarzik/pynode. NodeConn: an object which manages p2p connectivity to a bitcoin node NodeConnCB: a base class that describes the interface for receiving callbacks with network messages from a NodeConn """ import asyncore from collections import defaultdict from io import BytesIO import logging import socket import struct import sys import time from threading import RLock, Thread from test_framework.messages import * logger = logging.getLogger("TestFramework.mininode") MESSAGEMAP = { b"addr": msg_addr, b"block": msg_block, b"blocktxn": msg_blocktxn, b"cmpctblock": msg_cmpctblock, b"feefilter": msg_feefilter, b"getaddr": msg_getaddr, b"getblocks": msg_getblocks, b"getblocktxn": msg_getblocktxn, b"getdata": msg_getdata, b"getheaders": msg_getheaders, b"headers": msg_headers, b"inv": msg_inv, b"mempool": msg_mempool, b"ping": msg_ping, b"pong": msg_pong, b"reject": msg_reject, b"sendcmpct": msg_sendcmpct, b"sendheaders": msg_sendheaders, b"tx": msg_tx, b"verack": msg_verack, b"version": msg_version, } MAGIC_BYTES = { "mainnet": b"\xe3\xe1\xf3\xe8", "testnet3": b"\xf4\xe5\xf3\xf4", "regtest": b"\xda\xb5\xbf\xfa", } class NodeConnCB(): """Callback and helper functions for P2P connection to a bitcoind node. Individual testcases should subclass this and override the on_* methods if they want to alter message handling behaviour.""" def __init__(self): # Track whether we have a P2P connection open to the node self.connected = False self.connection = None # Track number of messages of each type received and the most recent # message of each type self.message_count = defaultdict(int) self.last_message = {} # A count of the number of ping messages we've sent to the node self.ping_counter = 1 # Message receiving methods def deliver(self, conn, message): """Receive message and dispatch message to appropriate callback. We keep a count of how many of each message type has been received and the most recent message of each type.""" with mininode_lock: try: command = message.command.decode('ascii') self.message_count[command] += 1 self.last_message[command] = message getattr(self, 'on_' + command)(conn, message) except: print("ERROR delivering %s (%s)" % (repr(message), sys.exc_info()[0])) raise # Callback methods. Can be overridden by subclasses in individual test # cases to provide custom message handling behaviour. def on_open(self, conn): self.connected = True def on_close(self, conn): self.connected = False self.connection = None def on_addr(self, conn, message): pass def on_block(self, conn, message): pass def on_blocktxn(self, conn, message): pass def on_cmpctblock(self, conn, message): pass def on_feefilter(self, conn, message): pass def on_getaddr(self, conn, message): pass def on_getblocks(self, conn, message): pass def on_getblocktxn(self, conn, message): pass def on_getdata(self, conn, message): pass def on_getheaders(self, conn, message): pass def on_headers(self, conn, message): pass def on_mempool(self, conn): pass def on_pong(self, conn, message): pass def on_reject(self, conn, message): pass def on_sendcmpct(self, conn, message): pass def on_sendheaders(self, conn, message): pass def on_tx(self, conn, message): pass def on_inv(self, conn, message): want = msg_getdata() for i in message.inv: if i.type != 0: want.inv.append(i) if len(want.inv): conn.send_message(want) def on_ping(self, conn, message): conn.send_message(msg_pong(message.nonce)) def on_verack(self, conn, message): - conn.ver_recv = conn.ver_send self.verack_received = True def on_version(self, conn, message): assert message.nVersion >= MIN_VERSION_SUPPORTED, "Version {} received. Test framework only supports versions greater than {}".format( message.nVersion, MIN_VERSION_SUPPORTED) conn.send_message(msg_verack()) conn.nServices = message.nServices # Connection helper methods def add_connection(self, conn): self.connection = conn def wait_for_disconnect(self, timeout=60): def test_function(): return not self.connected wait_until(test_function, timeout=timeout, lock=mininode_lock) # Message receiving helper methods def wait_for_block(self, blockhash, timeout=60): def test_function(): return self.last_message.get( "block") and self.last_message["block"].block.rehash() == blockhash wait_until(test_function, timeout=timeout, lock=mininode_lock) def wait_for_getdata(self, timeout=60): def test_function(): return self.last_message.get("getdata") wait_until(test_function, timeout=timeout, lock=mininode_lock) def wait_for_getheaders(self, timeout=60): def test_function(): return self.last_message.get("getheaders") wait_until(test_function, timeout=timeout, lock=mininode_lock) def wait_for_inv(self, expected_inv, timeout=60): """Waits for an INV message and checks that the first inv object in the message was as expected.""" if len(expected_inv) > 1: raise NotImplementedError( "wait_for_inv() will only verify the first inv object") def test_function(): return self.last_message.get("inv") and \ self.last_message["inv"].inv[0].type == expected_inv[0].type and \ self.last_message["inv"].inv[0].hash == expected_inv[0].hash wait_until(test_function, timeout=timeout, lock=mininode_lock) def wait_for_verack(self, timeout=60): def test_function(): return self.message_count["verack"] wait_until(test_function, timeout=timeout, lock=mininode_lock) # Message sending helper functions def send_message(self, message): if self.connection: self.connection.send_message(message) else: logger.error("Cannot send message. No connection to node!") def send_and_ping(self, message): self.send_message(message) self.sync_with_ping() # Sync up with the node def sync_with_ping(self, timeout=60): self.send_message(msg_ping(nonce=self.ping_counter)) def test_function(): if not self.last_message.get("pong"): return False return self.last_message["pong"].nonce == self.ping_counter wait_until(test_function, timeout=timeout, lock=mininode_lock) self.ping_counter += 1 class NodeConn(asyncore.dispatcher): """The actual NodeConn class This class provides an interface for a p2p connection to a specified node.""" - def __init__(self, dstaddr, dstport, rpc, callback, net="regtest", services=NODE_NETWORK, send_version=True): + def __init__(self, dstaddr, dstport, callback, net="regtest", services=NODE_NETWORK, send_version=True): asyncore.dispatcher.__init__(self, map=mininode_socket_map) self.dstaddr = dstaddr self.dstport = dstport self.create_socket(socket.AF_INET, socket.SOCK_STREAM) self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) self.sendbuf = b"" self.recvbuf = b"" - self.ver_send = 209 - self.ver_recv = 209 self.last_sent = 0 self.state = "connecting" self.network = net self.cb = callback self.disconnect = False self.nServices = 0 if send_version: # stuff version msg into sendbuf vt = msg_version() vt.nServices = services vt.addrTo.ip = self.dstaddr vt.addrTo.port = self.dstport vt.addrFrom.ip = "0.0.0.0" vt.addrFrom.port = 0 self.send_message(vt, True) logger.info('Connecting to Bitcoin Node: %s:%d' % (self.dstaddr, self.dstport)) try: self.connect((dstaddr, dstport)) except: self.handle_close() - self.rpc = rpc # Connection and disconnection methods def handle_connect(self): if self.state != "connected": logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport)) self.state = "connected" self.cb.on_open(self) def handle_close(self): logger.debug("Closing connection to: %s:%d" % (self.dstaddr, self.dstport)) self.state = "closed" self.recvbuf = b"" self.sendbuf = b"" try: self.close() except: pass self.cb.on_close(self) def disconnect_node(self): """ Disconnect the p2p connection. Called by the test logic thread. Causes the p2p connection to be disconnected on the next iteration of the asyncore loop.""" self.disconnect = True # Socket read methods def readable(self): return True def handle_read(self): with mininode_lock: t = self.recv(READ_BUFFER_SIZE) if len(t) > 0: self.recvbuf += t while True: msg = self.got_data() if msg == None: break self.got_message(msg) def got_data(self): try: with mininode_lock: if len(self.recvbuf) < 4: return None if self.recvbuf[:4] != MAGIC_BYTES[self.network]: raise ValueError("got garbage %s" % repr(self.recvbuf)) if len(self.recvbuf) < 4 + 12 + 4 + 4: return command = self.recvbuf[4:4+12].split(b"\x00", 1)[0] msglen = struct.unpack(" 0 or pre_connection) def handle_write(self): with mininode_lock: # asyncore does not expose socket connection, only the first read/write # event, thus we must check connection manually here to know when we # actually connect if self.state == "connecting": self.handle_connect() if not self.writable(): return try: sent = self.send(self.sendbuf) except: self.handle_close() return self.sendbuf = self.sendbuf[sent:] def send_message(self, message, pushbuf=False): if self.state != "connected" and not pushbuf: raise IOError('Not connected, no pushbuf') self._log_message("send", message) command = message.command data = message.serialize() tmsg = MAGIC_BYTES[self.network] tmsg += command tmsg += b"\x00" * (12 - len(command)) tmsg += struct.pack(" 500: log_message += "... (msg truncated)" logger.debug(log_message) # Keep our own socket map for asyncore, so that we can track disconnects # ourselves (to workaround an issue with closing an asyncore socket when # using select) mininode_socket_map = dict() # One lock for synchronizing all data access between the networking thread (see # NetworkThread below) and the thread running the test logic. For simplicity, # NodeConn acquires this lock whenever delivering a message to a NodeConnCB, # and whenever adding anything to the send buffer (in send_message()). This # lock should be acquired in the thread running the test logic to synchronize # access to any data shared with the NodeConnCB or NodeConn. mininode_lock = RLock() class NetworkThread(Thread): def run(self): while mininode_socket_map: # We check for whether to disconnect outside of the asyncore # loop to workaround the behavior of asyncore when using # select disconnected = [] for fd, obj in mininode_socket_map.items(): if obj.disconnect: disconnected.append(obj) [obj.handle_close() for obj in disconnected] asyncore.loop(0.1, use_poll=True, map=mininode_socket_map, count=1) logger.debug("Network thread closing") diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py index 8dc42f227..8dde623fd 100755 --- a/test/functional/test_framework/test_node.py +++ b/test/functional/test_framework/test_node.py @@ -1,265 +1,265 @@ #!/usr/bin/env python3 # Copyright (c) 2017 The Bitcoin Core developers # Distributed under the MIT software license, see the accompanying # file COPYING or http://www.opensource.org/licenses/mit-license.php. """Class for bitcoind node under test""" import decimal import errno import http.client import json import logging import os import subprocess import time from .authproxy import JSONRPCException from .mininode import COIN, ToHex, FromHex, CTransaction, NodeConn from .util import ( assert_equal, get_rpc_proxy, rpc_url, wait_until, p2p_port, ) BITCOIND_PROC_WAIT_TIMEOUT = 60 class TestNode(): """A class for representing a bitcoind node under test. This class contains: - state about the node (whether it's running, etc) - a Python subprocess.Popen object representing the running process - an RPC connection to the node - one or more P2P connections to the node To make things easier for the test writer, any unrecognised messages will be dispatched to the RPC connection.""" def __init__(self, i, dirname, extra_args, host, rpc_port, p2p_port, timewait, binary, stderr, mocktime, coverage_dir): self.index = i self.datadir = os.path.join(dirname, "node" + str(i)) self.host = host self.rpc_port = rpc_port self.p2p_port = p2p_port self.name = "testnode-{}".format(i) if timewait: self.rpc_timeout = timewait else: # Wait for up to 60 seconds for the RPC server to respond self.rpc_timeout = 60 if binary is None: self.binary = os.getenv("BITCOIND", "bitcoind") else: self.binary = binary self.stderr = stderr self.coverage_dir = coverage_dir # Most callers will just need to add extra args to the standard list below. For those callers that need more flexibity, they can just set the args property directly. self.extra_args = extra_args self.args = [self.binary, "-datadir=" + self.datadir, "-server", "-keypool=1", "-discover=0", "-rest", "-logtimemicros", "-debug", "-debugexclude=libevent", "-debugexclude=leveldb", "-mocktime=" + str(mocktime), "-uacomment=" + self.name] self.cli = TestNodeCLI( os.getenv("BITCOINCLI", "bitcoin-cli"), self.datadir) self.running = False self.process = None self.rpc_connected = False self.rpc = None self.url = None self.relay_fee_cache = None self.log = logging.getLogger('TestFramework.node%d' % i) self.p2ps = [] def __getattr__(self, name): """Dispatches any unrecognised messages to the RPC connection.""" assert self.rpc is not None, "Error: RPC not initialized" assert self.rpc_connected, "Error: No RPC connection" return getattr(self.rpc, name) def start(self, extra_args=None, stderr=None): """Start the node.""" if extra_args is None: extra_args = self.extra_args if stderr is None: stderr = self.stderr self.process = subprocess.Popen(self.args + extra_args, stderr=stderr) self.running = True self.log.debug("bitcoind started, waiting for RPC to come up") def wait_for_rpc_connection(self): """Sets up an RPC connection to the bitcoind process. Returns False if unable to connect.""" # Poll at a rate of four times per second poll_per_s = 4 for _ in range(poll_per_s * self.rpc_timeout): assert self.process.poll( ) is None, "bitcoind exited with status %i during initialization" % self.process.returncode try: self.rpc = get_rpc_proxy(rpc_url(self.datadir, self.host, self.rpc_port), self.index, timeout=self.rpc_timeout, coveragedir=self.coverage_dir) self.rpc.getblockcount() # If the call to getblockcount() succeeds then the RPC connection is up self.rpc_connected = True self.url = self.rpc.url self.log.debug("RPC successfully started") return except IOError as e: if e.errno != errno.ECONNREFUSED: # Port not yet open? raise # unknown IO error except JSONRPCException as e: # Initialization phase if e.error['code'] != -28: # RPC in warmup? raise # unknown JSON RPC exception except ValueError as e: # cookie file not found and no rpcuser or rpcassword. bitcoind still starting if "No RPC credentials" not in str(e): raise time.sleep(1.0 / poll_per_s) raise AssertionError("Unable to connect to bitcoind") def get_wallet_rpc(self, wallet_name): assert self.rpc_connected assert self.rpc wallet_path = "wallet/%s" % wallet_name return self.rpc / wallet_path def stop_node(self): """Stop the node.""" if not self.running: return self.log.debug("Stopping node") try: self.stop() except http.client.CannotSendRequest: self.log.exception("Unable to stop node.") del self.p2ps[:] def is_node_stopped(self): """Checks whether the node has stopped. Returns True if the node has stopped. False otherwise. This method is responsible for freeing resources (self.process).""" if not self.running: return True return_code = self.process.poll() if return_code is None: return False # process has stopped. Assert that it didn't return an error code. assert_equal(return_code, 0) self.running = False self.process = None self.rpc_connected = False self.rpc = None self.log.debug("Node stopped") return True def wait_until_stopped(self, timeout=BITCOIND_PROC_WAIT_TIMEOUT): wait_until(self.is_node_stopped, timeout=timeout) def node_encrypt_wallet(self, passphrase): """"Encrypts the wallet. This causes bitcoind to shutdown, so this method takes care of cleaning up resources.""" self.encryptwallet(passphrase) self.wait_until_stopped() def relay_fee(self, cached=True): if not self.relay_fee_cache or not cached: self.relay_fee_cache = self.getnetworkinfo()["relayfee"] return self.relay_fee_cache def calculate_fee(self, tx): # Relay fee is in satoshis per KB. Thus the 1000, and the COIN added # to get back to an amount of satoshis. billable_size_estimate = tx.billable_size() # Add some padding for signatures # NOTE: Fees must be calculated before signatures are added, # so they will never be included in the billable_size above. billable_size_estimate += len(tx.vin) * 81 return int(self.relay_fee() / 1000 * billable_size_estimate * COIN) def calculate_fee_from_txid(self, txid): ctx = FromHex(CTransaction(), self.getrawtransaction(txid)) return self.calculate_fee(ctx) def add_p2p_connection(self, p2p_conn, **kwargs): """Add a p2p connection to the node. This method adds the p2p connection to the self.p2ps list and also returns the connection to the caller.""" if 'dstport' not in kwargs: kwargs['dstport'] = p2p_port(self.index) if 'dstaddr' not in kwargs: kwargs['dstaddr'] = '127.0.0.1' self.p2ps.append(p2p_conn) - kwargs.update({'rpc': self.rpc, 'callback': p2p_conn}) + kwargs.update({'callback': p2p_conn}) p2p_conn.add_connection(NodeConn(**kwargs)) return p2p_conn @property def p2p(self): """Return the first p2p connection Convenience property - most tests only use a single p2p connection to each node, so this saves having to write node.p2ps[0] many times.""" assert self.p2ps, "No p2p connection" return self.p2ps[0] def disconnect_p2p(self, index=0): """Close the p2p connection to the node.""" # Connection could have already been closed by other end. Calling disconnect_p2p() # on an already disconnected p2p connection is not an error. if self.p2ps[index].connection is not None: self.p2ps[index].connection.disconnect_node() del self.p2ps[index] class TestNodeCLI(): """Interface to bitcoin-cli for an individual node""" def __init__(self, binary, datadir): self.args = [] self.binary = binary self.datadir = datadir self.input = None def __call__(self, *args, input=None): # TestNodeCLI is callable with bitcoin-cli command-line args self.args = [str(arg) for arg in args] self.input = input return self def __getattr__(self, command): def dispatcher(*args, **kwargs): return self.send_cli(command, *args, **kwargs) return dispatcher def send_cli(self, command, *args, **kwargs): """Run bitcoin-cli command. Deserializes returned string as python object.""" pos_args = [str(arg) for arg in args] named_args = [str(key) + "=" + str(value) for (key, value) in kwargs.items()] assert not ( pos_args and named_args), "Cannot use positional arguments and named arguments in the same bitcoin-cli call" p_args = [self.binary, "-datadir=" + self.datadir] + self.args if named_args: p_args += ["-named"] p_args += [command] + pos_args + named_args process = subprocess.Popen(p_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True) cli_stdout, cli_stderr = process.communicate(input=self.input) returncode = process.poll() if returncode: # Ignore cli_stdout, raise with cli_stderr raise subprocess.CalledProcessError( returncode, self.binary, output=cli_stderr) return json.loads(cli_stdout, parse_float=decimal.Decimal)