integration_tests.py 文件源码

def connect_socket(url, timeout=TIMEOUT, **kwargs):
    """set up a websocket and return the socket connection object"""

    signal.signal(
        signal.SIGALRM, 
        lambda s, f: timeout_handler(s, f, f'connecting ({timeout}s)')
    )
    signal.alarm(timeout)
    try:
        sock = create_connection(url, **kwargs)
        signal.alarm(0)
        return sock
    except Exception:
        signal.alarm(0)
        print(f'[X] Failed to connect, is runserver running on {url}?')
        raise
    except Exception:
        signal.alarm(0)
        raise