api_test.py 文件源码

def test_001_trampoline_timeout(self):
        server_sock = eventlet.listen(('127.0.0.1', 0))
        bound_port = server_sock.getsockname()[1]

        def server(sock):
            client, addr = sock.accept()
            eventlet.sleep(0.1)
        server_evt = eventlet.spawn(server, server_sock)
        eventlet.sleep(0)
        try:
            desc = eventlet.connect(('127.0.0.1', bound_port))
            hubs.trampoline(desc, read=True, write=False, timeout=0.001)
        except eventlet.Timeout:
            pass  # test passed
        else:
            assert False, "Didn't timeout"

        server_evt.wait()
        check_hub()