(self)
| 125 | self.wait_until(lambda: peer_after()['bytessent_per_msg'].get('ping', 0) >= peer_before['bytessent_per_msg'].get('ping', 0) + 32, timeout=1) |
| 126 | |
| 127 | def test_getnetworkinfo(self): |
| 128 | self.log.info("Test getnetworkinfo") |
| 129 | info = self.nodes[0].getnetworkinfo() |
| 130 | assert_equal(info['networkactive'], True) |
| 131 | assert_equal(info['connections'], 2) |
| 132 | assert_equal(info['connections_in'], 1) |
| 133 | assert_equal(info['connections_out'], 1) |
| 134 | |
| 135 | with self.nodes[0].assert_debug_log(expected_msgs=['SetNetworkActive: false\n']): |
| 136 | self.nodes[0].setnetworkactive(state=False) |
| 137 | assert_equal(self.nodes[0].getnetworkinfo()['networkactive'], False) |
| 138 | # Wait a bit for all sockets to close |
| 139 | self.wait_until(lambda: self.nodes[0].getnetworkinfo()['connections'] == 0, timeout=3) |
| 140 | |
| 141 | with self.nodes[0].assert_debug_log(expected_msgs=['SetNetworkActive: true\n']): |
| 142 | self.nodes[0].setnetworkactive(state=True) |
| 143 | # Connect nodes both ways. |
| 144 | self.connect_nodes(0, 1) |
| 145 | self.connect_nodes(1, 0) |
| 146 | |
| 147 | info = self.nodes[0].getnetworkinfo() |
| 148 | assert_equal(info['networkactive'], True) |
| 149 | assert_equal(info['connections'], 2) |
| 150 | assert_equal(info['connections_in'], 1) |
| 151 | assert_equal(info['connections_out'], 1) |
| 152 | |
| 153 | # check the `servicesnames` field |
| 154 | network_info = [node.getnetworkinfo() for node in self.nodes] |
| 155 | for info in network_info: |
| 156 | assert_net_servicesnames(int(info["localservices"], 0x10), info["localservicesnames"]) |
| 157 | |
| 158 | # Check dynamically generated networks list in getnetworkinfo help output. |
| 159 | assert "(ipv4, ipv6, onion, i2p, cjdns)" in self.nodes[0].help("getnetworkinfo") |
| 160 | |
| 161 | def test_getaddednodeinfo(self): |
| 162 | self.log.info("Test getaddednodeinfo") |
no test coverage detected