diff --git a/test/functional/assumevalid.py b/test/functional/assumevalid.py
index 36761d359e291..72da955641b26 100755
--- a/test/functional/assumevalid.py
+++ b/test/functional/assumevalid.py
@@ -67,7 +67,7 @@ def setup_network(self):
     def send_blocks_until_disconnected(self, p2p_conn):
         """Keep sending blocks to the node until we're disconnected."""
         for i in range(len(self.blocks)):
-            if not p2p_conn.connection:
+            if p2p_conn.state != "connected":
                 break
             try:
                 p2p_conn.send_message(msg_block(self.blocks[i]))
diff --git a/test/functional/bip9-softforks.py b/test/functional/bip9-softforks.py
index 904789301a108..ec4d1d9365296 100755
--- a/test/functional/bip9-softforks.py
+++ b/test/functional/bip9-softforks.py
@@ -246,7 +246,7 @@ def test_BIP(self, bipName, activated_version, invalidate, invalidatePostSignatu
         self.setup_network()
         self.test.add_all_connections(self.nodes)
         NetworkThread().start()
-        self.test.test_nodes[0].wait_for_verack()
+        self.test.p2p_connections[0].wait_for_verack()
 
     def get_tests(self):
         for test in itertools.chain(
diff --git a/test/functional/example_test.py b/test/functional/example_test.py
index ba40f33016fb5..54cabee51c24f 100755
--- a/test/functional/example_test.py
+++ b/test/functional/example_test.py
@@ -49,14 +49,14 @@ def __init__(self):
         # Stores a dictionary of all blocks received
         self.block_receive_map = defaultdict(int)
 
-    def on_block(self, conn, message):
+    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, conn, message):
+    def on_inv(self, message):
         """Override the standard on_inv callback"""
         pass
 
diff --git a/test/functional/maxuploadtarget.py b/test/functional/maxuploadtarget.py
index 88e2ff2e1614c..beb0d599d2306 100755
--- a/test/functional/maxuploadtarget.py
+++ b/test/functional/maxuploadtarget.py
@@ -22,10 +22,10 @@ def __init__(self):
         super().__init__()
         self.block_receive_map = defaultdict(int)
 
-    def on_inv(self, conn, message):
+    def on_inv(self, message):
         pass
 
-    def on_block(self, conn, message):
+    def on_block(self, message):
         message.block.calc_sha256()
         self.block_receive_map[message.block.sha256] += 1
 
diff --git a/test/functional/p2p-compactblocks.py b/test/functional/p2p-compactblocks.py
index d2c4d39305be6..5aba1db8730d9 100755
--- a/test/functional/p2p-compactblocks.py
+++ b/test/functional/p2p-compactblocks.py
@@ -25,21 +25,21 @@ def __init__(self):
         # so we can eg wait until a particular block is announced.
         self.announced_blockhashes = set()
 
-    def on_sendcmpct(self, conn, message):
+    def on_sendcmpct(self, message):
         self.last_sendcmpct.append(message)
 
-    def on_cmpctblock(self, conn, message):
+    def on_cmpctblock(self, message):
         self.block_announced = True
         self.last_message["cmpctblock"].header_and_shortids.header.calc_sha256()
         self.announced_blockhashes.add(self.last_message["cmpctblock"].header_and_shortids.header.sha256)
 
-    def on_headers(self, conn, message):
+    def on_headers(self, message):
         self.block_announced = True
         for x in self.last_message["headers"].headers:
             x.calc_sha256()
             self.announced_blockhashes.add(x.sha256)
 
-    def on_inv(self, conn, message):
+    def on_inv(self, message):
         for x in self.last_message["inv"].inv:
             if x.type == 2:
                 self.block_announced = True
@@ -60,7 +60,7 @@ def get_headers(self, locator, hashstop):
         msg = msg_getheaders()
         msg.locator.vHave = locator
         msg.hashstop = hashstop
-        self.connection.send_message(msg)
+        self.send_message(msg)
 
     def send_header_for_blocks(self, new_blocks):
         headers_message = msg_headers()
@@ -86,7 +86,7 @@ def send_await_disconnect(self, message, timeout=30):
         This is used when we want to send a message into the node that we expect
         will get us disconnected, eg an invalid block."""
         self.send_message(message)
-        wait_until(lambda: not self.connected, timeout=timeout, lock=mininode_lock)
+        wait_until(lambda: self.state != "connected", timeout=timeout, lock=mininode_lock)
 
 class CompactBlocksTest(BitcoinTestFramework):
     def set_test_params(self):
diff --git a/test/functional/p2p-feefilter.py b/test/functional/p2p-feefilter.py
index 624278df40db7..0ce3c3f42938d 100755
--- a/test/functional/p2p-feefilter.py
+++ b/test/functional/p2p-feefilter.py
@@ -27,7 +27,7 @@ def __init__(self):
         super().__init__()
         self.txinvs = []
 
-    def on_inv(self, conn, message):
+    def on_inv(self, message):
         for i in message.inv:
             if (i.type == 1):
                 self.txinvs.append(hashToHex(i.hash))
diff --git a/test/functional/p2p-leaktests.py b/test/functional/p2p-leaktests.py
index 719a03914d086..3634f3836dc75 100755
--- a/test/functional/p2p-leaktests.py
+++ b/test/functional/p2p-leaktests.py
@@ -30,43 +30,42 @@ def bad_message(self, message):
         self.unexpected_msg = True
         self.log.info("should not have received message: %s" % message.command)
 
-    def on_open(self, conn):
-        self.connected = True
+    def on_open(self):
         self.ever_connected = True
 
-    def on_version(self, conn, message): self.bad_message(message)
-    def on_verack(self, conn, message): self.bad_message(message)
-    def on_reject(self, conn, message): self.bad_message(message)
-    def on_inv(self, conn, message): self.bad_message(message)
-    def on_addr(self, conn, message): self.bad_message(message)
-    def on_getdata(self, conn, message): self.bad_message(message)
-    def on_getblocks(self, conn, message): self.bad_message(message)
-    def on_tx(self, conn, message): self.bad_message(message)
-    def on_block(self, conn, message): self.bad_message(message)
-    def on_getaddr(self, conn, message): self.bad_message(message)
-    def on_headers(self, conn, message): self.bad_message(message)
-    def on_getheaders(self, conn, message): self.bad_message(message)
-    def on_ping(self, conn, message): self.bad_message(message)
-    def on_mempool(self, conn): self.bad_message(message)
-    def on_pong(self, conn, message): self.bad_message(message)
-    def on_feefilter(self, conn, message): self.bad_message(message)
-    def on_sendheaders(self, conn, message): self.bad_message(message)
-    def on_sendcmpct(self, conn, message): self.bad_message(message)
-    def on_cmpctblock(self, conn, message): self.bad_message(message)
-    def on_getblocktxn(self, conn, message): self.bad_message(message)
-    def on_blocktxn(self, conn, message): self.bad_message(message)
+    def on_version(self, message): self.bad_message(message)
+    def on_verack(self, message): self.bad_message(message)
+    def on_reject(self, message): self.bad_message(message)
+    def on_inv(self, message): self.bad_message(message)
+    def on_addr(self, message): self.bad_message(message)
+    def on_getdata(self, message): self.bad_message(message)
+    def on_getblocks(self, message): self.bad_message(message)
+    def on_tx(self, message): self.bad_message(message)
+    def on_block(self, message): self.bad_message(message)
+    def on_getaddr(self, message): self.bad_message(message)
+    def on_headers(self, message): self.bad_message(message)
+    def on_getheaders(self, message): self.bad_message(message)
+    def on_ping(self, message): self.bad_message(message)
+    def on_mempool(self, message): self.bad_message(message)
+    def on_pong(self, message): self.bad_message(message)
+    def on_feefilter(self, message): self.bad_message(message)
+    def on_sendheaders(self, message): self.bad_message(message)
+    def on_sendcmpct(self, message): self.bad_message(message)
+    def on_cmpctblock(self, message): self.bad_message(message)
+    def on_getblocktxn(self, message): self.bad_message(message)
+    def on_blocktxn(self, message): self.bad_message(message)
 
 # Node that never sends a version. We'll use this to send a bunch of messages
 # anyway, and eventually get disconnected.
 class CNodeNoVersionBan(CLazyNode):
     # send a bunch of veracks without sending a message. This should get us disconnected.
     # NOTE: implementation-specific check here. Remove if bitcoind ban behavior changes
-    def on_open(self, conn):
-        super().on_open(conn)
+    def on_open(self):
+        super().on_open()
         for i in range(banscore):
             self.send_message(msg_verack())
 
-    def on_reject(self, conn, message): pass
+    def on_reject(self, message): pass
 
 # Node that never sends a version. This one just sits idle and hopes to receive
 # any message (it shouldn't!)
@@ -80,15 +79,15 @@ def __init__(self):
         self.version_received = False
         super().__init__()
 
-    def on_reject(self, conn, message): pass
-    def on_verack(self, conn, message): pass
+    def on_reject(self, message): pass
+    def on_verack(self, message): pass
     # When version is received, don't reply with a verack. Instead, see if the
     # node will give us a message that it shouldn't. This is not an exhaustive
     # list!
-    def on_version(self, conn, message):
+    def on_version(self, message):
         self.version_received = True
-        conn.send_message(msg_ping())
-        conn.send_message(msg_getaddr())
+        self.send_message(msg_ping())
+        self.send_message(msg_getaddr())
 
 class P2PLeakTest(BitcoinTestFramework):
     def set_test_params(self):
@@ -119,11 +118,11 @@ def run_test(self):
         time.sleep(5)
 
         #This node should have been banned
-        assert not no_version_bannode.connected
+        assert no_version_bannode.state != "connected"
 
         # These nodes should have been disconnected
-        assert not unsupported_service_bit5_node.connected
-        assert not unsupported_service_bit7_node.connected
+        assert unsupported_service_bit5_node.state != "connected"
+        assert unsupported_service_bit7_node.state != "connected"
 
         self.nodes[0].disconnect_p2ps()
 
diff --git a/test/functional/p2p-segwit.py b/test/functional/p2p-segwit.py
index b940bc40966a8..5776d99f6e1bd 100755
--- a/test/functional/p2p-segwit.py
+++ b/test/functional/p2p-segwit.py
@@ -31,13 +31,40 @@ def get_virtual_size(witness_block):
     vsize = int((3*base_size + total_size + 3)/4)
     return vsize
 
+def test_transaction_acceptance(rpc, p2p, tx, with_witness, accepted, reason=None):
+    """Send a transaction to the node and check that it's accepted to the mempool
+
+    - Submit the transaction over the p2p interface
+    - use the getrawmempool rpc to check for acceptance."""
+    tx_message = msg_tx(tx)
+    if with_witness:
+        tx_message = msg_witness_tx(tx)
+    p2p.send_message(tx_message)
+    p2p.sync_with_ping()
+    assert_equal(tx.hash in rpc.getrawmempool(), accepted)
+    if (reason != None and not accepted):
+        # Check the rejection reason as well.
+        with mininode_lock:
+            assert_equal(p2p.last_message["reject"].reason, reason)
+
+def test_witness_block(rpc, p2p, block, accepted, with_witness=True):
+    """Send a block to the node and check that it's accepted
+
+    - Submit the block over the p2p interface
+    - use the getbestblockhash rpc to check for acceptance."""
+    if with_witness:
+        p2p.send_message(msg_witness_block(block))
+    else:
+        p2p.send_message(msg_block(block))
+    p2p.sync_with_ping()
+    assert_equal(rpc.getbestblockhash() == block.hash, accepted)
+
 class TestNode(NodeConnCB):
-    def __init__(self, rpc):
+    def __init__(self):
         super().__init__()
         self.getdataset = set()
-        self.rpc = rpc
 
-    def on_getdata(self, conn, message):
+    def on_getdata(self, message):
         for inv in message.inv:
             self.getdataset.add(inv.hash)
 
@@ -68,27 +95,6 @@ def request_block(self, blockhash, inv_type, timeout=60):
         self.wait_for_block(blockhash, timeout)
         return self.last_message["block"].block
 
-    def test_transaction_acceptance(self, tx, with_witness, accepted, reason=None):
-        tx_message = msg_tx(tx)
-        if with_witness:
-            tx_message = msg_witness_tx(tx)
-        self.send_message(tx_message)
-        self.sync_with_ping()
-        assert_equal(tx.hash in self.rpc.getrawmempool(), accepted)
-        if (reason != None and not accepted):
-            # Check the rejection reason as well.
-            with mininode_lock:
-                assert_equal(self.last_message["reject"].reason, reason)
-
-    # Test whether a witness block had the correct effect on the tip
-    def test_witness_block(self, block, accepted, with_witness=True):
-        if with_witness:
-            self.send_message(msg_witness_block(block))
-        else:
-            self.send_message(msg_block(block))
-        self.sync_with_ping()
-        assert_equal(self.rpc.getbestblockhash() == block.hash, accepted)
-
 # Used to keep track of anyone-can-spend outputs that we can use in the tests
 class UTXO():
     def __init__(self, sha256, n, nValue):
@@ -142,7 +148,7 @@ def update_witness_block_with_transactions(self, block, tx_list, nonce=0):
     ''' Individual tests '''
     def test_witness_services(self):
         self.log.info("Verifying NODE_WITNESS service bit")
-        assert((self.test_node.connection.nServices & NODE_WITNESS) != 0)
+        assert((self.test_node.nServices & NODE_WITNESS) != 0)
 
 
     # See if sending a regular transaction works, and create a utxo
@@ -201,7 +207,7 @@ def test_unnecessary_witness_before_segwit_activation(self):
         self.update_witness_block_with_transactions(block, [tx])
         # Sending witness data before activation is not allowed (anti-spam
         # rule).
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
         # TODO: fix synchronization so we can test reject reason
         # Right now, bitcoind delays sending reject messages for blocks
         # until the future, making synchronization here difficult.
@@ -228,7 +234,7 @@ def test_unnecessary_witness_before_segwit_activation(self):
         tx2.vin.append(CTxIn(COutPoint(tx.sha256, 0), b""))
         tx2.vout.append(CTxOut(tx.vout[0].nValue-1000, scriptPubKey))
         tx2.rehash()
-        self.test_node.test_transaction_acceptance(tx2, False, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx2, False, True)
         self.nodes[0].generate(1)
         sync_blocks(self.nodes)
 
@@ -245,18 +251,18 @@ def test_unnecessary_witness_before_segwit_activation(self):
         tx3.rehash()
         # Note that this should be rejected for the premature witness reason,
         # rather than a policy check, since segwit hasn't activated yet.
-        self.std_node.test_transaction_acceptance(tx3, True, False, b'no-witness-yet')
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx3, True, False, b'no-witness-yet')
 
         # If we send without witness, it should be accepted.
-        self.std_node.test_transaction_acceptance(tx3, False, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx3, False, True)
 
         # Now create a new anyone-can-spend utxo for the next test.
         tx4 = CTransaction()
         tx4.vin.append(CTxIn(COutPoint(tx3.sha256, 0), CScript([p2sh_program])))
         tx4.vout.append(CTxOut(tx3.vout[0].nValue-1000, CScript([OP_TRUE])))
         tx4.rehash()
-        self.test_node.test_transaction_acceptance(tx3, False, True)
-        self.test_node.test_transaction_acceptance(tx4, False, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, False, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx4, False, True)
 
         self.nodes[0].generate(1)
         sync_blocks(self.nodes)
@@ -317,7 +323,7 @@ def test_witness_commitments(self):
         assert(msg_witness_block(block).serialize() != msg_block(block).serialize())
 
         # This empty block should be valid.
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Try to tweak the nonce
         block_2 = self.build_next_block()
@@ -328,7 +334,7 @@ def test_witness_commitments(self):
         assert(block_2.vtx[0].vout[-1] != block.vtx[0].vout[-1])
 
         # This should also be valid.
-        self.test_node.test_witness_block(block_2, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_2, accepted=True)
 
         # Now test commitments with actual transactions
         assert (len(self.utxo) > 0)
@@ -361,7 +367,7 @@ def test_witness_commitments(self):
         block_3.rehash()
         block_3.solve()
 
-        self.test_node.test_witness_block(block_3, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_3, accepted=False)
 
         # Add a different commitment with different nonce, but in the
         # right location, and with some funds burned(!).
@@ -375,7 +381,7 @@ def test_witness_commitments(self):
         block_3.rehash()
         assert(len(block_3.vtx[0].vout) == 4) # 3 OP_returns
         block_3.solve()
-        self.test_node.test_witness_block(block_3, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_3, accepted=True)
 
         # Finally test that a block with no witness transactions can
         # omit the commitment.
@@ -387,7 +393,7 @@ def test_witness_commitments(self):
         block_4.vtx.append(tx3)
         block_4.hashMerkleRoot = block_4.calc_merkle_root()
         block_4.solve()
-        self.test_node.test_witness_block(block_4, with_witness=False, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_4, with_witness=False, accepted=True)
 
         # Update available utxo's for use in later test.
         self.utxo.pop(0)
@@ -428,11 +434,11 @@ def test_block_malleability(self):
         # Change the nonce -- should not cause the block to be permanently
         # failed
         block.vtx[0].wit.vtxinwit[0].scriptWitness.stack = [ ser_uint256(1) ]
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Changing the witness nonce doesn't change the block hash
         block.vtx[0].wit.vtxinwit[0].scriptWitness.stack = [ ser_uint256(0) ]
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
 
     def test_witness_block_size(self):
@@ -497,7 +503,7 @@ def test_witness_block_size(self):
         # limit
         assert(len(block.serialize(True)) > 2*1024*1024)
 
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Now resize the second transaction to make the block fit.
         cur_length = len(block.vtx[-1].wit.vtxinwit[0].scriptWitness.stack[0])
@@ -507,7 +513,7 @@ def test_witness_block_size(self):
         block.solve()
         assert(get_virtual_size(block) == MAX_BLOCK_BASE_SIZE)
 
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Update available utxo's
         self.utxo.pop(0)
@@ -574,7 +580,7 @@ def test_extra_witness_data(self):
         self.update_witness_block_with_transactions(block, [tx])
 
         # Extra witness data should not be allowed.
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Try extra signature data.  Ok if we're not spending a witness output.
         block.vtx[1].wit.vtxinwit = []
@@ -583,7 +589,7 @@ def test_extra_witness_data(self):
         add_witness_commitment(block)
         block.solve()
 
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Now try extra witness/signature data on an input that DOES require a
         # witness
@@ -599,7 +605,7 @@ def test_extra_witness_data(self):
         self.update_witness_block_with_transactions(block, [tx2])
 
         # This has extra witness data, so it should fail.
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Now get rid of the extra witness, but add extra scriptSig data
         tx2.vin[0].scriptSig = CScript([OP_TRUE])
@@ -611,7 +617,7 @@ def test_extra_witness_data(self):
         block.solve()
 
         # This has extra signature data for a witness input, so it should fail.
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Now get rid of the extra scriptsig on the witness input, and verify
         # success (even with extra scriptsig data in the non-witness input)
@@ -620,7 +626,7 @@ def test_extra_witness_data(self):
         add_witness_commitment(block)
         block.solve()
 
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Update utxo for later tests
         self.utxo.pop(0)
@@ -653,14 +659,14 @@ def test_max_witness_push_length(self):
         tx2.rehash()
 
         self.update_witness_block_with_transactions(block, [tx, tx2])
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Now reduce the length of the stack element
         tx2.wit.vtxinwit[0].scriptWitness.stack[0] = b'a'*(MAX_SCRIPT_ELEMENT_SIZE)
 
         add_witness_commitment(block)
         block.solve()
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Update the utxo for later tests
         self.utxo.pop()
@@ -695,7 +701,7 @@ def test_max_witness_program_length(self):
 
         self.update_witness_block_with_transactions(block, [tx, tx2])
 
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Try again with one less byte in the witness program
         witness_program = CScript([b'a'*520]*19 + [OP_DROP]*62 + [OP_TRUE])
@@ -710,7 +716,7 @@ def test_max_witness_program_length(self):
         tx2.rehash()
         block.vtx = [block.vtx[0]]
         self.update_witness_block_with_transactions(block, [tx, tx2])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         self.utxo.pop()
         self.utxo.append(UTXO(tx2.sha256, 0, tx2.vout[0].nValue))
@@ -736,7 +742,7 @@ def test_witness_input_length(self):
 
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Try various ways to spend tx that should all break.
         # This "broken" transaction serializer will not normalize
@@ -771,7 +777,7 @@ def serialize_with_witness(self):
 
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx2])
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Now try using a too short vtxinwit
         tx2.wit.vtxinwit.pop()
@@ -779,7 +785,7 @@ def serialize_with_witness(self):
 
         block.vtx = [block.vtx[0]]
         self.update_witness_block_with_transactions(block, [tx2])
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Now make one of the intermediate witnesses be incorrect
         tx2.wit.vtxinwit.append(CTxInWitness())
@@ -788,13 +794,13 @@ def serialize_with_witness(self):
 
         block.vtx = [block.vtx[0]]
         self.update_witness_block_with_transactions(block, [tx2])
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Fix the broken witness and the block should be accepted.
         tx2.wit.vtxinwit[5].scriptWitness.stack = [b'a', witness_program]
         block.vtx = [block.vtx[0]]
         self.update_witness_block_with_transactions(block, [tx2])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         self.utxo.pop()
         self.utxo.append(UTXO(tx2.sha256, 0, tx2.vout[0].nValue))
@@ -834,11 +840,11 @@ def test_witness_tx_relay_before_segwit_activation(self):
         # its from)
         assert_equal(len(self.nodes[0].getrawmempool()), 0)
         assert_equal(len(self.nodes[1].getrawmempool()), 0)
-        self.old_node.test_transaction_acceptance(tx, with_witness=True, accepted=False)
-        self.test_node.test_transaction_acceptance(tx, with_witness=True, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.old_node, tx, with_witness=True, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=True, accepted=False)
 
         # But eliminating the witness should fix it
-        self.test_node.test_transaction_acceptance(tx, with_witness=False, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=False, accepted=True)
 
         # Cleanup: mine the first transaction and update utxo
         self.nodes[0].generate(1)
@@ -870,11 +876,11 @@ def test_tx_relay_after_segwit_activation(self):
         # Verify that unnecessary witnesses are rejected.
         self.test_node.announce_tx_and_wait_for_getdata(tx)
         assert_equal(len(self.nodes[0].getrawmempool()), 0)
-        self.test_node.test_transaction_acceptance(tx, with_witness=True, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=True, accepted=False)
 
         # Verify that removing the witness succeeds.
         self.test_node.announce_tx_and_wait_for_getdata(tx)
-        self.test_node.test_transaction_acceptance(tx, with_witness=False, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=False, accepted=True)
 
         # Now try to add extra witness data to a valid witness tx.
         witness_program = CScript([OP_TRUE])
@@ -899,24 +905,24 @@ def test_tx_relay_after_segwit_activation(self):
 
         # Node will not be blinded to the transaction
         self.std_node.announce_tx_and_wait_for_getdata(tx3)
-        self.std_node.test_transaction_acceptance(tx3, True, False, b'tx-size')
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx3, True, False, b'tx-size')
         self.std_node.announce_tx_and_wait_for_getdata(tx3)
-        self.std_node.test_transaction_acceptance(tx3, True, False, b'tx-size')
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx3, True, False, b'tx-size')
 
         # Remove witness stuffing, instead add extra witness push on stack
         tx3.vout[0] = CTxOut(tx2.vout[0].nValue-1000, CScript([OP_TRUE]))
         tx3.wit.vtxinwit[0].scriptWitness.stack = [CScript([CScriptNum(1)]), witness_program ]
         tx3.rehash()
 
-        self.test_node.test_transaction_acceptance(tx2, with_witness=True, accepted=True)
-        self.test_node.test_transaction_acceptance(tx3, with_witness=True, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx2, with_witness=True, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, with_witness=True, accepted=False)
 
         # Get rid of the extra witness, and verify acceptance.
         tx3.wit.vtxinwit[0].scriptWitness.stack = [ witness_program ]
         # Also check that old_node gets a tx announcement, even though this is
         # a witness transaction.
         self.old_node.wait_for_inv([CInv(1, tx2.sha256)]) # wait until tx2 was inv'ed
-        self.test_node.test_transaction_acceptance(tx3, with_witness=True, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, with_witness=True, accepted=True)
         self.old_node.wait_for_inv([CInv(1, tx3.sha256)])
 
         # Test that getrawtransaction returns correct witness information
@@ -955,20 +961,20 @@ def test_block_relay(self, segwit_activated):
 
         self.test_node.announce_block_and_wait_for_getdata(block1, use_header=False)
         assert(self.test_node.last_message["getdata"].inv[0].type == blocktype)
-        self.test_node.test_witness_block(block1, True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block1, True)
 
         block2 = self.build_next_block(nVersion=4)
         block2.solve()
 
         self.test_node.announce_block_and_wait_for_getdata(block2, use_header=True)
         assert(self.test_node.last_message["getdata"].inv[0].type == blocktype)
-        self.test_node.test_witness_block(block2, True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block2, True)
 
         block3 = self.build_next_block(nVersion=(VB_TOP_BITS | (1<<15)))
         block3.solve()
         self.test_node.announce_block_and_wait_for_getdata(block3, use_header=True)
         assert(self.test_node.last_message["getdata"].inv[0].type == blocktype)
-        self.test_node.test_witness_block(block3, True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block3, True)
 
         # Check that we can getdata for witness blocks or regular blocks,
         # and the right thing happens.
@@ -998,7 +1004,7 @@ def test_block_relay(self, segwit_activated):
             # This gives us a witness commitment.
             assert(len(block.vtx[0].wit.vtxinwit) == 1)
             assert(len(block.vtx[0].wit.vtxinwit[0].scriptWitness.stack) == 1)
-            self.test_node.test_witness_block(block, accepted=True)
+            test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
             # Now try to retrieve it...
             rpc_block = self.nodes[0].getblock(block.hash, False)
             non_wit_block = self.test_node.request_block(block.sha256, 2)
@@ -1052,7 +1058,7 @@ def test_standardness_v0(self, segwit_activated):
         p2sh_tx.rehash()
 
         # Mine it on test_node to create the confirmed output.
-        self.test_node.test_transaction_acceptance(p2sh_tx, with_witness=True, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2sh_tx, with_witness=True, accepted=True)
         self.nodes[0].generate(1)
         sync_blocks(self.nodes)
 
@@ -1064,7 +1070,7 @@ def test_standardness_v0(self, segwit_activated):
         tx.vout.append(CTxOut(8000, scriptPubKey)) # Might burn this later
         tx.rehash()
 
-        self.std_node.test_transaction_acceptance(tx, with_witness=True, accepted=segwit_activated)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx, with_witness=True, accepted=segwit_activated)
 
         # Now create something that looks like a P2PKH output. This won't be spendable.
         scriptPubKey = CScript([OP_0, hash160(witness_hash)])
@@ -1081,7 +1087,7 @@ def test_standardness_v0(self, segwit_activated):
             tx2.vout = [CTxOut(p2sh_tx.vout[0].nValue-1000, scriptPubKey)]
         tx2.rehash()
 
-        self.std_node.test_transaction_acceptance(tx2, with_witness=True, accepted=segwit_activated)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx2, with_witness=True, accepted=segwit_activated)
 
         # Now update self.utxo for later tests.
         tx3 = CTransaction()
@@ -1094,13 +1100,13 @@ def test_standardness_v0(self, segwit_activated):
             tx3.wit.vtxinwit.append(CTxInWitness())
             tx3.wit.vtxinwit[0].scriptWitness.stack = [witness_program]
             tx3.rehash()
-            self.test_node.test_transaction_acceptance(tx3, with_witness=True, accepted=True)
+            test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, with_witness=True, accepted=True)
         else:
             # tx and tx2 didn't go anywhere; just clean up the p2sh_tx output.
             tx3.vin = [CTxIn(COutPoint(p2sh_tx.sha256, 0), CScript([witness_program]))]
             tx3.vout = [CTxOut(p2sh_tx.vout[0].nValue-1000, witness_program)]
             tx3.rehash()
-            self.test_node.test_transaction_acceptance(tx3, with_witness=True, accepted=True)
+            test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, with_witness=True, accepted=True)
 
         self.nodes[0].generate(1)
         sync_blocks(self.nodes)
@@ -1124,7 +1130,7 @@ def test_segwit_versions(self):
             tx.rehash()
             block = self.build_next_block()
             self.update_witness_block_with_transactions(block, [tx])
-            self.test_node.test_witness_block(block, accepted=True)
+            test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
             self.utxo.pop(0)
             for i in range(NUM_TESTS):
                 self.utxo.append(UTXO(tx.sha256, i, split_value))
@@ -1143,8 +1149,8 @@ def test_segwit_versions(self):
             tx.vin = [CTxIn(COutPoint(self.utxo[0].sha256, self.utxo[0].n), b"")]
             tx.vout = [CTxOut(self.utxo[0].nValue-1000, scriptPubKey)]
             tx.rehash()
-            self.std_node.test_transaction_acceptance(tx, with_witness=True, accepted=False)
-            self.test_node.test_transaction_acceptance(tx, with_witness=True, accepted=True)
+            test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx, with_witness=True, accepted=False)
+            test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=True, accepted=True)
             self.utxo.pop(0)
             temp_utxo.append(UTXO(tx.sha256, 0, tx.vout[0].nValue))
 
@@ -1163,8 +1169,8 @@ def test_segwit_versions(self):
         tx2.rehash()
         # Gets accepted to test_node, because standardness of outputs isn't
         # checked with fRequireStandard
-        self.test_node.test_transaction_acceptance(tx2, with_witness=True, accepted=True)
-        self.std_node.test_transaction_acceptance(tx2, with_witness=True, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx2, with_witness=True, accepted=True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, tx2, with_witness=True, accepted=False)
         temp_utxo.pop() # last entry in temp_utxo was the output we just spent
         temp_utxo.append(UTXO(tx2.sha256, 0, tx2.vout[0].nValue))
 
@@ -1180,7 +1186,7 @@ def test_segwit_versions(self):
         tx3.rehash()
         # Spending a higher version witness output is not allowed by policy,
         # even with fRequireStandard=false.
-        self.test_node.test_transaction_acceptance(tx3, with_witness=True, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, with_witness=True, accepted=False)
         self.test_node.sync_with_ping()
         with mininode_lock:
             assert(b"reserved for soft-fork upgrades" in self.test_node.last_message["reject"].reason)
@@ -1188,7 +1194,7 @@ def test_segwit_versions(self):
         # Building a block with the transaction must be valid, however.
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx2, tx3])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
         sync_blocks(self.nodes)
 
         # Add utxo to our list
@@ -1206,7 +1212,7 @@ def test_premature_coinbase_witness_spend(self):
         # This next line will rehash the coinbase and update the merkle
         # root, and solve.
         self.update_witness_block_with_transactions(block, [])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         spend_tx = CTransaction()
         spend_tx.vin = [CTxIn(COutPoint(block.vtx[0].sha256, 0), b"")]
@@ -1220,13 +1226,13 @@ def test_premature_coinbase_witness_spend(self):
         sync_blocks(self.nodes)
         block2 = self.build_next_block()
         self.update_witness_block_with_transactions(block2, [spend_tx])
-        self.test_node.test_witness_block(block2, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block2, accepted=False)
 
         # Advancing one more block should allow the spend.
         self.nodes[0].generate(1)
         block2 = self.build_next_block()
         self.update_witness_block_with_transactions(block2, [spend_tx])
-        self.test_node.test_witness_block(block2, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block2, accepted=True)
         sync_blocks(self.nodes)
 
 
@@ -1247,11 +1253,11 @@ def test_signature_version_1(self):
         tx.vout.append(CTxOut(self.utxo[0].nValue-1000, scriptPubKey))
         tx.rehash()
 
-        self.test_node.test_transaction_acceptance(tx, with_witness=True, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=True, accepted=True)
         # Mine this transaction in preparation for following tests.
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
         sync_blocks(self.nodes)
         self.utxo.pop(0)
 
@@ -1268,19 +1274,19 @@ def test_signature_version_1(self):
                 # Too-large input value
                 sign_P2PK_witness_input(witness_program, tx, 0, hashtype, prev_utxo.nValue+1, key)
                 self.update_witness_block_with_transactions(block, [tx])
-                self.test_node.test_witness_block(block, accepted=False)
+                test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
                 # Too-small input value
                 sign_P2PK_witness_input(witness_program, tx, 0, hashtype, prev_utxo.nValue-1, key)
                 block.vtx.pop() # remove last tx
                 self.update_witness_block_with_transactions(block, [tx])
-                self.test_node.test_witness_block(block, accepted=False)
+                test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
                 # Now try correct value
                 sign_P2PK_witness_input(witness_program, tx, 0, hashtype, prev_utxo.nValue, key)
                 block.vtx.pop()
                 self.update_witness_block_with_transactions(block, [tx])
-                self.test_node.test_witness_block(block, accepted=True)
+                test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
                 prev_utxo = UTXO(tx.sha256, 0, tx.vout[0].nValue)
 
@@ -1304,7 +1310,7 @@ def test_signature_version_1(self):
 
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         block = self.build_next_block()
         used_sighash_single_out_of_bounds = False
@@ -1346,7 +1352,7 @@ def test_signature_version_1(self):
             # Test the block periodically, if we're close to maxblocksize
             if (get_virtual_size(block) > MAX_BLOCK_BASE_SIZE - 1000):
                 self.update_witness_block_with_transactions(block, [])
-                self.test_node.test_witness_block(block, accepted=True)
+                test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
                 block = self.build_next_block()
 
         if (not used_sighash_single_out_of_bounds):
@@ -1354,7 +1360,7 @@ def test_signature_version_1(self):
         # Test the transactions we've added to the block
         if (len(block.vtx) > 1):
             self.update_witness_block_with_transactions(block, [])
-            self.test_node.test_witness_block(block, accepted=True)
+            test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Now test witness version 0 P2PKH transactions
         pubkeyhash = hash160(pubkey)
@@ -1376,7 +1382,7 @@ def test_signature_version_1(self):
         tx2.vin[0].scriptSig = CScript([signature, pubkey])
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx, tx2])
-        self.test_node.test_witness_block(block, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=False)
 
         # Move the signature to the witness.
         block.vtx.pop()
@@ -1386,7 +1392,7 @@ def test_signature_version_1(self):
         tx2.rehash()
 
         self.update_witness_block_with_transactions(block, [tx2])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         temp_utxos.pop(0)
 
@@ -1405,7 +1411,7 @@ def test_signature_version_1(self):
             index += 1
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         for i in range(len(tx.vout)):
             self.utxo.append(UTXO(tx.sha256, i, tx.vout[i].nValue))
@@ -1432,10 +1438,10 @@ def test_p2sh_witness(self, segwit_activated):
         tx.rehash()
 
         # Verify mempool acceptance and block validity
-        self.test_node.test_transaction_acceptance(tx, with_witness=False, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=False, accepted=True)
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx])
-        self.test_node.test_witness_block(block, accepted=True, with_witness=segwit_activated)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True, with_witness=segwit_activated)
         sync_blocks(self.nodes)
 
         # Now test attempts to spend the output.
@@ -1449,12 +1455,12 @@ def test_p2sh_witness(self, segwit_activated):
         # will require a witness to spend a witness program regardless of
         # segwit activation.  Note that older bitcoind's that are not
         # segwit-aware would also reject this for failing CLEANSTACK.
-        self.test_node.test_transaction_acceptance(spend_tx, with_witness=False, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, spend_tx, with_witness=False, accepted=False)
 
         # Try to put the witness script in the scriptSig, should also fail.
         spend_tx.vin[0].scriptSig = CScript([p2wsh_pubkey, b'a'])
         spend_tx.rehash()
-        self.test_node.test_transaction_acceptance(spend_tx, with_witness=False, accepted=False)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, spend_tx, with_witness=False, accepted=False)
 
         # Now put the witness script in the witness, should succeed after
         # segwit activates.
@@ -1464,7 +1470,7 @@ def test_p2sh_witness(self, segwit_activated):
         spend_tx.wit.vtxinwit[0].scriptWitness.stack = [ b'a', witness_program ]
 
         # Verify mempool acceptance
-        self.test_node.test_transaction_acceptance(spend_tx, with_witness=True, accepted=segwit_activated)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, spend_tx, with_witness=True, accepted=segwit_activated)
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [spend_tx])
 
@@ -1472,9 +1478,9 @@ def test_p2sh_witness(self, segwit_activated):
         # should be valid.  If we're after activation, then sending this with
         # witnesses should be valid.
         if segwit_activated:
-            self.test_node.test_witness_block(block, accepted=True)
+            test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
         else:
-            self.test_node.test_witness_block(block, accepted=True, with_witness=False)
+            test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True, with_witness=False)
 
         # Update self.utxo
         self.utxo.pop(0)
@@ -1558,7 +1564,7 @@ def test_witness_sigops(self):
 
         block_1 = self.build_next_block()
         self.update_witness_block_with_transactions(block_1, [tx])
-        self.test_node.test_witness_block(block_1, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_1, accepted=True)
 
         tx2 = CTransaction()
         # If we try to spend the first n-1 outputs from tx, that should be
@@ -1575,7 +1581,7 @@ def test_witness_sigops(self):
 
         block_2 = self.build_next_block()
         self.update_witness_block_with_transactions(block_2, [tx2])
-        self.test_node.test_witness_block(block_2, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_2, accepted=False)
 
         # Try dropping the last input in tx2, and add an output that has
         # too many sigops (contributing to legacy sigop count).
@@ -1588,14 +1594,14 @@ def test_witness_sigops(self):
         tx2.rehash()
         block_3 = self.build_next_block()
         self.update_witness_block_with_transactions(block_3, [tx2])
-        self.test_node.test_witness_block(block_3, accepted=False)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_3, accepted=False)
 
         # If we drop the last checksig in this output, the tx should succeed.
         block_4 = self.build_next_block()
         tx2.vout[-1].scriptPubKey = CScript([OP_CHECKSIG]*(checksig_count-1))
         tx2.rehash()
         self.update_witness_block_with_transactions(block_4, [tx2])
-        self.test_node.test_witness_block(block_4, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_4, accepted=True)
 
         # Reset the tip back down for the next test
         sync_blocks(self.nodes)
@@ -1611,7 +1617,7 @@ def test_witness_sigops(self):
         tx2.wit.vtxinwit[-1].scriptWitness.stack = [ witness_program_justright ]
         tx2.rehash()
         self.update_witness_block_with_transactions(block_5, [tx2])
-        self.test_node.test_witness_block(block_5, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block_5, accepted=True)
 
         # TODO: test p2sh sigop counting
 
@@ -1689,7 +1695,7 @@ def test_uncompressed_pubkey(self):
         # Confirm it in a block.
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Now try to spend it. Send it to a P2WSH output, which we'll
         # use in the next test.
@@ -1708,11 +1714,11 @@ def test_uncompressed_pubkey(self):
         tx2.rehash()
 
         # Should fail policy test.
-        self.test_node.test_transaction_acceptance(tx2, True, False, b'non-mandatory-script-verify-flag (Using non-compressed keys in segwit)')
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx2, True, False, b'non-mandatory-script-verify-flag (Using non-compressed keys in segwit)')
         # But passes consensus.
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx2])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Test 2: P2WSH
         # Try to spend the P2WSH output created in last test.
@@ -1728,11 +1734,11 @@ def test_uncompressed_pubkey(self):
         sign_P2PK_witness_input(witness_program, tx3, 0, SIGHASH_ALL, tx2.vout[0].nValue, key)
 
         # Should fail policy test.
-        self.test_node.test_transaction_acceptance(tx3, True, False, b'non-mandatory-script-verify-flag (Using non-compressed keys in segwit)')
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx3, True, False, b'non-mandatory-script-verify-flag (Using non-compressed keys in segwit)')
         # But passes consensus.
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx3])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Test 3: P2SH(P2WSH)
         # Try to spend the P2SH output created in the last test.
@@ -1745,10 +1751,10 @@ def test_uncompressed_pubkey(self):
         sign_P2PK_witness_input(witness_program, tx4, 0, SIGHASH_ALL, tx3.vout[0].nValue, key)
 
         # Should fail policy test.
-        self.test_node.test_transaction_acceptance(tx4, True, False, b'non-mandatory-script-verify-flag (Using non-compressed keys in segwit)')
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx4, True, False, b'non-mandatory-script-verify-flag (Using non-compressed keys in segwit)')
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx4])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
 
         # Test 4: Uncompressed pubkeys should still be valid in non-segwit
         # transactions.
@@ -1760,10 +1766,10 @@ def test_uncompressed_pubkey(self):
         tx5.vin[0].scriptSig = CScript([signature, pubkey])
         tx5.rehash()
         # Should pass policy and consensus.
-        self.test_node.test_transaction_acceptance(tx5, True, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx5, True, True)
         block = self.build_next_block()
         self.update_witness_block_with_transactions(block, [tx5])
-        self.test_node.test_witness_block(block, accepted=True)
+        test_witness_block(self.nodes[0].rpc, self.test_node, block, accepted=True)
         self.utxo.append(UTXO(tx5.sha256, 0, tx5.vout[0].nValue))
 
     def test_non_standard_witness(self):
@@ -1793,7 +1799,7 @@ def test_non_standard_witness(self):
             tx.vout.append(CTxOut(outputvalue, CScript([OP_HASH160, p2sh, OP_EQUAL])))
         tx.rehash()
         txid = tx.sha256
-        self.test_node.test_transaction_acceptance(tx, with_witness=False, accepted=True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, tx, with_witness=False, accepted=True)
 
         self.nodes[0].generate(1)
         sync_blocks(self.nodes)
@@ -1818,45 +1824,45 @@ def test_non_standard_witness(self):
         # Testing native P2WSH
         # Witness stack size, excluding witnessScript, over 100 is non-standard
         p2wsh_txs[0].wit.vtxinwit[0].scriptWitness.stack = [pad] * 101 + [scripts[0]]
-        self.std_node.test_transaction_acceptance(p2wsh_txs[0], True, False, b'bad-witness-nonstandard')
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2wsh_txs[0], True, False, b'bad-witness-nonstandard')
         # Non-standard nodes should accept
-        self.test_node.test_transaction_acceptance(p2wsh_txs[0], True, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2wsh_txs[0], True, True)
 
         # Stack element size over 80 bytes is non-standard
         p2wsh_txs[1].wit.vtxinwit[0].scriptWitness.stack = [pad * 81] * 100 + [scripts[1]]
-        self.std_node.test_transaction_acceptance(p2wsh_txs[1], True, False, b'bad-witness-nonstandard')
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2wsh_txs[1], True, False, b'bad-witness-nonstandard')
         # Non-standard nodes should accept
-        self.test_node.test_transaction_acceptance(p2wsh_txs[1], True, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2wsh_txs[1], True, True)
         # Standard nodes should accept if element size is not over 80 bytes
         p2wsh_txs[1].wit.vtxinwit[0].scriptWitness.stack = [pad * 80] * 100 + [scripts[1]]
-        self.std_node.test_transaction_acceptance(p2wsh_txs[1], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2wsh_txs[1], True, True)
 
         # witnessScript size at 3600 bytes is standard
         p2wsh_txs[2].wit.vtxinwit[0].scriptWitness.stack = [pad, pad, scripts[2]]
-        self.test_node.test_transaction_acceptance(p2wsh_txs[2], True, True)
-        self.std_node.test_transaction_acceptance(p2wsh_txs[2], True, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2wsh_txs[2], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2wsh_txs[2], True, True)
 
         # witnessScript size at 3601 bytes is non-standard
         p2wsh_txs[3].wit.vtxinwit[0].scriptWitness.stack = [pad, pad, pad, scripts[3]]
-        self.std_node.test_transaction_acceptance(p2wsh_txs[3], True, False, b'bad-witness-nonstandard')
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2wsh_txs[3], True, False, b'bad-witness-nonstandard')
         # Non-standard nodes should accept
-        self.test_node.test_transaction_acceptance(p2wsh_txs[3], True, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2wsh_txs[3], True, True)
 
         # Repeating the same tests with P2SH-P2WSH
         p2sh_txs[0].wit.vtxinwit[0].scriptWitness.stack = [pad] * 101 + [scripts[0]]
-        self.std_node.test_transaction_acceptance(p2sh_txs[0], True, False, b'bad-witness-nonstandard')
-        self.test_node.test_transaction_acceptance(p2sh_txs[0], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2sh_txs[0], True, False, b'bad-witness-nonstandard')
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2sh_txs[0], True, True)
         p2sh_txs[1].wit.vtxinwit[0].scriptWitness.stack = [pad * 81] * 100 + [scripts[1]]
-        self.std_node.test_transaction_acceptance(p2sh_txs[1], True, False, b'bad-witness-nonstandard')
-        self.test_node.test_transaction_acceptance(p2sh_txs[1], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2sh_txs[1], True, False, b'bad-witness-nonstandard')
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2sh_txs[1], True, True)
         p2sh_txs[1].wit.vtxinwit[0].scriptWitness.stack = [pad * 80] * 100 + [scripts[1]]
-        self.std_node.test_transaction_acceptance(p2sh_txs[1], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2sh_txs[1], True, True)
         p2sh_txs[2].wit.vtxinwit[0].scriptWitness.stack = [pad, pad, scripts[2]]
-        self.test_node.test_transaction_acceptance(p2sh_txs[2], True, True)
-        self.std_node.test_transaction_acceptance(p2sh_txs[2], True, True)
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2sh_txs[2], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2sh_txs[2], True, True)
         p2sh_txs[3].wit.vtxinwit[0].scriptWitness.stack = [pad, pad, pad, scripts[3]]
-        self.std_node.test_transaction_acceptance(p2sh_txs[3], True, False, b'bad-witness-nonstandard')
-        self.test_node.test_transaction_acceptance(p2sh_txs[3], True, True)
+        test_transaction_acceptance(self.nodes[1].rpc, self.std_node, p2sh_txs[3], True, False, b'bad-witness-nonstandard')
+        test_transaction_acceptance(self.nodes[0].rpc, self.test_node, p2sh_txs[3], True, True)
 
         self.nodes[0].generate(1)  # Mine and clean up the mempool of non-standard node
         # Valid but non-standard transactions in a block should be accepted by standard node
@@ -1870,11 +1876,11 @@ def test_non_standard_witness(self):
     def run_test(self):
         # Setup the p2p connections and start up the network thread.
         # self.test_node sets NODE_WITNESS|NODE_NETWORK
-        self.test_node = self.nodes[0].add_p2p_connection(TestNode(self.nodes[0].rpc), services=NODE_NETWORK|NODE_WITNESS)
+        self.test_node = self.nodes[0].add_p2p_connection(TestNode(), services=NODE_NETWORK|NODE_WITNESS)
         # self.old_node sets only NODE_NETWORK
-        self.old_node = self.nodes[0].add_p2p_connection(TestNode(self.nodes[0].rpc), services=NODE_NETWORK)
+        self.old_node = self.nodes[0].add_p2p_connection(TestNode(), services=NODE_NETWORK)
         # self.std_node is for testing node1 (fRequireStandard=true)
-        self.std_node = self.nodes[1].add_p2p_connection(TestNode(self.nodes[1].rpc), services=NODE_NETWORK|NODE_WITNESS)
+        self.std_node = self.nodes[1].add_p2p_connection(TestNode(), services=NODE_NETWORK|NODE_WITNESS)
 
         NetworkThread().start() # Start up network handling in another thread
 
diff --git a/test/functional/p2p-timeouts.py b/test/functional/p2p-timeouts.py
index 14a3bf48fbc99..21ae29eb6a0c9 100755
--- a/test/functional/p2p-timeouts.py
+++ b/test/functional/p2p-timeouts.py
@@ -28,7 +28,7 @@
 from test_framework.util import *
 
 class TestNode(NodeConnCB):
-    def on_version(self, conn, message):
+    def on_version(self, message):
         # Don't send a verack in response
         pass
 
diff --git a/test/functional/p2p-versionbits-warning.py b/test/functional/p2p-versionbits-warning.py
index 464ca5a312540..a6265f6d9cf4c 100755
--- a/test/functional/p2p-versionbits-warning.py
+++ b/test/functional/p2p-versionbits-warning.py
@@ -24,7 +24,7 @@
 VB_PATTERN = re.compile("^Warning.*versionbit")
 
 class TestNode(NodeConnCB):
-    def on_inv(self, conn, message):
+    def on_inv(self, message):
         pass
 
 class VersionBitsWarningTest(BitcoinTestFramework):
diff --git a/test/functional/sendheaders.py b/test/functional/sendheaders.py
index 68c0d95b4f991..b589d0e8a5625 100755
--- a/test/functional/sendheaders.py
+++ b/test/functional/sendheaders.py
@@ -113,6 +113,7 @@
 class BaseNode(NodeConnCB):
     def __init__(self):
         super().__init__()
+
         self.block_announced = False
         self.last_blockhash_announced = None
 
@@ -121,18 +122,18 @@ def send_get_data(self, block_hashes):
         msg = msg_getdata()
         for x in block_hashes:
             msg.inv.append(CInv(2, x))
-        self.connection.send_message(msg)
+        self.send_message(msg)
 
     def send_get_headers(self, locator, hashstop):
         msg = msg_getheaders()
         msg.locator.vHave = locator
         msg.hashstop = hashstop
-        self.connection.send_message(msg)
+        self.send_message(msg)
 
     def send_block_inv(self, blockhash):
         msg = msg_inv()
         msg.inv = [CInv(2, blockhash)]
-        self.connection.send_message(msg)
+        self.send_message(msg)
 
     def send_header_for_blocks(self, new_blocks):
         headers_message = msg_headers()
@@ -155,11 +156,11 @@ def wait_for_block_announcement(self, block_hash, timeout=60):
         test_function = lambda: self.last_blockhash_announced == block_hash
         wait_until(test_function, timeout=timeout, lock=mininode_lock)
 
-    def on_inv(self, conn, message):
+    def on_inv(self, message):
         self.block_announced = True
         self.last_blockhash_announced = message.inv[-1].hash
 
-    def on_headers(self, conn, message):
+    def on_headers(self, message):
         if len(message.headers):
             self.block_announced = True
             message.headers[-1].calc_sha256()
diff --git a/test/functional/test_framework/comptool.py b/test/functional/test_framework/comptool.py
index 723826bae4c70..2f64fba7536d8 100755
--- a/test/functional/test_framework/comptool.py
+++ b/test/functional/test_framework/comptool.py
@@ -43,7 +43,6 @@ 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 = {}
@@ -58,26 +57,23 @@ def __init__(self, block_store, tx_store):
         self.lastInv = []
         self.closed = False
 
-    def on_close(self, conn):
+    def on_close(self):
         self.closed = True
 
-    def add_connection(self, conn):
-        self.conn = conn
-
-    def on_headers(self, conn, message):
+    def on_headers(self, 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):
+    def on_getheaders(self, message):
         response = self.block_store.headers_for(message.locator, message.hashstop)
         if response is not None:
-            conn.send_message(response)
+            self.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)]
+    def on_getdata(self, message):
+        [self.send_message(r) for r in self.block_store.get_blocks(message.inv)]
+        [self.send_message(r) for r in self.tx_store.get_transactions(message.inv)]
 
         for i in message.inv:
             if i.type == 1 or i.type == 1 | (1 << 30): # MSG_TX or MSG_WITNESS_TX
@@ -85,16 +81,16 @@ def on_getdata(self, conn, message):
             elif i.type == 2 or i.type == 2 | (1 << 30): # MSG_BLOCK or MSG_WITNESS_BLOCK
                 self.block_request_map[i.hash] = True
 
-    def on_inv(self, conn, message):
+    def on_inv(self, message):
         self.lastInv = [x.hash for x in message.inv]
 
-    def on_pong(self, conn, message):
+    def on_pong(self, 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):
+    def on_reject(self, message):
         if message.message == b'tx':
             self.tx_reject_map[message.data] = RejectResult(message.code, message.reason)
         if message.message == b'block':
@@ -102,30 +98,30 @@ def on_reject(self, conn, message):
 
     def send_inv(self, obj):
         mtype = 2 if isinstance(obj, CBlock) else 1
-        self.conn.send_message(msg_inv([CInv(mtype, obj.sha256)]))
+        self.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)
+        self.send_message(m)
 
     def send_header(self, header):
         m = msg_headers()
         m.headers.append(header)
-        self.conn.send_message(m)
+        self.send_message(m)
 
     # This assumes BIP31
     def send_ping(self, nonce):
         self.pingMap[nonce] = True
-        self.conn.send_message(msg_ping(nonce))
+        self.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())
+        self.send_message(msg_mempool())
 
 # TestInstance:
 #
@@ -166,8 +162,7 @@ class TestManager():
 
     def __init__(self, testgen, datadir):
         self.test_generator = testgen
-        self.connections    = []
-        self.test_nodes     = []
+        self.p2p_connections= []
         self.block_store    = BlockStore(datadir)
         self.tx_store       = TxStore(datadir)
         self.ping_counter   = 1
@@ -175,28 +170,24 @@ def __init__(self, testgen, datadir):
     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), test_node))
-            # Make sure the TestNode (callback class) has a reference to its
-            # associated NodeConn
-            test_node.add_connection(self.connections[-1])
+            node = TestNode(self.block_store, self.tx_store)
+            node.peer_connect('127.0.0.1', p2p_port(i))
+            self.p2p_connections.append(node)
 
     def clear_all_connections(self):
-        self.connections    = []
-        self.test_nodes     = []
+        self.p2p_connections = []
 
     def wait_for_disconnections(self):
         def disconnected():
-            return all(node.closed for node in self.test_nodes)
+            return all(node.closed for node in self.p2p_connections)
         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)
+        return all(node.wait_for_verack() for node in self.p2p_connections)
 
     def wait_for_pings(self, counter):
         def received_pongs():
-            return all(node.received_ping_response(counter) for node in self.test_nodes)
+            return all(node.received_ping_response(counter) for node in self.p2p_connections)
         wait_until(received_pongs, lock=mininode_lock)
 
     # sync_blocks: Wait for all connections to request the blockhash given
@@ -206,17 +197,17 @@ 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
+                for node in self.p2p_connections
             )
 
         # --> 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 ]
+        [ c.send_getheaders() for c in self.p2p_connections ]
 
         # Send ping and wait for response -- synchronization hack
-        [ c.cb.send_ping(self.ping_counter) for c in self.connections ]
+        [ c.send_ping(self.ping_counter) for c in self.p2p_connections ]
         self.wait_for_pings(self.ping_counter)
         self.ping_counter += 1
 
@@ -226,42 +217,42 @@ def sync_transaction(self, txhash, 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
+                for node in self.p2p_connections
             )
 
         # --> 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 ]
+        [ c.send_mempool() for c in self.p2p_connections ]
 
         # Send ping and wait for response -- synchronization hack
-        [ c.cb.send_ping(self.ping_counter) for c in self.connections ]
+        [ c.send_ping(self.ping_counter) for c in self.p2p_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 ]
+            [ c.lastInv.sort() for c in self.p2p_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:
+            for c in self.p2p_connections:
                 if outcome is None:
-                    if c.cb.bestblockhash != self.connections[0].cb.bestblockhash:
+                    if c.bestblockhash != self.p2p_connections[0].bestblockhash:
                         return False
                 elif isinstance(outcome, RejectResult): # Check that block was rejected w/ code
-                    if c.cb.bestblockhash == blockhash:
+                    if c.bestblockhash == blockhash:
                         return False
-                    if blockhash not in c.cb.block_reject_map:
+                    if blockhash not in c.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))
+                    if not outcome.match(c.block_reject_map[blockhash]):
+                        logger.error('Block rejected with %s instead of expected %s: %064x' % (c.block_reject_map[blockhash], outcome, blockhash))
                         return False
-                elif ((c.cb.bestblockhash == blockhash) != outcome):
+                elif ((c.bestblockhash == blockhash) != outcome):
                     return False
             return True
 
@@ -273,21 +264,21 @@ def check_results(self, blockhash, outcome):
     # 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:
+            for c in self.p2p_connections:
                 if outcome is None:
                     # Make sure the mempools agree with each other
-                    if c.cb.lastInv != self.connections[0].cb.lastInv:
+                    if c.lastInv != self.p2p_connections[0].lastInv:
                         return False
                 elif isinstance(outcome, RejectResult): # Check that tx was rejected w/ code
-                    if txhash in c.cb.lastInv:
+                    if txhash in c.lastInv:
                         return False
-                    if txhash not in c.cb.tx_reject_map:
+                    if txhash not in c.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))
+                    if not outcome.match(c.tx_reject_map[txhash]):
+                        logger.error('Tx rejected with %s instead of expected %s: %064x' % (c.tx_reject_map[txhash], outcome, txhash))
                         return False
-                elif ((txhash in c.cb.lastInv) != outcome):
+                elif ((txhash in c.lastInv) != outcome):
                     return False
             return True
 
@@ -332,25 +323,25 @@ def run(self):
                         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:
+                        for c in self.p2p_connections:
+                            if first_block_with_hash and block.sha256 in c.block_request_map and c.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
+                                c.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 ]
+                            [ c.send_inv(block) for c in self.p2p_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 ]
+                            [ c.send_message(msg_block(block)) for c in self.p2p_connections ]
+                            [ c.send_ping(self.ping_counter) for c in self.p2p_connections ]
                             self.wait_for_pings(self.ping_counter)
                             self.ping_counter += 1
                         if (not self.check_results(tip, outcome)):
@@ -360,7 +351,7 @@ def run(self):
                 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 ]
+                    [ c.send_header(block_header) for c in self.p2p_connections ]
 
                 else:  # Tx test runner
                     assert(isinstance(b_or_t, CTransaction))
@@ -369,11 +360,11 @@ def run(self):
                     # 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
+                        for c in self.p2p_connections:
+                            c.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 ]
+                        [ c.send_inv(tx) for c in self.p2p_connections ]
                         self.sync_transaction(tx.sha256, 1)
                         if (not self.check_mempool(tx.sha256, outcome)):
                             raise AssertionError("Test failed at test %d" % test_number)
@@ -381,26 +372,26 @@ def run(self):
                         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 ]
+                    [ c.send_message(msg_inv(invqueue)) for c in self.p2p_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 ]
+                    [ c.send_message(msg_inv(invqueue)) for c in self.p2p_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 ]
+                    [ c.send_message(msg_inv(invqueue)) for c in self.p2p_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)
 
-        [ c.disconnect_node() for c in self.connections ]
+        [ c.disconnect_node() for c in self.p2p_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 24c96b5681b5c..c580d99c79e20 100755
--- a/test/functional/test_framework/mininode.py
+++ b/test/functional/test_framework/mininode.py
@@ -20,10 +20,10 @@
 import socket
 import struct
 import sys
-import time
 from threading import RLock, Thread
 
 from test_framework.messages import *
+from test_framework.util import wait_until
 
 logger = logging.getLogger("TestFramework.mininode")
 
@@ -57,173 +57,34 @@
     "regtest": b"\xfa\xbf\xb5\xda",   # regtest
 }
 
-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):
-        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):
-        test_function = lambda: not self.connected
-        wait_until(test_function, timeout=timeout, lock=mininode_lock)
-
-    # Message receiving helper methods
-
-    def wait_for_block(self, blockhash, timeout=60):
-        test_function = lambda: 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):
-        test_function = lambda: self.last_message.get("getdata")
-        wait_until(test_function, timeout=timeout, lock=mininode_lock)
-
-    def wait_for_getheaders(self, timeout=60):
-        test_function = lambda: 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")
-        test_function = lambda: 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):
-        test_function = lambda: self.message_count["verack"]
-        wait_until(test_function, timeout=timeout, lock=mininode_lock)
-
-    # Message sending helper functions
+class NodeConn(asyncore.dispatcher):
+    """A low-level connection object to a node's P2P interface.
 
-    def send_message(self, message):
-        if self.connection:
-            self.connection.send_message(message)
-        else:
-            logger.error("Cannot send message. No connection to node!")
+    This class is responsible for:
 
-    def send_and_ping(self, message):
-        self.send_message(message)
-        self.sync_with_ping()
+    - opening and closing the TCP connection to the node
+    - reading bytes from and writing bytes to the socket
+    - deserializing and serializing the P2P message header
+    - logging messages as they are sent and received
 
-    # Sync up with the node
-    def sync_with_ping(self, timeout=60):
-        self.send_message(msg_ping(nonce=self.ping_counter))
-        test_function = lambda: self.last_message.get("pong") and self.last_message["pong"].nonce == self.ping_counter
-        wait_until(test_function, timeout=timeout, lock=mininode_lock)
-        self.ping_counter += 1
+    This class contains no logic for handing the P2P message payloads. It must be
+    sub-classed and the on_message() callback overridden.
 
-class NodeConn(asyncore.dispatcher):
-    """The actual NodeConn class
+    TODO: rename this class P2PConnection."""
 
-    This class provides an interface for a p2p connection to a specified node."""
+    def __init__(self):
+        super().__init__(map=mininode_socket_map)
 
-    def __init__(self, dstaddr, dstport, callback, net="regtest", services=NODE_NETWORK|NODE_WITNESS, send_version=True):
-        asyncore.dispatcher.__init__(self, map=mininode_socket_map)
+    def peer_connect(self, dstaddr, dstport, net="regtest"):
         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.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))
 
@@ -232,15 +93,22 @@ def __init__(self, dstaddr, dstport, callback, net="regtest", services=NODE_NETW
         except:
             self.handle_close()
 
+    def peer_disconnect(self):
+        # Connection could have already been closed by other end.
+        if self.state == "connected":
+            self.disconnect_node()
+
     # Connection and disconnection methods
 
     def handle_connect(self):
+        """asyncore callback when a connection is opened."""
         if self.state != "connected":
             logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport))
             self.state = "connected"
-            self.cb.on_open(self)
+            self.on_open()
 
     def handle_close(self):
+        """asyncore callback when a connection is closed."""
         logger.debug("Closing connection to: %s:%d" % (self.dstaddr, self.dstport))
         self.state = "closed"
         self.recvbuf = b""
@@ -249,10 +117,10 @@ def handle_close(self):
             self.close()
         except:
             pass
-        self.cb.on_close(self)
+        self.on_close()
 
     def disconnect_node(self):
-        """ Disconnect the p2p connection.
+        """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."""
@@ -260,16 +128,19 @@ def disconnect_node(self):
 
     # Socket read methods
 
-    def readable(self):
-        return True
-
     def handle_read(self):
+        """asyncore callback when data is read from the socket."""
         t = self.recv(8192)
         if len(t) > 0:
             self.recvbuf += t
-            self.got_data()
+            self._on_data()
+
+    def _on_data(self):
+        """Try to read P2P messages from the recv buffer.
 
-    def got_data(self):
+        This method reads data from the buffer in a loop. It deserializes,
+        parses and verifies the P2P header, then passes the P2P payload to
+        the on_message callback for processing."""
         try:
             while True:
                 if len(self.recvbuf) < 4:
@@ -294,26 +165,27 @@ def got_data(self):
                 f = BytesIO(msg)
                 t = MESSAGEMAP[command]()
                 t.deserialize(f)
-                self.got_message(t)
+                self._log_message("receive", t)
+                self.on_message(t)
         except Exception as e:
             logger.exception('Error reading message:', repr(e))
             raise
 
-    def got_message(self, message):
-        if self.last_sent + 30 * 60 < time.time():
-            self.send_message(MESSAGEMAP[b'ping']())
-        self._log_message("receive", message)
-        self.cb.deliver(self, message)
+    def on_message(self, message):
+        """Callback for processing a P2P payload. Must be overridden by derived class."""
+        raise NotImplementedError
 
     # Socket write methods
 
     def writable(self):
+        """asyncore method to determine whether the handle_write() callback should be called on the next loop."""
         with mininode_lock:
             pre_connection = self.state == "connecting"
             length = len(self.sendbuf)
         return (length > 0 or pre_connection)
 
     def handle_write(self):
+        """asyncore callback when data should be written to the socket."""
         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
@@ -331,6 +203,10 @@ def handle_write(self):
             self.sendbuf = self.sendbuf[sent:]
 
     def send_message(self, message, pushbuf=False):
+        """Send a P2P message over the socket.
+
+        This method takes a P2P payload, builds the P2P header and adds
+        the message to the send buffer to be sent over the socket."""
         if self.state != "connected" and not pushbuf:
             raise IOError('Not connected, no pushbuf')
         self._log_message("send", message)
@@ -353,11 +229,11 @@ def send_message(self, message, pushbuf=False):
                     self.sendbuf = tmsg
             else:
                 self.sendbuf += tmsg
-            self.last_sent = time.time()
 
     # Class utility methods
 
     def _log_message(self, direction, msg):
+        """Logs a message being sent or received over the connection."""
         if direction == "send":
             log_message = "Send message to "
         elif direction == "receive":
@@ -368,6 +244,154 @@ def _log_message(self, direction, msg):
         logger.debug(log_message)
 
 
+class NodeConnCB(NodeConn):
+    """A high-level P2P interface class for communicating with a Bitcoin node.
+
+    This class provides high-level callbacks for processing P2P message
+    payloads, as well as convenience methods for interacting with the
+    node over P2P.
+
+    Individual testcases should subclass this and override the on_* methods
+    if they want to alter message handling behaviour.
+
+    TODO: rename this class P2PInterface"""
+    def __init__(self):
+        super().__init__()
+
+        # 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
+
+        # The network services received from the peer
+        self.nServices = 0
+
+    def peer_connect(self, *args, services=NODE_NETWORK|NODE_WITNESS, send_version=True, **kwargs):
+        super().peer_connect(*args, **kwargs)
+
+        if send_version:
+            # Send a version msg
+            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)
+
+    # Message receiving methods
+
+    def on_message(self, 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)(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):
+        pass
+
+    def on_close(self):
+        pass
+
+    def on_addr(self, message): pass
+    def on_block(self, message): pass
+    def on_blocktxn(self, message): pass
+    def on_cmpctblock(self, message): pass
+    def on_feefilter(self, message): pass
+    def on_getaddr(self, message): pass
+    def on_getblocks(self, message): pass
+    def on_getblocktxn(self, message): pass
+    def on_getdata(self, message): pass
+    def on_getheaders(self, message): pass
+    def on_headers(self, message): pass
+    def on_mempool(self, message): pass
+    def on_pong(self, message): pass
+    def on_reject(self, message): pass
+    def on_sendcmpct(self, message): pass
+    def on_sendheaders(self, message): pass
+    def on_tx(self, message): pass
+
+    def on_inv(self, message):
+        want = msg_getdata()
+        for i in message.inv:
+            if i.type != 0:
+                want.inv.append(i)
+        if len(want.inv):
+            self.send_message(want)
+
+    def on_ping(self, message):
+        self.send_message(msg_pong(message.nonce))
+
+    def on_verack(self, message):
+        self.verack_received = True
+
+    def on_version(self, message):
+        assert message.nVersion >= MIN_VERSION_SUPPORTED, "Version {} received. Test framework only supports versions greater than {}".format(message.nVersion, MIN_VERSION_SUPPORTED)
+        self.send_message(msg_verack())
+        self.nServices = message.nServices
+
+    # Connection helper methods
+
+    def wait_for_disconnect(self, timeout=60):
+        test_function = lambda: self.state != "connected"
+        wait_until(test_function, timeout=timeout, lock=mininode_lock)
+
+    # Message receiving helper methods
+
+    def wait_for_block(self, blockhash, timeout=60):
+        test_function = lambda: 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):
+        test_function = lambda: self.last_message.get("getdata")
+        wait_until(test_function, timeout=timeout, lock=mininode_lock)
+
+    def wait_for_getheaders(self, timeout=60):
+        test_function = lambda: 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")
+        test_function = lambda: 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):
+        test_function = lambda: self.message_count["verack"]
+        wait_until(test_function, timeout=timeout, lock=mininode_lock)
+
+    # Message sending helper functions
+
+    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))
+        test_function = lambda: self.last_message.get("pong") and self.last_message["pong"].nonce == self.ping_counter
+        wait_until(test_function, timeout=timeout, lock=mininode_lock)
+        self.ping_counter += 1
+
+
 # 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)
diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py
index 34b458482a86a..a9248c764e3d4 100755
--- a/test/functional/test_framework/test_node.py
+++ b/test/functional/test_framework/test_node.py
@@ -14,7 +14,6 @@
 import time
 
 from .authproxy import JSONRPCException
-from .mininode import NodeConn
 from .util import (
     assert_equal,
     get_rpc_proxy,
@@ -158,7 +157,7 @@ def node_encrypt_wallet(self, passphrase):
         self.encryptwallet(passphrase)
         self.wait_until_stopped()
 
-    def add_p2p_connection(self, p2p_conn, **kwargs):
+    def add_p2p_connection(self, p2p_conn, *args, **kwargs):
         """Add a p2p connection to the node.
 
         This method adds the p2p connection to the self.p2ps list and also
@@ -167,9 +166,9 @@ def add_p2p_connection(self, p2p_conn, **kwargs):
             kwargs['dstport'] = p2p_port(self.index)
         if 'dstaddr' not in kwargs:
             kwargs['dstaddr'] = '127.0.0.1'
+
+        p2p_conn.peer_connect(*args, **kwargs)
         self.p2ps.append(p2p_conn)
-        kwargs.update({'callback': p2p_conn})
-        p2p_conn.add_connection(NodeConn(**kwargs))
 
         return p2p_conn
 
@@ -185,10 +184,8 @@ def p2p(self):
     def disconnect_p2ps(self):
         """Close all p2p connections to the node."""
         for p in self.p2ps:
-            # Connection could have already been closed by other end.
-            if p.connection is not None:
-                p.connection.disconnect_node()
-        self.p2ps = []
+            p.peer_disconnect()
+        del self.p2ps[:]
 
 
 class TestNodeCLI():