diff --git a/test/functional/assumevalid.py b/test/functional/assumevalid.py
index 65685c48b..36761d359 100755
--- a/test/functional/assumevalid.py
+++ b/test/functional/assumevalid.py
@@ -39,13 +39,12 @@ from test_framework.mininode import (CBlockHeader,
                                      CTxIn,
                                      CTxOut,
                                      NetworkThread,
-                                     NodeConn,
                                      NodeConnCB,
                                      msg_block,
                                      msg_headers)
 from test_framework.script import (CScript, OP_TRUE)
 from test_framework.test_framework import BitcoinTestFramework
-from test_framework.util import (p2p_port, assert_equal)
+from test_framework.util import assert_equal
 
 class BaseNode(NodeConnCB):
     def send_header_for_blocks(self, new_blocks):
@@ -65,13 +64,13 @@ class AssumeValidTest(BitcoinTestFramework):
         # signature so we can pass in the block hash as assumevalid.
         self.start_node(0)
 
-    def send_blocks_until_disconnected(self, node):
+    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 node.connection:
+            if not p2p_conn.connection:
                 break
             try:
-                node.send_message(msg_block(self.blocks[i]))
+                p2p_conn.send_message(msg_block(self.blocks[i]))
             except IOError as e:
                 assert str(e) == 'Not connected, no pushbuf'
                 break
@@ -97,13 +96,10 @@ class AssumeValidTest(BitcoinTestFramework):
     def run_test(self):
 
         # Connect to node0
-        node0 = BaseNode()
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
-        node0.add_connection(connections[0])
+        p2p0 = self.nodes[0].add_p2p_connection(BaseNode())
 
         NetworkThread().start()  # Start up network handling in another thread
-        node0.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         # Build the blockchain
         self.tip = int(self.nodes[0].getbestblockhash(), 16)
@@ -165,37 +161,33 @@ class AssumeValidTest(BitcoinTestFramework):
 
         # Start node1 and node2 with assumevalid so they accept a block with a bad signature.
         self.start_node(1, extra_args=["-assumevalid=" + hex(block102.sha256)])
-        node1 = BaseNode()  # connects to node1
-        connections.append(NodeConn('127.0.0.1', p2p_port(1), self.nodes[1], node1))
-        node1.add_connection(connections[1])
-        node1.wait_for_verack()
+        p2p1 = self.nodes[1].add_p2p_connection(BaseNode())
+        p2p1.wait_for_verack()
 
         self.start_node(2, extra_args=["-assumevalid=" + hex(block102.sha256)])
-        node2 = BaseNode()  # connects to node2
-        connections.append(NodeConn('127.0.0.1', p2p_port(2), self.nodes[2], node2))
-        node2.add_connection(connections[2])
-        node2.wait_for_verack()
+        p2p2 = self.nodes[2].add_p2p_connection(BaseNode())
+        p2p2.wait_for_verack()
 
         # send header lists to all three nodes
-        node0.send_header_for_blocks(self.blocks[0:2000])
-        node0.send_header_for_blocks(self.blocks[2000:])
-        node1.send_header_for_blocks(self.blocks[0:2000])
-        node1.send_header_for_blocks(self.blocks[2000:])
-        node2.send_header_for_blocks(self.blocks[0:200])
+        p2p0.send_header_for_blocks(self.blocks[0:2000])
+        p2p0.send_header_for_blocks(self.blocks[2000:])
+        p2p1.send_header_for_blocks(self.blocks[0:2000])
+        p2p1.send_header_for_blocks(self.blocks[2000:])
+        p2p2.send_header_for_blocks(self.blocks[0:200])
 
         # Send blocks to node0. Block 102 will be rejected.
-        self.send_blocks_until_disconnected(node0)
+        self.send_blocks_until_disconnected(p2p0)
         self.assert_blockchain_height(self.nodes[0], 101)
 
         # Send all blocks to node1. All blocks will be accepted.
         for i in range(2202):
-            node1.send_message(msg_block(self.blocks[i]))
+            p2p1.send_message(msg_block(self.blocks[i]))
         # Syncing 2200 blocks can take a while on slow systems. Give it plenty of time to sync.
-        node1.sync_with_ping(120)
+        p2p1.sync_with_ping(120)
         assert_equal(self.nodes[1].getblock(self.nodes[1].getbestblockhash())['height'], 2202)
 
         # Send blocks to node2. Block 102 will be rejected.
-        self.send_blocks_until_disconnected(node2)
+        self.send_blocks_until_disconnected(p2p2)
         self.assert_blockchain_height(self.nodes[2], 101)
 
 if __name__ == '__main__':
diff --git a/test/functional/bip65-cltv-p2p.py b/test/functional/bip65-cltv-p2p.py
index 2cd6df6e3..307332479 100755
--- a/test/functional/bip65-cltv-p2p.py
+++ b/test/functional/bip65-cltv-p2p.py
@@ -66,15 +66,12 @@ class BIP65Test(BitcoinTestFramework):
         self.setup_clean_chain = True
 
     def run_test(self):
-        node0 = NodeConnCB()
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
-        node0.add_connection(connections[0])
+        self.nodes[0].add_p2p_connection(NodeConnCB())
 
         NetworkThread().start() # Start up network handling in another thread
 
         # wait_for_verack ensures that the P2P connection is fully up.
-        node0.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         self.log.info("Mining %d blocks", CLTV_HEIGHT - 2)
         self.coinbase_blocks = self.nodes[0].generate(CLTV_HEIGHT - 2)
@@ -95,7 +92,7 @@ class BIP65Test(BitcoinTestFramework):
         block.hashMerkleRoot = block.calc_merkle_root()
         block.solve()
 
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(self.nodes[0].getbestblockhash(), block.hash)
 
         self.log.info("Test that blocks must now be at least version 4")
@@ -104,15 +101,15 @@ class BIP65Test(BitcoinTestFramework):
         block = create_block(tip, create_coinbase(CLTV_HEIGHT), block_time)
         block.nVersion = 3
         block.solve()
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(int(self.nodes[0].getbestblockhash(), 16), tip)
 
