Function ws_run

test/modules/http2/test_800_websockets.py:26–84

Run the h2ws test client in various scenarios with given input and timings. :param env: the test environment :param path: the path on the Apache server to CONNECt to :param authority: the host:port to use as :param do_input: a Callable for sending input to h2ws :param in

(env: H2TestEnv, path, authority=None, do_input=None, inbytes=None,
           send_close=True, timeout=5, scenario='ws-stdin',
           wait_close: float = 0.0)

def ws_run(env: H2TestEnv, path, authority=None, do_input=None, inbytes=None,
           send_close=True, timeout=5, scenario='ws-stdin',
           wait_close: float = 0.0) -> Tuple[ExecResult, List[str], List[WsFrame]]:
    """ Run the h2ws test client in various scenarios with given input and
        timings.
    :param env: the test environment
    :param path: the path on the Apache server to CONNECt to
    :param authority: the host:port to use as
    :param do_input: a Callable for sending input to h2ws
    :param inbytes: fixed bytes to send to h2ws, unless do_input is given
    :param send_close: send a CLOSE WebSockets frame at the end
    :param timeout: timeout for waiting on h2ws to finish
    :param scenario: name of scenario h2ws should run in
    :param wait_close: time to wait before closing input
    :return: ExecResult with exit_code/stdout/stderr of run
    """
    h2ws = os.path.join(env.clients_dir, 'h2ws')
    if not os.path.exists(h2ws):
        pytest.fail(f'test client not build: {h2ws}')
    if authority is None:
        authority = f'cgi.{env.http_tld}:{env.http_port}'
    args = [
        h2ws, '-vv', '-c', f'localhost:{env.http_port}',
        f'ws://{authority}{path}',
        scenario
    ]
    # we write all output to files, because we manipulate input timings
    # and would run in deadlock situations with h2ws blocking operations
    # because its output is not consumed
    start = datetime.now()
    with open(f'{env.gen_dir}/h2ws.stdout', 'w') as fdout:
        with open(f'{env.gen_dir}/h2ws.stderr', 'w') as fderr:
            proc = subprocess.Popen(args=args, stdin=subprocess.PIPE,
                                    stdout=fdout, stderr=fderr)
            if do_input is not None:
                do_input(proc)
            elif inbytes is not None:
                proc.stdin.write(inbytes)
                proc.stdin.flush()

            if wait_close > 0:
                time.sleep(wait_close)
            try:
                inbytes = WsFrame.client_close(code=1000).to_network() if send_close else None
                proc.communicate(input=inbytes, timeout=timeout)
            except subprocess.TimeoutExpired:
                log.error(f'ws_run: timeout expired')
                proc.kill()
                proc.communicate(timeout=timeout)
    end = datetime.now()
    lines = open(f'{env.gen_dir}/h2ws.stdout').read().splitlines()
    infos = [line for line in lines if re.match(r'^\[\d+] ', line)]
    hex_content = ' '.join([line for line in lines if not re.match(r'^\[\d+] ', line)])
    if len(infos) > 0 and infos[0] == '[1] :status: 200':
        frames = WsFrameReader.parse(bytearray.fromhex(hex_content))
    else:
        frames = bytearray.fromhex(hex_content)
    return ExecResult(args=args, exit_code=proc.returncode,