Function test_gzip

benchmarks/gzip_benchmark.py:225–252

(benchmark: BenchmarkFixture, case: BenchmarkCase)

@pytest.mark.parametrize("case", CASES, ids=lambda case: case.id)
@pytest.mark.benchmark(max_time=0.5, max_rounds=10)
def test_gzip(benchmark: BenchmarkFixture, case: BenchmarkCase) -> None:
    # Payload construction is intentionally outside the measured region. Cases
    # are function-scoped, so only one source payload is resident at a time.
    payload = make_payload(case.payload_kind, case.size)
    messages: tuple[Message, ...] = (
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [(b"content-type", b"application/json"), (b"content-length", str(len(payload)).encode())],
        },
        {"type": "http.response.body", "body": payload},
    )
    app = GZipMiddleware(StaticResponseApp(messages), minimum_size=0, compresslevel=case.level)
    scope: Scope = {"type": "http", "headers": [(b"accept-encoding", b"gzip")]}
    with closing(ASGIRunner()) as runner:
        sent = benchmark.pedantic(runner.run, args=(app, scope), rounds=1)

    assert len(sent) == 2
    assert sent[0]["type"] == "http.response.start"
    assert (b"content-encoding", b"gzip") in sent[0]["headers"]
    assert sent[1]["type"] == "http.response.body"
    compressed = sent[1]["body"]
    assert isinstance(compressed, bytes)

    assert gzip.decompress(compressed) == payload
    benchmark.extra_info["input_bytes"] = len(payload)
    benchmark.extra_info["output_bytes"] = len(compressed)
    benchmark.extra_info["compression_ratio"] = len(compressed) / len(payload)


@pytest.mark.benchmark(max_time=0.5, max_rounds=1)