-        wait_until(lambda: "reject" in node0.last_message.keys(), lock=mininode_lock)
+        wait_until(lambda: "reject" in self.nodes[0].p2p.last_message.keys(), lock=mininode_lock)
         with mininode_lock:
-            assert_equal(node0.last_message["reject"].code, REJECT_OBSOLETE)
-            assert_equal(node0.last_message["reject"].reason, b'bad-version(0x00000003)')
-            assert_equal(node0.last_message["reject"].data, block.sha256)
-            del node0.last_message["reject"]
+            assert_equal(self.nodes[0].p2p.last_message["reject"].code, REJECT_OBSOLETE)
+            assert_equal(self.nodes[0].p2p.last_message["reject"].reason, b'bad-version(0x00000003)')
+            assert_equal(self.nodes[0].p2p.last_message["reject"].data, block.sha256)
+            del self.nodes[0].p2p.last_message["reject"]
 
         self.log.info("Test that invalid-according-to-cltv transactions cannot appear in a block")
         block.nVersion = 4
@@ -125,7 +122,7 @@ class BIP65Test(BitcoinTestFramework):
         # First we show that this tx is valid except for CLTV by getting it
         # accepted to the mempool (which we can achieve with
         # -promiscuousmempoolflags).
-        node0.send_and_ping(msg_tx(spendtx))
+        self.nodes[0].p2p.send_and_ping(msg_tx(spendtx))
         assert spendtx.hash in self.nodes[0].getrawmempool()
 
         # Now we verify that a block with this transaction is invalid.
@@ -133,18 +130,18 @@ class BIP65Test(BitcoinTestFramework):
         block.hashMerkleRoot = block.calc_merkle_root()
         block.solve()
 
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(int(self.nodes[0].getbestblockhash(), 16), tip)
 
-        wait_until(lambda: "reject" in node0.last_message.keys(), lock=mininode_lock)
+        wait_until(lambda: "reject" in self.nodes[0].p2p.last_message.keys(), lock=mininode_lock)
         with mininode_lock:
-            assert node0.last_message["reject"].code in [REJECT_INVALID, REJECT_NONSTANDARD]
-            assert_equal(node0.last_message["reject"].data, block.sha256)
-            if node0.last_message["reject"].code == REJECT_INVALID:
+            assert self.nodes[0].p2p.last_message["reject"].code in [REJECT_INVALID, REJECT_NONSTANDARD]
+            assert_equal(self.nodes[0].p2p.last_message["reject"].data, block.sha256)
+            if self.nodes[0].p2p.last_message["reject"].code == REJECT_INVALID:
                 # Generic rejection when a block is invalid
-                assert_equal(node0.last_message["reject"].reason, b'block-validation-failed')
+                assert_equal(self.nodes[0].p2p.last_message["reject"].reason, b'block-validation-failed')
             else:
-                assert b'Negative locktime' in node0.last_message["reject"].reason
+                assert b'Negative locktime' in self.nodes[0].p2p.last_message["reject"].reason
 
         self.log.info("Test that a version 4 block with a valid-according-to-CLTV transaction is accepted")
         spendtx = cltv_validate(self.nodes[0], spendtx, CLTV_HEIGHT - 1)
@@ -155,7 +152,7 @@ class BIP65Test(BitcoinTestFramework):
         block.hashMerkleRoot = block.calc_merkle_root()
         block.solve()
 
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(int(self.nodes[0].getbestblockhash(), 16), block.sha256)
 
 
diff --git a/test/functional/bipdersig-p2p.py b/test/functional/bipdersig-p2p.py
index c620d3e15..e5febde42 100755
--- a/test/functional/bipdersig-p2p.py
+++ b/test/functional/bipdersig-p2p.py
@@ -54,14 +54,12 @@ class BIP66Test(BitcoinTestFramework):
         self.setup_clean_chain = True
 
     def run_test(self):
-        node0 = NodeConnCB()
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
-        node0.add_connection(connections[0])
+        self.nodes[0].add_p2p_connection(NodeConnCB())
+
         NetworkThread().start() # Start up network handling in another thread
 
         # wait_for_verack ensures that the P2P connection is fully up.
-        node0.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         self.log.info("Mining %d blocks", DERSIG_HEIGHT - 2)
         self.coinbase_blocks = self.nodes[0].generate(DERSIG_HEIGHT - 2)
@@ -83,7 +81,7 @@ class BIP66Test(BitcoinTestFramework):
         block.rehash()
         block.solve()
 
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(self.nodes[0].getbestblockhash(), block.hash)
 
         self.log.info("Test that blocks must now be at least version 3")
@@ -93,15 +91,15 @@ class BIP66Test(BitcoinTestFramework):
         block.nVersion = 2
         block.rehash()
         block.solve()
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(int(self.nodes[0].getbestblockhash(), 16), tip)
 
-        wait_until(lambda: "reject" in node0.last_message.keys(), lock=mininode_lock)
+        wait_until(lambda: "reject" in self.nodes[0].p2p.last_message.keys(), lock=mininode_lock)
         with mininode_lock:
-            assert_equal(node0.last_message["reject"].code, REJECT_OBSOLETE)
-            assert_equal(node0.last_message["reject"].reason, b'bad-version(0x00000002)')
-            assert_equal(node0.last_message["reject"].data, block.sha256)
-            del node0.last_message["reject"]
+            assert_equal(self.nodes[0].p2p.last_message["reject"].code, REJECT_OBSOLETE)
+            assert_equal(self.nodes[0].p2p.last_message["reject"].reason, b'bad-version(0x00000002)')
+            assert_equal(self.nodes[0].p2p.last_message["reject"].data, block.sha256)
+            del self.nodes[0].p2p.last_message["reject"]
 
         self.log.info("Test that transactions with non-DER signatures cannot appear in a block")
         block.nVersion = 3
@@ -114,7 +112,7 @@ class BIP66Test(BitcoinTestFramework):
         # First we show that this tx is valid except for DERSIG by getting it
         # accepted to the mempool (which we can achieve with
         # -promiscuousmempoolflags).
