Function test_upstream_error

test/mitmproxy/proxy/layers/http/test_http2.py:269–300

(tctx)

def test_upstream_error(tctx):
    playbook, cff = start_h2_client(tctx)
    flow = Placeholder(HTTPFlow)
    server = Placeholder(Server)
    err = Placeholder(bytes)
    assert (
        playbook
        >> DataReceived(
            tctx.client,
            cff.build_headers_frame(
                example_request_headers, flags=["END_STREAM"]
            ).serialize(),
        )
        << http.HttpRequestHeadersHook(flow)
        >> reply()
        << http.HttpRequestHook(flow)
        >> reply()
        << OpenConnection(server)
        >> reply("oops server <> error")
        << http.HttpErrorHook(flow)
        >> reply()
        << SendData(tctx.client, err)
    )
    frames = decode_frames(err())
    assert [type(x) for x in frames] == [
        hyperframe.frame.HeadersFrame,
        hyperframe.frame.DataFrame,
    ]
    d = frames[1]
    assert isinstance(d, hyperframe.frame.DataFrame)
    assert b"502 Bad Gateway" in d.data
    assert b"server &lt;&gt; error" in d.data


@pytest.mark.parametrize("trailers", ["trailers", ""])