test/functional/net.py

   1 #!/usr/bin/env python3
   2 # Copyright (c) 2017 The Bitcoin Core developers
   3 # Distributed under the MIT software license, see the accompanying
   4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
   5 """Test RPC calls related to net.
   6
   7 Tests correspond to code in rpc/net.cpp.
   8 """
   9
  10 import time
  11
  12 from test_framework.test_framework import BitcoinTestFramework
  13 from test_framework.util import (
  14     assert_equal,
  15     start_nodes,
  16     connect_nodes_bi,
  17 )
  18
  19
  20 class NetTest(BitcoinTestFramework):
  21     def __init__(self):
  22         super().__init__()
  23         self.setup_clean_chain = True
  24         self.num_nodes = 2
  25
  26     def setup_network(self):
  27         self.nodes = start_nodes(self.num_nodes, self.options.tmpdir)
  28         connect_nodes_bi(self.nodes, 0, 1)
  29         self.is_network_split = False
  30         self.sync_all()
  31
  32     def run_test(self):
  33         assert_equal(self.nodes[0].getnetworkinfo()['networkactive'], True)
  34         assert_equal(self.nodes[0].getnetworkinfo()['connections'], 2) # bilateral connection
  35
  36         self.nodes[0].setnetworkactive(False)
  37         assert_equal(self.nodes[0].getnetworkinfo()['networkactive'], False)
  38         timeout = 3
  39         while self.nodes[0].getnetworkinfo()['connections'] != 0:
  40             # Wait a bit for all sockets to close
  41             assert timeout > 0, 'not all connections closed in time'
  42             timeout -= 0.1
  43             time.sleep(0.1)
  44
  45         self.nodes[0].setnetworkactive(True)
  46         connect_nodes_bi(self.nodes, 0, 1)
  47         assert_equal(self.nodes[0].getnetworkinfo()['networkactive'], True)
  48         assert_equal(self.nodes[0].getnetworkinfo()['connections'], 2)
  49
  50
  51 if __name__ == '__main__':
  52     NetTest().main()