-        node0.send_and_ping(msg_tx(spendtx))
+        self.nodes[0].p2p.send_and_ping(msg_tx(spendtx))
         assert spendtx.hash in self.nodes[0].getrawmempool()
 
         # Now we verify that a block with this transaction is invalid.
@@ -123,23 +121,23 @@ class BIP66Test(BitcoinTestFramework):
         block.rehash()
         block.solve()
 
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(int(self.nodes[0].getbestblockhash(), 16), tip)
 
-        wait_until(lambda: "reject" in node0.last_message.keys(), lock=mininode_lock)
+        wait_until(lambda: "reject" in self.nodes[0].p2p.last_message.keys(), lock=mininode_lock)
         with mininode_lock:
             # We can receive different reject messages depending on whether
             # bitcoind is running with multiple script check threads. If script
             # check threads are not in use, then transaction script validation
             # happens sequentially, and bitcoind produces more specific reject
             # reasons.
-            assert node0.last_message["reject"].code in [REJECT_INVALID, REJECT_NONSTANDARD]
-            assert_equal(node0.last_message["reject"].data, block.sha256)
-            if node0.last_message["reject"].code == REJECT_INVALID:
+            assert self.nodes[0].p2p.last_message["reject"].code in [REJECT_INVALID, REJECT_NONSTANDARD]
+            assert_equal(self.nodes[0].p2p.last_message["reject"].data, block.sha256)
+            if self.nodes[0].p2p.last_message["reject"].code == REJECT_INVALID:
                 # Generic rejection when a block is invalid
-                assert_equal(node0.last_message["reject"].reason, b'block-validation-failed')
+                assert_equal(self.nodes[0].p2p.last_message["reject"].reason, b'block-validation-failed')
             else:
-                assert b'Non-canonical DER signature' in node0.last_message["reject"].reason
+                assert b'Non-canonical DER signature' in self.nodes[0].p2p.last_message["reject"].reason
 
         self.log.info("Test that a version 3 block with a DERSIG-compliant transaction is accepted")
         block.vtx[1] = create_transaction(self.nodes[0],
@@ -148,7 +146,7 @@ class BIP66Test(BitcoinTestFramework):
         block.rehash()
         block.solve()
 
-        node0.send_and_ping(msg_block(block))
+        self.nodes[0].p2p.send_and_ping(msg_block(block))
         assert_equal(int(self.nodes[0].getbestblockhash(), 16), block.sha256)
 
 if __name__ == '__main__':
diff --git a/test/functional/example_test.py b/test/functional/example_test.py
index 87d73ad14..ba40f3301 100755
--- a/test/functional/example_test.py
+++ b/test/functional/example_test.py
@@ -18,7 +18,6 @@ from test_framework.blocktools import (create_block, create_coinbase)
 from test_framework.mininode import (
     CInv,
     NetworkThread,
-    NodeConn,
     NodeConnCB,
     mininode_lock,
     msg_block,
@@ -28,7 +27,6 @@ from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_equal,
     connect_nodes,
-    p2p_port,
     wait_until,
 )
 
@@ -134,16 +132,13 @@ class ExampleTest(BitcoinTestFramework):
         """Main test logic"""
 
         # Create a P2P connection to one of the nodes
-        node0 = BaseNode()
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
-        node0.add_connection(connections[0])
+        self.nodes[0].add_p2p_connection(BaseNode())
 
         # Start up network handling in another thread. This needs to be called
         # after the P2P connections have been created.
         NetworkThread().start()
         # wait_for_verack ensures that the P2P connection is fully up.
-        node0.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         # Generating a block on one of the nodes will get us out of IBD
         blocks = [int(self.nodes[0].generate(nblocks=1)[0], 16)]
@@ -180,7 +175,7 @@ class ExampleTest(BitcoinTestFramework):
             block.solve()
             block_message = msg_block(block)
             # Send message is used to send a P2P message to the node over our NodeConn connection
-            node0.send_message(block_message)
+            self.nodes[0].p2p.send_message(block_message)
             self.tip = block.sha256
             blocks.append(self.tip)
             self.block_time += 1
@@ -193,28 +188,26 @@ class ExampleTest(BitcoinTestFramework):
         connect_nodes(self.nodes[1], 2)
 
         self.log.info("Add P2P connection to node2")
-        node2 = BaseNode()
-        connections.append(NodeConn('127.0.0.1', p2p_port(2), self.nodes[2], node2))
-        node2.add_connection(connections[1])
-        node2.wait_for_verack()
+        self.nodes[2].add_p2p_connection(BaseNode())
+        self.nodes[2].p2p.wait_for_verack()
 
         self.log.info("Wait for node2 reach current tip. Test that it has propagated all the blocks to us")
 
         getdata_request = msg_getdata()
         for block in blocks:
             getdata_request.inv.append(CInv(2, block))
-        node2.send_message(getdata_request)
+        self.nodes[2].p2p.send_message(getdata_request)
 
         # wait_until() will loop until a predicate condition is met. Use it to test properties of the
         # NodeConnCB objects.
-        wait_until(lambda: sorted(blocks) == sorted(list(node2.block_receive_map.keys())), timeout=5, lock=mininode_lock)
+        wait_until(lambda: sorted(blocks) == sorted(list(self.nodes[2].p2p.block_receive_map.keys())), timeout=5, lock=mininode_lock)
 
         self.log.info("Check that each block was received only once")
         # The network thread uses a global lock on data access to the NodeConn objects when sending and receiving
         # messages. The test thread should acquire the global lock before accessing any NodeConn data to avoid locking
         # and synchronization issues. Note wait_until() acquires this global lock when testing the predicate.
         with mininode_lock:
-            for block in node2.block_receive_map.values():
+            for block in self.nodes[2].p2p.block_receive_map.values():
                 assert_equal(block, 1)
 
 if __name__ == '__main__':
diff --git a/test/functional/maxuploadtarget.py b/test/functional/maxuploadtarget.py
index 1f402798e..9c92aa1dc 100755
--- a/test/functional/maxuploadtarget.py
+++ b/test/functional/maxuploadtarget.py
@@ -49,19 +49,17 @@ class MaxUploadTest(BitcoinTestFramework):
         # Generate some old blocks
         self.nodes[0].generate(130)
 
-        # test_nodes[0] will only request old blocks
-        # test_nodes[1] will only request new blocks
-        # test_nodes[2] will test resetting the counters
-        test_nodes = []
-        connections = []
+        # p2p_conns[0] will only request old blocks
+        # p2p_conns[1] will only request new blocks
+        # p2p_conns[2] will test resetting the counters
+        p2p_conns = []
 
         for i in range(3):
-            test_nodes.append(TestNode())
-            connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_nodes[i]))
-            test_nodes[i].add_connection(connections[i])
+            p2p_conns.append(self.nodes[0].add_p2p_connection(TestNode()))
 
         NetworkThread().start() # Start up network handling in another thread
