Repository: bitcoin/bitcoin
File: test/functional/p2p_handshake.py (Python)

Code before:
    def run_test(self):
        node = self.nodes[0]
        self.log.info("Check that lacking desired service flags leads to disconnect (non-pruned peers)")
        self.test_desirable_service_flags(node, [NODE_NONE, NODE_NETWORK, NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_FULL, expect_disconnect=True)
        self.test_desirable_service_flags(node, [NODE_NETWORK | NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_FULL, expect_disconnect=False)

        self.log.info("Check that limited peers are only desired if the local chain is close to the tip (<24h)")
        self.generate_at_mocktime(int(time.time()) - 25 * 3600)  # tip outside the 24h window, should fail
        self.test_desirable_service_flags(node, [NODE_NETWORK_LIMITED | NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_FULL, expect_disconnect=True)
        self.generate_at_mocktime(int(time.time()) - 23 * 3600)  # tip inside the 24h window, should succeed
        self.test_desirable_service_flags(node, [NODE_NETWORK_LIMITED | NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_PRUNED, expect_disconnect=False)

        self.log.info("Check that feeler connections get disconnected immediately")
        with node.assert_debug_log([f"feeler connection completed"]):
            self.add_outbound_connection(node, "feeler", NODE_NONE, wait_for_disconnect=True)

        # TODO: re-add test introduced in commit 5d2fb14bafe4e80c0a482d99e5ebde07c477f000
        # ("test: p2p: check that connecting to ourself leads to disconnect") once
        # the race condition causing issue #30368 is fixed


if __name__ == '__main__':
    P2PHandshakeTest().main()


Review comment (bug):
```suggestion
            self.wait_until(lambda: len(node.getpeerinfo()) == 0)
```

nit in the last commit: I am not a fan of using `assert_debug_log` to sync the test logic. Otherwise the test may fail on slow machines.

If you want to keep it, please make the sync explicit by increasing the timeout of `assert_debug_log`. Otherwise, you can do a proper wait (as suggested) by moving the `wait_until` into the inner scope.

Code after:

    def run_test(self):
        node = self.nodes[0]
        self.log.info("Check that lacking desired service flags leads to disconnect (non-pruned peers)")
        self.test_desirable_service_flags(node, [NODE_NONE, NODE_NETWORK, NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_FULL, expect_disconnect=True)
        self.test_desirable_service_flags(node, [NODE_NETWORK | NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_FULL, expect_disconnect=False)

        self.log.info("Check that limited peers are only desired if the local chain is close to the tip (<24h)")
        self.generate_at_mocktime(int(time.time()) - 25 * 3600)  # tip outside the 24h window, should fail
        self.test_desirable_service_flags(node, [NODE_NETWORK_LIMITED | NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_FULL, expect_disconnect=True)
        self.generate_at_mocktime(int(time.time()) - 23 * 3600)  # tip inside the 24h window, should succeed
        self.test_desirable_service_flags(node, [NODE_NETWORK_LIMITED | NODE_WITNESS],
                                          DESIRABLE_SERVICE_FLAGS_PRUNED, expect_disconnect=False)

        self.log.info("Check that feeler connections get disconnected immediately")
        with node.assert_debug_log([f"feeler connection completed"]):
            self.add_outbound_connection(node, "feeler", NODE_NONE, wait_for_disconnect=True)

        self.log.info("Check that connecting to ourself leads to immediate disconnect")
        with node.assert_debug_log(["connected to self", "disconnecting"]):
            node_listen_addr = f"127.0.0.1:{p2p_port(0)}"
            node.addconnection(node_listen_addr, "outbound-full-relay", self.options.v2transport)
            self.wait_until(lambda: len(node.getpeerinfo()) == 0)


if __name__ == '__main__':
    P2PHandshakeTest().main()