Function _test_echo

test/mitmproxy/addons/test_proxyserver.py:687–775

(client: H3Client, strict: bool)

async def _test_echo(client: H3Client, strict: bool) -> None:
    def assert_no_data(response: H3Response):
        if strict:
            assert response.data is None
        else:
            assert not response.data

    headers = [
        (b":scheme", b"https"),
        (b":authority", b"example.mitmproxy.org"),
        (b":method", b"GET"),
        (b":path", b"/test"),
    ]
    r1 = await client.request(
        headers=headers + [(b"x-request", b"justheaders")],
        data=None,
        trailers=None,
    ).wait_result()
    assert r1.headers == [
        (b":status", b"200"),
        (b"x-response", b"justheaders"),
    ]
    assert_no_data(r1)
    assert r1.trailers is None

    r2 = await client.request(
        headers=headers + [(b"x-request", b"hasdata")],
        data=b"echo",
        trailers=None,
    ).wait_result()
    assert r2.headers == [
        (b":status", b"200"),
        (b"x-response", b"hasdata"),
    ]
    assert r2.data == b"echo"
    assert r2.trailers is None

    r3 = await client.request(
        headers=headers + [(b"x-request", b"nodata")],
        data=None,
        trailers=[(b"x-request", b"buttrailers")],
    ).wait_result()
    assert r3.headers == [
        (b":status", b"200"),
        (b"x-response", b"nodata"),
    ]
    assert_no_data(r3)
    assert r3.trailers == [(b"x-response", b"buttrailers")]

    r4 = await client.request(
        headers=headers + [(b"x-request", b"this")],
        data=b"has",
        trailers=[(b"x-request", b"everything")],
    ).wait_result()
    assert r4.headers == [
        (b":status", b"200"),
        (b"x-response", b"this"),
    ]