-        [x.wait_for_verack() for x in test_nodes]
+        for p2pc in p2p_conns:
+            p2pc.wait_for_verack()
 
         # Test logic begins here
 
@@ -83,7 +81,7 @@ class MaxUploadTest(BitcoinTestFramework):
         big_new_block = self.nodes[0].getbestblockhash()
         big_new_block = int(big_new_block, 16)
 
-        # test_nodes[0] will test what happens if we just keep requesting the
+        # p2p_conns[0] will test what happens if we just keep requesting the
         # the same big old block too many times (expect: disconnect)
 
         getdata_request = msg_getdata()
@@ -97,34 +95,34 @@ class MaxUploadTest(BitcoinTestFramework):
         # 576MB will be reserved for relaying new blocks, so expect this to
         # succeed for ~235 tries.
         for i in range(success_count):
-            test_nodes[0].send_message(getdata_request)
-            test_nodes[0].sync_with_ping()
-            assert_equal(test_nodes[0].block_receive_map[big_old_block], i+1)
+            p2p_conns[0].send_message(getdata_request)
+            p2p_conns[0].sync_with_ping()
+            assert_equal(p2p_conns[0].block_receive_map[big_old_block], i+1)
 
         assert_equal(len(self.nodes[0].getpeerinfo()), 3)
         # At most a couple more tries should succeed (depending on how long 
         # the test has been running so far).
         for i in range(3):
-            test_nodes[0].send_message(getdata_request)
-        test_nodes[0].wait_for_disconnect()
+            p2p_conns[0].send_message(getdata_request)
+        p2p_conns[0].wait_for_disconnect()
         assert_equal(len(self.nodes[0].getpeerinfo()), 2)
         self.log.info("Peer 0 disconnected after downloading old block too many times")
 
-        # Requesting the current block on test_nodes[1] should succeed indefinitely,
+        # Requesting the current block on p2p_conns[1] should succeed indefinitely,
         # even when over the max upload target.
         # We'll try 800 times
         getdata_request.inv = [CInv(2, big_new_block)]
         for i in range(800):
-            test_nodes[1].send_message(getdata_request)
-            test_nodes[1].sync_with_ping()
-            assert_equal(test_nodes[1].block_receive_map[big_new_block], i+1)
+            p2p_conns[1].send_message(getdata_request)
+            p2p_conns[1].sync_with_ping()
+            assert_equal(p2p_conns[1].block_receive_map[big_new_block], i+1)
 
         self.log.info("Peer 1 able to repeatedly download new block")
 
-        # But if test_nodes[1] tries for an old block, it gets disconnected too.
+        # But if p2p_conns[1] tries for an old block, it gets disconnected too.
         getdata_request.inv = [CInv(2, big_old_block)]
-        test_nodes[1].send_message(getdata_request)
-        test_nodes[1].wait_for_disconnect()
+        p2p_conns[1].send_message(getdata_request)
+        p2p_conns[1].wait_for_disconnect()
         assert_equal(len(self.nodes[0].getpeerinfo()), 1)
 
         self.log.info("Peer 1 disconnected after trying to download old block")
@@ -132,39 +130,38 @@ class MaxUploadTest(BitcoinTestFramework):
         self.log.info("Advancing system time on node to clear counters...")
 
         # If we advance the time by 24 hours, then the counters should reset,
-        # and test_nodes[2] should be able to retrieve the old block.
+        # and p2p_conns[2] should be able to retrieve the old block.
         self.nodes[0].setmocktime(int(time.time()))
-        test_nodes[2].sync_with_ping()
-        test_nodes[2].send_message(getdata_request)
-        test_nodes[2].sync_with_ping()
-        assert_equal(test_nodes[2].block_receive_map[big_old_block], 1)
+        p2p_conns[2].sync_with_ping()
+        p2p_conns[2].send_message(getdata_request)
+        p2p_conns[2].sync_with_ping()
+        assert_equal(p2p_conns[2].block_receive_map[big_old_block], 1)
 
         self.log.info("Peer 2 able to download old block")
 
-        [c.disconnect_node() for c in connections]
+        for i in range(3):
+            self.nodes[0].disconnect_p2p()
 
         #stop and start node 0 with 1MB maxuploadtarget, whitelist 127.0.0.1
         self.log.info("Restarting nodes with -whitelist=127.0.0.1")
         self.stop_node(0)
         self.start_node(0, ["-whitelist=127.0.0.1", "-maxuploadtarget=1", "-blockmaxsize=999000"])
 
-        #recreate/reconnect a test node
-        test_nodes = [TestNode()]
-        connections = [NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_nodes[0])]
-        test_nodes[0].add_connection(connections[0])
+        # Reconnect to self.nodes[0]
+        self.nodes[0].add_p2p_connection(TestNode())
 
         NetworkThread().start() # Start up network handling in another thread
-        test_nodes[0].wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         #retrieve 20 blocks which should be enough to break the 1MB limit
         getdata_request.inv = [CInv(2, big_new_block)]
         for i in range(20):
-            test_nodes[0].send_message(getdata_request)
-            test_nodes[0].sync_with_ping()
-            assert_equal(test_nodes[0].block_receive_map[big_new_block], i+1)
+            self.nodes[0].p2p.send_message(getdata_request)
+            self.nodes[0].p2p.sync_with_ping()
+            assert_equal(self.nodes[0].p2p.block_receive_map[big_new_block], i+1)
 
         getdata_request.inv = [CInv(2, big_old_block)]
