From 52e15aa4d067fc4ace12c80be5c82e85c04fcfec Mon Sep 17 00:00:00 2001 From: John Newbery Date: Wed, 29 Mar 2017 14:07:39 -0400 Subject: [PATCH] Adds helper functions to NodeConnCB This commit adds some helper functions to NodeConnCB which are useful for many tests: - NodeConnCB now keeps track of the number of each message type that it's received and the most recent message of each type. Many tests assert on the most recent block, tx or reject message. - NodeConnCB now keeps track of its connection state by setting a connected boolean in on_open() and on_close() - NodeConnCB now has wait_for_block, wait_for_getdata, wait_for_getheaders, wait_for_inv and wait_for_verack methods I have updated the individual test cases to make sure that there are no namespace problems that cause them to fail with these new definitions. Future commits will remove the duplicate code. --- test/functional/maxuploadtarget.py | 9 -- test/functional/p2p-acceptblock.py | 11 -- test/functional/p2p-leaktests.py | 8 +- test/functional/p2p-mempool.py | 9 -- test/functional/test_framework/comptool.py | 4 +- test/functional/test_framework/mininode.py | 144 ++++++++++++++------- 6 files changed, 107 insertions(+), 78 deletions(-) diff --git a/test/functional/maxuploadtarget.py b/test/functional/maxuploadtarget.py index 9b42bf276..696933512 100755 --- a/test/functional/maxuploadtarget.py +++ b/test/functional/maxuploadtarget.py @@ -44,15 +44,6 @@ class TestNode(NodeConnCB): except KeyError as e: self.block_receive_map[message.block.sha256] = 1 - # Spin until verack message is received from the node. - # We use this to signal that our test can begin. This - # is called from the testing thread, so it needs to acquire - # the global lock. - def wait_for_verack(self): - def veracked(): - return self.verack_received - return wait_until(veracked, timeout=10) - def wait_for_disconnect(self): def disconnected(): return self.peer_disconnected diff --git a/test/functional/p2p-acceptblock.py b/test/functional/p2p-acceptblock.py index c09945baa..e2aad3f1d 100755 --- a/test/functional/p2p-acceptblock.py +++ b/test/functional/p2p-acceptblock.py @@ -70,17 +70,6 @@ class TestNode(NodeConnCB): def on_getdata(self, conn, message): self.last_getdata = message - # Spin until verack message is received from the node. - # We use this to signal that our test can begin. This - # is called from the testing thread, so it needs to acquire - # the global lock. - def wait_for_verack(self): - while True: - with mininode_lock: - if self.verack_received: - return - time.sleep(0.05) - # Wrapper for the NodeConn's send_message function def send_message(self, message): self.connection.send_message(message) diff --git a/test/functional/p2p-leaktests.py b/test/functional/p2p-leaktests.py index 5853ec86f..2423d4a96 100755 --- a/test/functional/p2p-leaktests.py +++ b/test/functional/p2p-leaktests.py @@ -23,6 +23,7 @@ class CLazyNode(NodeConnCB): self.connection = None self.unexpected_msg = False self.connected = False + self.ever_connected = False def add_connection(self, conn): self.connection = conn @@ -36,6 +37,7 @@ class CLazyNode(NodeConnCB): def on_open(self, conn): self.connected = True + self.ever_connected = True def on_version(self, conn, message): self.bad_message(message) def on_verack(self, conn, message): self.bad_message(message) @@ -121,7 +123,9 @@ class P2PLeakTest(BitcoinTestFramework): NetworkThread().start() # Start up network handling in another thread - assert(wait_until(lambda: no_version_bannode.connected and no_version_idlenode.connected and no_verack_idlenode.version_received, timeout=10)) + assert wait_until(lambda: no_version_bannode.ever_connected, timeout=10) + assert wait_until(lambda: no_version_idlenode.ever_connected, timeout=10) + assert wait_until(lambda: no_verack_idlenode.version_received, timeout=10) # Mine a block and make sure that it's not sent to the connected nodes self.nodes[0].generate(1) @@ -130,7 +134,7 @@ class P2PLeakTest(BitcoinTestFramework): time.sleep(5) #This node should have been banned - assert(no_version_bannode.connection.state == "closed") + assert not no_version_bannode.connected [conn.disconnect_node() for conn in connections] diff --git a/test/functional/p2p-mempool.py b/test/functional/p2p-mempool.py index 5064ce74a..c17af173c 100755 --- a/test/functional/p2p-mempool.py +++ b/test/functional/p2p-mempool.py @@ -38,15 +38,6 @@ class TestNode(NodeConnCB): except KeyError as e: self.block_receive_map[message.block.sha256] = 1 - # Spin until verack message is received from the node. - # We use this to signal that our test can begin. This - # is called from the testing thread, so it needs to acquire - # the global lock. - def wait_for_verack(self): - def veracked(): - return self.verack_received - return wait_until(veracked, timeout=10) - def wait_for_disconnect(self): def disconnected(): return self.peer_disconnected diff --git a/test/functional/test_framework/comptool.py b/test/functional/test_framework/comptool.py index 25c18bda8..cb9a938f9 100755 --- a/test/functional/test_framework/comptool.py +++ b/test/functional/test_framework/comptool.py @@ -192,9 +192,7 @@ class TestManager(object): return wait_until(disconnected, timeout=10) def wait_for_verack(self): - def veracked(): - return all(node.verack_received for node in self.test_nodes) - return wait_until(veracked, timeout=10) + [node.wait_for_verack() for node in self.test_nodes] def wait_for_pings(self, counter): def received_pongs(): diff --git a/test/functional/test_framework/mininode.py b/test/functional/test_framework/mininode.py index d57d46f2f..3b4d05df6 100755 --- a/test/functional/test_framework/mininode.py +++ b/test/functional/test_framework/mininode.py @@ -20,21 +20,22 @@ msg_block, msg_tx, msg_headers, etc.: ser_*, deser_*: functions that handle serialization/deserialization """ -import struct -import socket import asyncore -import time -import sys -import random -from .util import hex_str_to_bytes, bytes_to_hex_str -from io import BytesIO from codecs import encode +from collections import defaultdict +import copy import hashlib -from threading import RLock -from threading import Thread +from io import BytesIO import logging -import copy +import random +import socket +import struct +import sys +import time +from threading import RLock, Thread + from test_framework.siphash import siphash256 +from test_framework.util import hex_str_to_bytes, bytes_to_hex_str BIP0031_VERSION = 60000 MY_VERSION = 70014 # past bip-31 for ping/pong @@ -1465,30 +1466,57 @@ class msg_witness_blocktxn(msg_blocktxn): r += self.block_transactions.serialize(with_witness=True) return r -# This is what a callback should look like for NodeConn -# Reimplement the on_* functions to provide handling for events class NodeConnCB(object): + """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): - self.verack_received = False + # 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 + # deliver_sleep_time is helpful for debugging race conditions in p2p # tests; it causes message delivery to sleep for the specified time # before acquiring the global lock and delivering the next message. self.deliver_sleep_time = None + # Remember the services our peer has advertised self.peer_services = None - self.connection = None - self.ping_counter = 1 - self.last_pong = msg_pong() + + # 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. + + Optionally waits for deliver_sleep_time before dispatching message. + """ + deliver_sleep = self.get_deliver_sleep_time() if deliver_sleep is not None: time.sleep(deliver_sleep) with mininode_lock: try: - getattr(self, 'on_' + message.command.decode('ascii'))(conn, message) + command = message.command.decode('ascii') + self.message_count[command] += 1 + self.last_message[command] = message + getattr(self, 'on_' + command)(conn, message) except: - logger.exception("ERROR delivering %s" % repr(message)) + print("ERROR delivering %s (%s)" % (repr(message), + sys.exc_info()[0])) def set_deliver_sleep_time(self, value): with mininode_lock: @@ -1498,14 +1526,20 @@ class NodeConnCB(object): with mininode_lock: return self.deliver_sleep_time - # Callbacks which can be overridden by subclasses - ################################################# + # 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_alert(self, conn, message): pass def on_block(self, conn, message): pass def on_blocktxn(self, conn, message): pass - def on_close(self, conn): pass def on_cmpctblock(self, conn, message): pass def on_feefilter(self, conn, message): pass def on_getaddr(self, conn, message): pass @@ -1515,7 +1549,7 @@ class NodeConnCB(object): def on_getheaders(self, conn, message): pass def on_headers(self, conn, message): pass def on_mempool(self, conn): pass - def on_open(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 @@ -1533,9 +1567,6 @@ class NodeConnCB(object): if conn.ver_send > BIP0031_VERSION: conn.send_message(msg_pong(message.nonce)) - def on_pong(self, conn, message): - self.last_pong = message - def on_verack(self, conn, message): conn.ver_recv = conn.ver_send self.verack_received = True @@ -1548,15 +1579,53 @@ class NodeConnCB(object): conn.ver_recv = conn.ver_send conn.nServices = message.nServices - # Helper functions - ################## + # Connection helper methods def add_connection(self, conn): self.connection = conn - # Wrapper for the NodeConn's send_message function + def wait_for_disconnect(self, timeout=60): + test_function = lambda: not self.connected + assert wait_until(test_function, timeout=timeout) + + # Message receiving helper methods + + def sync(self, test_function, timeout=60): + while timeout > 0: + with mininode_lock: + if test_function(): + return + time.sleep(0.05) + timeout -= 0.05 + raise AssertionError("Sync failed to complete") + + def wait_for_block(self, blockhash, timeout=60): + test_function = lambda: self.last_message.get("block") and self.last_message["block"].block.rehash() == blockhash + self.sync(test_function, timeout) + + def wait_for_getdata(self, timeout=60): + test_function = lambda: self.last_message.get("getdata") + self.sync(test_function, timeout) + + def wait_for_getheaders(self, timeout=60): + test_function = lambda: self.last_message.get("getheaders") + self.sync(test_function, timeout) + + def wait_for_inv(self, expected_inv, timeout=60): + test_function = lambda: self.last_message.get("inv") and self.last_message["inv"] != expected_inv + self.sync(test_function, timeout) + + def wait_for_verack(self, timeout=60): + test_function = lambda: self.message_count["verack"] + self.sync(test_function, timeout) + + # Message sending helper functions + def send_message(self, message): - self.connection.send_message(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) @@ -1564,28 +1633,15 @@ class NodeConnCB(object): # Sync up with the node def sync_with_ping(self, timeout=60): - def received_pong(): - return (self.last_pong.nonce == self.ping_counter) self.send_message(msg_ping(nonce=self.ping_counter)) - success = wait_until(received_pong, timeout=timeout) + test_function = lambda: self.last_message.get("pong") and self.last_message["pong"].nonce == self.ping_counter + success = wait_until(test_function, timeout = timeout) if not success: logger.error("sync_with_ping failed!") raise AssertionError("sync_with_ping failed!") self.ping_counter += 1 - return success - # Spin until verack message is received from the node. - # Tests may want to use this as a signal that the test can begin. - # This can be called from the testing thread, so it needs to acquire the - # global lock. - def wait_for_verack(self): - while True: - with mininode_lock: - if self.verack_received: - return - time.sleep(0.05) - # The actual NodeConn class # This class provides an interface for a p2p connection to a specified node class NodeConn(asyncore.dispatcher):