Function test_upstream_error

test/mitmproxy/proxy/layers/http/test_http3.py:614–651

(tctx: context.Context)

def test_upstream_error(tctx: context.Context):
    playbook, cff = start_h3_proxy(tctx)
    flow = tutils.Placeholder(HTTPFlow)
    server = tutils.Placeholder(connection.Server)
    err = tutils.Placeholder(bytes)
    assert (
        playbook
        # request client
        >> cff.receive_headers(example_request_headers, end_stream=True)
        << (request := http.HttpRequestHeadersHook(flow))
        << cff.send_decoder()  # for receive_headers
        >> tutils.reply(to=request)
        << http.HttpRequestHook(flow)
        >> tutils.reply()
        # request server
        << commands.OpenConnection(server)
        >> tutils.reply("oops server <> error")
        << http.HttpErrorHook(flow)
        >> tutils.reply()
        << cff.send_headers(
            [
                (b":status", b"502"),
                (b"server", version.MITMPROXY.encode()),
                (b"content-type", b"text/html"),
            ]
        )
        << quic.SendQuicStreamData(
            tctx.client,
            stream_id=0,
            data=err,
            end_stream=True,
        )
        >> cff.receive_decoder()  # for send_headers
    )
    assert cff.is_done
    data = decode_frame(FrameType.DATA, err())
    assert b"502 Bad Gateway" in data
    assert b"server &lt;&gt; error" in data


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