-        test_nodes[0].send_and_ping(getdata_request)
+        self.nodes[0].p2p.send_and_ping(getdata_request)
         assert_equal(len(self.nodes[0].getpeerinfo()), 1) #node is still connected because of the whitelist
 
         self.log.info("Peer still connected after trying to download old block (whitelisted)")
diff --git a/test/functional/p2p-acceptblock.py b/test/functional/p2p-acceptblock.py
index 220b77636..fbe5a7802 100755
--- a/test/functional/p2p-acceptblock.py
+++ b/test/functional/p2p-acceptblock.py
@@ -78,14 +78,10 @@ class AcceptBlockTest(BitcoinTestFramework):
 
     def run_test(self):
         # Setup the p2p connections and start up the network thread.
-        test_node = NodeConnCB()   # connects to node0
-        min_work_node = NodeConnCB()  # connects to node1
-
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_node))
-        connections.append(NodeConn('127.0.0.1', p2p_port(1), self.nodes[1], min_work_node))
-        test_node.add_connection(connections[0])
-        min_work_node.add_connection(connections[1])
+        # test_node connects to node0 (not whitelisted)
+        test_node = self.nodes[0].add_p2p_connection(NodeConnCB())
+        # min_work_node connects to node1
+        min_work_node = self.nodes[1].add_p2p_connection(NodeConnCB())
 
         NetworkThread().start() # Start up network handling in another thread
 
@@ -209,12 +205,9 @@ class AcceptBlockTest(BitcoinTestFramework):
 
         # The node should have requested the blocks at some point, so
         # disconnect/reconnect first
-        connections[0].disconnect_node()
-        test_node.wait_for_disconnect()
 
-        test_node = NodeConnCB()   # connects to node (not whitelisted)
-        connections[0] = NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_node)
-        test_node.add_connection(connections[0])
+        self.nodes[0].disconnect_p2p()
+        test_node = self.nodes[0].add_p2p_connection(NodeConnCB())
 
         test_node.wait_for_verack()
         test_node.send_message(msg_block(block_h1f))
@@ -298,9 +291,8 @@ class AcceptBlockTest(BitcoinTestFramework):
         except AssertionError:
             test_node.wait_for_disconnect()
 
-            test_node = NodeConnCB()   # connects to node (not whitelisted)
-            connections[0] = NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_node)
-            test_node.add_connection(connections[0])
+            self.nodes[0].disconnect_p2p()
+            test_node = self.nodes[0].add_p2p_connection(NodeConnCB())
 
             NetworkThread().start() # Start up network handling in another thread
             test_node.wait_for_verack()
@@ -323,7 +315,5 @@ class AcceptBlockTest(BitcoinTestFramework):
         sync_blocks([self.nodes[0], self.nodes[1]])
         self.log.info("Successfully synced nodes 1 and 0")
 
-        [ c.disconnect_node() for c in connections ]
-
 if __name__ == '__main__':
     AcceptBlockTest().main()
diff --git a/test/functional/p2p-compactblocks.py b/test/functional/p2p-compactblocks.py
index e3f986d80..d2c4d3930 100755
--- a/test/functional/p2p-compactblocks.py
+++ b/test/functional/p2p-compactblocks.py
@@ -788,23 +788,12 @@ class CompactBlocksTest(BitcoinTestFramework):
 
     def run_test(self):
         # Setup the p2p connections and start up the network thread.
-        self.test_node = TestNode()
-        self.segwit_node = TestNode()
-        self.old_node = TestNode()  # version 1 peer <--> segwit node
-
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], self.test_node))
-        connections.append(NodeConn('127.0.0.1', p2p_port(1), self.nodes[1],
-                    self.segwit_node, services=NODE_NETWORK|NODE_WITNESS))
-        connections.append(NodeConn('127.0.0.1', p2p_port(1), self.nodes[1],
-                    self.old_node, services=NODE_NETWORK))
-        self.test_node.add_connection(connections[0])
-        self.segwit_node.add_connection(connections[1])
-        self.old_node.add_connection(connections[2])
+        self.test_node = self.nodes[0].add_p2p_connection(TestNode())
+        self.segwit_node = self.nodes[1].add_p2p_connection(TestNode(), services=NODE_NETWORK|NODE_WITNESS)
+        self.old_node = self.nodes[1].add_p2p_connection(TestNode(), services=NODE_NETWORK)
 
         NetworkThread().start()  # Start up network handling in another thread
 
-        # Test logic begins here
         self.test_node.wait_for_verack()
 
         # We will need UTXOs to construct transactions in later tests.
diff --git a/test/functional/p2p-feefilter.py b/test/functional/p2p-feefilter.py
index 8c92365ce..624278df4 100755
--- a/test/functional/p2p-feefilter.py
+++ b/test/functional/p2p-feefilter.py
@@ -48,25 +48,23 @@ class FeeFilterTest(BitcoinTestFramework):
         sync_blocks(self.nodes)
 
         # Setup the p2p connections and start up the network thread.
-        test_node = TestNode()
-        connection = NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_node)
-        test_node.add_connection(connection)
+        self.nodes[0].add_p2p_connection(TestNode())
         NetworkThread().start()
-        test_node.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         # Test that invs are received for all txs at feerate of 20 sat/byte
         node1.settxfee(Decimal("0.00020000"))
         txids = [node1.sendtoaddress(node1.getnewaddress(), 1) for x in range(3)]
-        assert(allInvsMatch(txids, test_node))
-        test_node.clear_invs()
+        assert(allInvsMatch(txids, self.nodes[0].p2p))
+        self.nodes[0].p2p.clear_invs()
 
         # Set a filter of 15 sat/byte
-        test_node.send_and_ping(msg_feefilter(15000))
+        self.nodes[0].p2p.send_and_ping(msg_feefilter(15000))
 
         # Test that txs are still being received (paying 20 sat/byte)
         txids = [node1.sendtoaddress(node1.getnewaddress(), 1) for x in range(3)]
-        assert(allInvsMatch(txids, test_node))
-        test_node.clear_invs()
+        assert(allInvsMatch(txids, self.nodes[0].p2p))
+        self.nodes[0].p2p.clear_invs()
 
         # Change tx fee rate to 10 sat/byte and test they are no longer received
         node1.settxfee(Decimal("0.00010000"))
