Function test_request_trailers

test/mitmproxy/proxy/layers/http/test_http2.py:225–266

(tctx: Context, open_h2_server_conn: Server, stream)

@pytest.mark.parametrize("stream", ["stream", ""])
def test_request_trailers(tctx: Context, open_h2_server_conn: Server, stream):
    playbook, cff = start_h2_client(tctx)
    tctx.server = open_h2_server_conn

    def enable_streaming(flow: HTTPFlow):
        flow.request.stream = bool(stream)

    flow = Placeholder(HTTPFlow)
    server_data1 = Placeholder(bytes)
    server_data2 = Placeholder(bytes)
    (
        playbook
        >> DataReceived(
            tctx.client,
            cff.build_headers_frame(example_request_headers).serialize()
            + cff.build_data_frame(b"Hello, World!").serialize(),
        )
        << http.HttpRequestHeadersHook(flow)
        >> reply(side_effect=enable_streaming)
    )
    if stream:
        playbook << SendData(tctx.server, server_data1)
    assert (
        playbook
        >> DataReceived(
            tctx.client,
            cff.build_headers_frame(
                example_request_trailers, flags=["END_STREAM"]
            ).serialize(),
        )
        << http.HttpRequestHook(flow)
        >> reply()
        << SendData(tctx.server, server_data2)
    )
    frames = decode_frames(server_data1.setdefault(b"") + server_data2())
    assert [type(x) for x in frames] == [
        hyperframe.frame.SettingsFrame,
        hyperframe.frame.WindowUpdateFrame,
        hyperframe.frame.HeadersFrame,
        hyperframe.frame.DataFrame,
        hyperframe.frame.HeadersFrame,
    ]


def test_upstream_error(tctx):