@@ -82,14 +80,14 @@ class FeeFilterTest(BitcoinTestFramework):
         # as well.
         node0.settxfee(Decimal("0.00020000"))
         txids = [node0.sendtoaddress(node0.getnewaddress(), 1)]
-        assert(allInvsMatch(txids, test_node))
-        test_node.clear_invs()
+        assert(allInvsMatch(txids, self.nodes[0].p2p))
+        self.nodes[0].p2p.clear_invs()
 
         # Remove fee filter and check that txs are received again
-        test_node.send_and_ping(msg_feefilter(0))
+        self.nodes[0].p2p.send_and_ping(msg_feefilter(0))
         txids = [node1.sendtoaddress(node1.getnewaddress(), 1) for x in range(3)]
-        assert(allInvsMatch(txids, test_node))
-        test_node.clear_invs()
+        assert(allInvsMatch(txids, self.nodes[0].p2p))
+        self.nodes[0].p2p.clear_invs()
 
 if __name__ == '__main__':
     FeeFilterTest().main()
diff --git a/test/functional/p2p-fingerprint.py b/test/functional/p2p-fingerprint.py
index fe60c6cd4..4b6446fc5 100755
--- a/test/functional/p2p-fingerprint.py
+++ b/test/functional/p2p-fingerprint.py
@@ -14,7 +14,6 @@ from test_framework.blocktools import (create_block, create_coinbase)
 from test_framework.mininode import (
     CInv,
     NetworkThread,
-    NodeConn,
     NodeConnCB,
     msg_headers,
     msg_block,
@@ -77,11 +76,7 @@ class P2PFingerprintTest(BitcoinTestFramework):
     # This does not currently test that stale blocks timestamped within the
     # last month but that have over a month's worth of work are also withheld.
     def run_test(self):
-        node0 = NodeConnCB()
-
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], node0))
-        node0.add_connection(connections[0])
+        node0 = self.nodes[0].add_p2p_connection(NodeConnCB())
 
         NetworkThread().start()
         node0.wait_for_verack()
diff --git a/test/functional/p2p-leaktests.py b/test/functional/p2p-leaktests.py
index 1dc8f72cd..2499624f2 100755
--- a/test/functional/p2p-leaktests.py
+++ b/test/functional/p2p-leaktests.py
@@ -97,24 +97,13 @@ class P2PLeakTest(BitcoinTestFramework):
         self.extra_args = [['-banscore='+str(banscore)]]
 
     def run_test(self):
-        no_version_bannode = CNodeNoVersionBan()
-        no_version_idlenode = CNodeNoVersionIdle()
-        no_verack_idlenode = CNodeNoVerackIdle()
-        unsupported_service_bit5_node = CLazyNode()
-        unsupported_service_bit7_node = CLazyNode()
-
         self.nodes[0].setmocktime(1501545600)  # August 1st 2017
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], no_version_bannode, send_version=False))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], no_version_idlenode, send_version=False))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], no_verack_idlenode))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], unsupported_service_bit5_node, services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_5))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], unsupported_service_bit7_node, services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_7))
-        no_version_bannode.add_connection(connections[0])
-        no_version_idlenode.add_connection(connections[1])
-        no_verack_idlenode.add_connection(connections[2])
-        unsupported_service_bit5_node.add_connection(connections[3])
-        unsupported_service_bit7_node.add_connection(connections[4])
+
+        no_version_bannode = self.nodes[0].add_p2p_connection(CNodeNoVersionBan(), send_version=False)
+        no_version_idlenode = self.nodes[0].add_p2p_connection(CNodeNoVersionIdle(), send_version=False)
+        no_verack_idlenode = self.nodes[0].add_p2p_connection(CNodeNoVerackIdle())
+        unsupported_service_bit5_node = self.nodes[0].add_p2p_connection(CLazyNode(), services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_5)
+        unsupported_service_bit7_node = self.nodes[0].add_p2p_connection(CLazyNode(), services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_7)
 
         NetworkThread().start()  # Start up network handling in another thread
 
@@ -137,7 +126,8 @@ class P2PLeakTest(BitcoinTestFramework):
         assert not unsupported_service_bit5_node.connected
         assert not unsupported_service_bit7_node.connected
 
-        [conn.disconnect_node() for conn in connections]
+        for _ in range(5):
+            self.nodes[0].disconnect_p2p()
 
         # Wait until all connections are closed
         wait_until(lambda: len(self.nodes[0].getpeerinfo()) == 0)
@@ -152,13 +142,8 @@ class P2PLeakTest(BitcoinTestFramework):
         self.log.info("Service bits 5 and 7 are allowed after August 1st 2018")
         self.nodes[0].setmocktime(1533168000)  # August 2nd 2018
 
-        allowed_service_bit5_node = NodeConnCB()
-        allowed_service_bit7_node = NodeConnCB()
-
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], allowed_service_bit5_node, services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_5))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], allowed_service_bit7_node, services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_7))
-        allowed_service_bit5_node.add_connection(connections[5])
-        allowed_service_bit7_node.add_connection(connections[6])
+        allowed_service_bit5_node = self.nodes[0].add_p2p_connection(NodeConnCB(), services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_5)
+        allowed_service_bit7_node = self.nodes[0].add_p2p_connection(NodeConnCB(), services=NODE_NETWORK|NODE_UNSUPPORTED_SERVICE_BIT_7)
 
         NetworkThread().start()  # Network thread stopped when all previous NodeConnCBs disconnected. Restart it
 
diff --git a/test/functional/p2p-mempool.py b/test/functional/p2p-mempool.py
index 40fcde260..be467c422 100755
--- a/test/functional/p2p-mempool.py
+++ b/test/functional/p2p-mempool.py
@@ -19,16 +19,14 @@ class P2PMempoolTests(BitcoinTestFramework):
         self.extra_args = [["-peerbloomfilters=0"]]
 
     def run_test(self):
-        #connect a mininode
-        aTestNode = NodeConnCB()
-        node = NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], aTestNode)
-        aTestNode.add_connection(node)
+        # Add a p2p connection
+        self.nodes[0].add_p2p_connection(NodeConnCB())
         NetworkThread().start()
-        aTestNode.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         #request mempool
-        aTestNode.send_message(msg_mempool())
-        aTestNode.wait_for_disconnect()
+        self.nodes[0].p2p.send_message(msg_mempool())
+        self.nodes[0].p2p.wait_for_disconnect()
 
         #mininode must be disconnected at this point
         assert_equal(len(self.nodes[0].getpeerinfo()), 0)
diff --git a/test/functional/p2p-segwit.py b/test/functional/p2p-segwit.py
index 49de1a6e8..22da7f2db 100755
--- a/test/functional/p2p-segwit.py
+++ b/test/functional/p2p-segwit.py
@@ -1868,19 +1868,12 @@ class SegWitTest(BitcoinTestFramework):
 
     def run_test(self):
         # Setup the p2p connections and start up the network thread.
-        self.test_node = TestNode() # sets NODE_WITNESS|NODE_NETWORK
-        self.old_node = TestNode()  # only NODE_NETWORK
-        self.std_node = TestNode() # for testing node1 (fRequireStandard=true)
-
-        self.p2p_connections = [self.test_node, self.old_node]
-
-        self.connections = []
-        self.connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], self.test_node, services=NODE_NETWORK|NODE_WITNESS))
-        self.connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], self.old_node, services=NODE_NETWORK))
-        self.connections.append(NodeConn('127.0.0.1', p2p_port(1), self.nodes[1], self.std_node, services=NODE_NETWORK|NODE_WITNESS))
-        self.test_node.add_connection(self.connections[0])
-        self.old_node.add_connection(self.connections[1])
-        self.std_node.add_connection(self.connections[2])
+        # self.test_node sets NODE_WITNESS|NODE_NETWORK
+        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(), services=NODE_NETWORK)
+        # self.std_node is for testing node1 (fRequireStandard=true)
+        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 51d4769ef..14a3bf48f 100755
--- a/test/functional/p2p-timeouts.py
+++ b/test/functional/p2p-timeouts.py
@@ -39,46 +39,37 @@ class TimeoutsTest(BitcoinTestFramework):
 
     def run_test(self):
         # Setup the p2p connections and start up the network thread.
-        self.no_verack_node = TestNode() # never send verack
-        self.no_version_node = TestNode() # never send version (just ping)
-        self.no_send_node = TestNode() # never send anything
-
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], self.no_verack_node))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], self.no_version_node, send_version=False))
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], self.no_send_node, send_version=False))
-        self.no_verack_node.add_connection(connections[0])
-        self.no_version_node.add_connection(connections[1])
-        self.no_send_node.add_connection(connections[2])
+        no_verack_node = self.nodes[0].add_p2p_connection(TestNode())
+        no_version_node = self.nodes[0].add_p2p_connection(TestNode(), send_version=False)
+        no_send_node = self.nodes[0].add_p2p_connection(TestNode(), send_version=False)
 
         NetworkThread().start()  # Start up network handling in another thread
 
         sleep(1)
 
-        assert(self.no_verack_node.connected)
-        assert(self.no_version_node.connected)
-        assert(self.no_send_node.connected)
+        assert no_verack_node.connected
+        assert no_version_node.connected
+        assert no_send_node.connected
 
-        ping_msg = msg_ping()
-        connections[0].send_message(ping_msg)
-        connections[1].send_message(ping_msg)
+        no_verack_node.send_message(msg_ping())
+        no_version_node.send_message(msg_ping())
 
         sleep(30)
 
-        assert "version" in self.no_verack_node.last_message
+        assert "version" in no_verack_node.last_message
 
-        assert(self.no_verack_node.connected)
-        assert(self.no_version_node.connected)
-        assert(self.no_send_node.connected)
+        assert no_verack_node.connected
+        assert no_version_node.connected
+        assert no_send_node.connected
 
-        connections[0].send_message(ping_msg)
-        connections[1].send_message(ping_msg)
+        no_verack_node.send_message(msg_ping())
+        no_version_node.send_message(msg_ping())
 
         sleep(31)
 
-        assert(not self.no_verack_node.connected)
-        assert(not self.no_version_node.connected)
-        assert(not self.no_send_node.connected)
+        assert not no_verack_node.connected
+        assert not no_version_node.connected
+        assert not no_send_node.connected
 
 if __name__ == '__main__':
     TimeoutsTest().main()
diff --git a/test/functional/p2p-versionbits-warning.py b/test/functional/p2p-versionbits-warning.py
index f9bef2580..464ca5a31 100755
--- a/test/functional/p2p-versionbits-warning.py
+++ b/test/functional/p2p-versionbits-warning.py
@@ -64,16 +64,12 @@ class VersionBitsWarningTest(BitcoinTestFramework):
 
     def run_test(self):
         # Setup the p2p connection and start up the network thread.
-        test_node = TestNode()
-
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_node))
-        test_node.add_connection(connections[0])
+        self.nodes[0].add_p2p_connection(TestNode())
 
         NetworkThread().start() # Start up network handling in another thread
 
         # Test logic begins here
-        test_node.wait_for_verack()
+        self.nodes[0].p2p.wait_for_verack()
 
         # 1. Have the node mine one period worth of blocks
         self.nodes[0].generate(VB_PERIOD)
@@ -81,7 +77,7 @@ class VersionBitsWarningTest(BitcoinTestFramework):
         # 2. Now build one period of blocks on the tip, with < VB_THRESHOLD
         # blocks signaling some unknown bit.
         nVersion = VB_TOP_BITS | (1<<VB_UNKNOWN_BIT)
-        self.send_blocks_with_version(test_node, VB_THRESHOLD-1, nVersion)
+        self.send_blocks_with_version(self.nodes[0].p2p, VB_THRESHOLD-1, nVersion)
 
         # Fill rest of period with regular version blocks
         self.nodes[0].generate(VB_PERIOD - VB_THRESHOLD + 1)
@@ -92,7 +88,7 @@ class VersionBitsWarningTest(BitcoinTestFramework):
 
         # 3. Now build one period of blocks with >= VB_THRESHOLD blocks signaling
         # some unknown bit
-        self.send_blocks_with_version(test_node, VB_THRESHOLD, nVersion)
+        self.send_blocks_with_version(self.nodes[0].p2p, VB_THRESHOLD, nVersion)
         self.nodes[0].generate(VB_PERIOD - VB_THRESHOLD)
         # Might not get a versionbits-related alert yet, as we should
         # have gotten a different alert due to more than 51/100 blocks
diff --git a/test/functional/sendheaders.py b/test/functional/sendheaders.py
index 08de2ffa0..82cbc5a11 100755
--- a/test/functional/sendheaders.py
+++ b/test/functional/sendheaders.py
@@ -181,7 +181,7 @@ class SendHeadersTest(BitcoinTestFramework):
     # mine count blocks and return the new tip
     def mine_blocks(self, count):
         # Clear out last block announcement from each p2p listener
-        [ x.clear_last_announcement() for x in self.p2p_connections ]
+        [x.clear_last_announcement() for x in self.nodes[0].p2ps]
         self.nodes[0].generate(count)
         return int(self.nodes[0].getbestblockhash(), 16)
 
@@ -193,7 +193,7 @@ class SendHeadersTest(BitcoinTestFramework):
     def mine_reorg(self, length):
         self.nodes[0].generate(length) # make sure all invalidated blocks are node0's
         sync_blocks(self.nodes, wait=0.1)
-        for x in self.p2p_connections:
+        for x in self.nodes[0].p2ps:
             x.wait_for_block_announcement(int(self.nodes[0].getbestblockhash(), 16))
             x.clear_last_announcement()
 
@@ -206,18 +206,10 @@ class SendHeadersTest(BitcoinTestFramework):
 
     def run_test(self):
         # Setup the p2p connections and start up the network thread.
-        inv_node = TestNode()
-        test_node = TestNode()
-
-        self.p2p_connections = [inv_node, test_node]
-
-        connections = []
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], inv_node))
+        inv_node = self.nodes[0].add_p2p_connection(TestNode())
         # Set nServices to 0 for test_node, so no block download will occur outside of
         # direct fetching
-        connections.append(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test_node, services=NODE_WITNESS))
-        inv_node.add_connection(connections[0])
-        test_node.add_connection(connections[1])
+        test_node = self.nodes[0].add_p2p_connection(TestNode(), services=NODE_WITNESS)
 
         NetworkThread().start() # Start up network handling in another thread
 
diff --git a/test/functional/test_framework/test_node.py b/test/functional/test_framework/test_node.py
index 12dab57a0..8b28064c4 100755
--- a/test/functional/test_framework/test_node.py
+++ b/test/functional/test_framework/test_node.py
@@ -13,13 +13,15 @@ import os
 import subprocess
 import time
 
+from .authproxy import JSONRPCException
+from .mininode import NodeConn
 from .util import (
     assert_equal,
     get_rpc_proxy,
     rpc_url,
     wait_until,
+    p2p_port,
 )
-from .authproxy import JSONRPCException
 
 BITCOIND_PROC_WAIT_TIMEOUT = 60
 
@@ -31,9 +33,11 @@ class TestNode():
     - state about the node (whether it's running, etc)
     - a Python subprocess.Popen object representing the running process
     - an RPC connection to the node
+    - one or more P2P connections to the node
+
 
-    To make things easier for the test writer, a bit of magic is happening under the covers.
-    Any unrecognised messages will be dispatched to the RPC connection."""
+    To make things easier for the test writer, any unrecognised messages will
+    be dispatched to the RPC connection."""
 
     def __init__(self, i, dirname, extra_args, rpchost, timewait, binary, stderr, mocktime, coverage_dir):
         self.index = i
@@ -63,10 +67,12 @@ class TestNode():
         self.url = None
         self.log = logging.getLogger('TestFramework.node%d' % i)
 
-    def __getattr__(self, *args, **kwargs):
+        self.p2ps = []
+
+    def __getattr__(self, name):
         """Dispatches any unrecognised messages to the RPC connection."""
         assert self.rpc_connected and self.rpc is not None, "Error: no RPC connection"
-        return self.rpc.__getattr__(*args, **kwargs)
+        return getattr(self.rpc, name)
 
     def start(self, extra_args=None, stderr=None):
         """Start the node."""
@@ -119,6 +125,7 @@ class TestNode():
             self.stop()
         except http.client.CannotSendRequest:
             self.log.exception("Unable to stop node.")
+        del self.p2ps[:]
 
     def is_node_stopped(self):
         """Checks whether the node has stopped.
@@ -151,6 +158,38 @@ class TestNode():
         self.encryptwallet(passphrase)
         self.wait_until_stopped()
 
+    def add_p2p_connection(self, p2p_conn, **kwargs):
+        """Add a p2p connection to the node.
+
+        This method adds the p2p connection to the self.p2ps list and also
+        returns the connection to the caller."""
+        if 'dstport' not in kwargs:
+            kwargs['dstport'] = p2p_port(self.index)
+        if 'dstaddr' not in kwargs:
+            kwargs['dstaddr'] = '127.0.0.1'
+        self.p2ps.append(p2p_conn)
+        kwargs.update({'rpc': self.rpc, 'callback': p2p_conn})
+        p2p_conn.add_connection(NodeConn(**kwargs))
+
+        return p2p_conn
+
+    @property
+    def p2p(self):
+        """Return the first p2p connection
+
+        Convenience property - most tests only use a single p2p connection to each
+        node, so this saves having to write node.p2ps[0] many times."""
+        assert self.p2ps, "No p2p connection"
+        return self.p2ps[0]
+
+    def disconnect_p2p(self, index=0):
+        """Close the p2p connection to the node."""
+        # Connection could have already been closed by other end. Calling disconnect_p2p()
+        # on an already disconnected p2p connection is not an error.
+        if self.p2ps[index].connection is not None:
+            self.p2ps[index].connection.disconnect_node()
+        del self.p2ps[index]
+
 class TestNodeCLI():
     """Interface to bitcoin-cli for an individual node"""