brain-of-reese/tests/unit/test_compression.py

"""Unit tests: ``app.core.compression.SelectiveGZipMiddleware`` (phase 167, D1, LOCKED A1).

Pins the three contracts of the selective-gzip layer on a tiny ASGI app —

* a ~50 KB JSON route compresses (``Content-Encoding: gzip``, the body
  ``gzip.decompress``es to the original, ``Vary`` mentions
  ``accept-encoding``), an ``Accept-Encoding: identity`` client gets
  identity bytes, a ~50-byte route stays under the floor, and a response
  that already set ``content-encoding`` is forwarded byte-identical (never
  double-encoded — the fronting-nginx contract);

* the EXACT-path skip: a route named ``/api/chat`` is NOT compressed and
  streams frame-by-frame (the raw-ASGI interleaving proof: every frame
  reaches the client's ``send`` BEFORE the app yields the next one — a
  buffering middleware fails this deterministically — plus the
  ``list(stream)`` client idiom the existing SSE tests use), while
  ``/api/chat-images/x`` shares the prefix and MUST compress;

* non-HTTP scopes (``lifespan``/``websocket``) pass straight through.

Plus the config kill switch: ``compress_min_bytes=0`` → ``create_app``
adds no gzip layer at all; ``-1`` → the validator refuses startup naming
the field; the middleware sits INSIDE ``SecurityHeadersMiddleware`` (A20)
and OUTSIDE ``CachingMiddleware``.
"""
from __future__ import annotations

import asyncio
import gzip
import json
from typing import Any

import pytest
from fastapi.testclient import TestClient
from pydantic import ValidationError
from starlette.applications import Starlette
from starlette.middleware.gzip import GZipMiddleware
from starlette.requests import Request
from starlette.responses import Response, StreamingResponse
from starlette.routing import Route
from starlette.types import Message, Receive, Scope, Send

import app.main as main_mod
from app.config import Settings
from app.core.compression import DEFAULT_SKIP_PATHS, SelectiveGZipMiddleware


def _settings(**kwargs: Any) -> Settings:
    """Build Settings without reading a .env file (the test_config pattern)."""
    kwargs.setdefault("_env_file", None)
    return Settings(**kwargs)  # pyright: ignore[reportCallIssue]  (kwarg exists at runtime)


# ~50 KB of highly compressible JSON — the /api/docs/tree shape class.
_BIG_BODY = json.dumps(
    {
        "items": [
            {"n": i, "path": f"source/docs/file-{i}.md", "title": f"Doc {i}"}
            for i in range(1000)
        ]
    }
).encode()

# ~50 bytes — comfortably under the 860-byte floor (compression would inflate).
_SMALL_BODY = json.dumps({"ok": True, "note": "intentionally tiny body"}).encode("utf-8")

# Five SSE frames (~2 KB total — ABOVE the floor, so a missing skip cannot
# hide behind the minimum_size exemption).
_SSE_FRAMES = [
    f"event: thinking\ndata: {'x' * 400}\n\n".encode() for _ in range(5)
]

_PRE_ENCODED_BODY = gzip.compress(_BIG_BODY)


def _inner_app() -> Starlette:
    async def big(request: Request) -> Response:
        return Response(_BIG_BODY, media_type="application/json")

    async def small(request: Request) -> Response:
        return Response(_SMALL_BODY, media_type="application/json")

    async def chat(request: Request) -> StreamingResponse:
        async def frames():
            for frame in _SSE_FRAMES:
                yield frame

        return StreamingResponse(frames(), media_type="text/event-stream")

    async def chat_images(request: Request) -> Response:
        # Shares the /api/chat PREFIX but is a different endpoint — the
        # skip must be EXACT, so this one compresses.
        return Response(_BIG_BODY, media_type="application/json")

    async def pre_encoded(request: Request) -> Response:
        # A body that arrived already encoded (e.g. from a future CDN
        # layer): the responder skips it, never double-encodes.
        return Response(
            _PRE_ENCODED_BODY,
            media_type="application/json",
            headers={"Content-Encoding": "gzip"},
        )

    return Starlette(
        routes=[
            Route("/big", big),
            Route("/small", small),
            Route("/api/chat", chat, methods=["POST"]),
            Route("/api/chat-images/x", chat_images),
            Route("/pre-encoded", pre_encoded),
        ]
    )


def _client(minimum_size: int = 860) -> TestClient:
    return TestClient(
        SelectiveGZipMiddleware(_inner_app(), minimum_size=minimum_size)
    )


# ---------------------------------------------------------------------------
# The gzip path (Accept-Encoding negotiation, the floor, the pre-encoded skip)
# ---------------------------------------------------------------------------


def test_large_json_compresses_decompresses_and_sets_vary() -> None:
    client = _client()
    response = client.get("/big", headers={"Accept-Encoding": "gzip"})
    assert response.status_code == 200
    assert response.headers["content-encoding"] == "gzip"
    assert "accept-encoding" in response.headers.get("vary", "").lower()
    # httpx transparently decodes: the documented JSON arrives whole.
    assert response.content == _BIG_BODY
    assert len(response.content) > 860  # the seed really is above the floor

    # The RAW wire bytes are a real gzip stream of the original body.
    with client.stream("GET", "/big", headers={"Accept-Encoding": "gzip"}) as streamed:
        raw = b"".join(list(streamed.iter_raw()))
    assert gzip.decompress(raw) == _BIG_BODY
    assert len(raw) < len(_BIG_BODY)  # ~8-12x on this JSON; order of magnitude suffices


def test_identity_client_is_not_compressed() -> None:
    client = _client()
    response = client.get("/big", headers={"Accept-Encoding": "identity"})
    assert response.status_code == 200
    assert "content-encoding" not in response.headers
    assert response.content == _BIG_BODY


def test_small_response_stays_under_the_floor() -> None:
    client = _client()
    response = client.get("/small", headers={"Accept-Encoding": "gzip"})
    assert response.status_code == 200
    assert "content-encoding" not in response.headers
    assert response.content == _SMALL_BODY


def test_pre_encoded_body_is_never_double_encoded() -> None:
    client = _client()
    with client.stream("GET", "/pre-encoded", headers={"Accept-Encoding": "gzip"}) as streamed:
        assert streamed.headers["content-encoding"] == "gzip"
        raw = b"".join(list(streamed.iter_raw()))
    # Byte-identical passthrough of the already-encoded body (Starlette's
    # "skip when content-encoding is set" rule — the nginx coexistence pin).
    assert raw == _PRE_ENCODED_BODY
    assert gzip.decompress(raw) == _BIG_BODY


# ---------------------------------------------------------------------------
# The EXACT-path skip (A15: the SSE stream is never compressed, never buffered)
# ---------------------------------------------------------------------------


def test_default_skip_paths_constant() -> None:
    assert DEFAULT_SKIP_PATHS == ("/api/chat",)


def test_chat_route_is_not_compressed_and_streams_whole() -> None:
    """The client-level pin (the ``list(stream)`` idiom the SSE tests use):
    a route named /api/chat carries NO Content-Encoding and its raw bytes
    are exactly the SSE frames, concatenated, above the size floor."""
    client = _client()
    with client.stream(
        "POST", "/api/chat", json={"message": "hi"}, headers={"Accept-Encoding": "gzip"}
    ) as streamed:
        assert streamed.status_code == 200
        assert "content-encoding" not in streamed.headers
        chunks = list(streamed.iter_raw())
    assert b"".join(chunks) == b"".join(_SSE_FRAMES)
    assert sum(len(c) for c in chunks) > 860  # the skip beats the floor, not size


def test_skipped_stream_arrives_frame_by_frame_uncompressed() -> None:
    """The frame-by-frame proof (deterministic, no timing): every frame
    reaches the client's ``send`` BEFORE the app produces the next one.
    A buffering middleware (BaseHTTPMiddleware, or a gzip pass) yields
    all ``app:*`` events first — this strict interleaving fails then."""
    events: list[str] = []  # shared: the app logs a yield, send logs delivery
    sent: list[Message] = []

    async def sse_app(scope: Scope, receive: Receive, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [(b"content-type", b"text/event-stream")],
            }
        )
        for i, frame in enumerate(_SSE_FRAMES):
            events.append(f"app:frame-{i}")
            await send({"type": "http.response.body", "body": frame, "more_body": True})
        events.append("app:final")
        await send({"type": "http.response.body", "body": b"", "more_body": False})

    async def recording_send(message: Message) -> None:
        events.append(f"client:{message['type']}")
        sent.append(message)

    async def receive() -> Message:
        return {"type": "http.disconnect"}

    middleware = SelectiveGZipMiddleware(sse_app, minimum_size=860)
    asyncio.run(
        middleware(
            {"type": "http", "path": "/api/chat", "headers": [(b"accept-encoding", b"gzip")]},
            receive,
            recording_send,
        )
    )

    expected = ["client:http.response.start"]
    for i in range(len(_SSE_FRAMES)):
        expected += [f"app:frame-{i}", "client:http.response.body"]
    expected += ["app:final", "client:http.response.body"]
    assert events == expected

    start = sent[0]
    headers = {k.decode().lower() for k, _ in start["headers"]}  # pyright: ignore[reportGeneralTypeIssues]
    assert "content-encoding" not in headers
    body = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
    assert body == b"".join(_SSE_FRAMES)


def _raw_harness(
    *,
    path: str,
    app: Any,
) -> list[Message]:
    """Drive the middleware at the raw ASGI level; return the exact
    messages that reached the client's ``send``."""
    sent: list[Message] = []

    async def recording_send(message: Message) -> None:
        sent.append(message)

    async def receive() -> Message:
        return {"type": "http.disconnect"}

    scope: Scope = {
        "type": "http",
        "path": path,
        "headers": [(b"accept-encoding", b"gzip")],
    }
    middleware = SelectiveGZipMiddleware(app, minimum_size=860)
    asyncio.run(middleware(scope, receive, recording_send))
    return sent


def test_same_stream_at_a_non_skipped_path_goes_through_gzip() -> None:
    """The skip is PATH-driven: the identical streaming app mounted at a
    non-skipped path is compressed by the composed GZipMiddleware."""

    async def json_stream_app(scope: Scope, receive: Receive, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [(b"content-type", b"application/json")],
            }
        )
        for i, frame in enumerate(_SSE_FRAMES):
            await send(
                {"type": "http.response.body", "body": frame, "more_body": i < len(_SSE_FRAMES) - 1}
            )

    sent = _raw_harness(path="/api/other-stream", app=json_stream_app)
    start_headers = {
        k.decode().lower(): v.decode() for k, v in sent[0]["headers"]  # pyright: ignore[reportGeneralTypeIssues]
    }
    assert start_headers.get("content-encoding") == "gzip"
    raw = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
    assert gzip.decompress(raw) == b"".join(_SSE_FRAMES)


def test_reframed_small_body_stays_identity_under_the_floor() -> None:
    """The CachingMiddleware seam (starlette 1.x ``BaseHTTPMiddleware``
    re-frames every response it passes up: chunk with ``more_body=True``
    + empty terminator): a body whose ``content-length`` is below the
    floor is re-buffered into ONE buffered message so Starlette's own
    small-response exemption fires — identity bytes, no ``Vary`` append."""
    payload = b'{"status":"ok","version":"0.1.0"}'  # 33 bytes — the /api/health class

    async def reframed_app(scope: Scope, receive: Receive, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [
                    (b"content-type", b"application/json"),
                    (b"content-length", str(len(payload)).encode()),
                ],
            }
        )
        await send({"type": "http.response.body", "body": payload, "more_body": True})
        await send({"type": "http.response.body", "body": b"", "more_body": False})

    sent = _raw_harness(path="/api/health", app=reframed_app)
    start_headers = {
        k.decode().lower(): v.decode() for k, v in sent[0]["headers"]  # pyright: ignore[reportGeneralTypeIssues]
    }
    assert "content-encoding" not in start_headers
    assert "vary" not in start_headers  # the exemption adds nothing
    body = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
    assert body == payload


def test_lying_content_length_that_outgrows_the_floor_compresses() -> None:
    """A body that outgrows the floor despite a small ``content-length``
    stops the hold and streams through the normal compressing path — the
    buffer is bounded by the floor and correctness never depends on the
    header being honest."""

    async def lying_app(scope: Scope, receive: Receive, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [
                    (b"content-type", b"application/json"),
                    (b"content-length", b"5"),  # a lie
                ],
            }
        )
        await send({"type": "http.response.body", "body": _BIG_BODY, "more_body": True})
        await send({"type": "http.response.body", "body": b"", "more_body": False})

    sent = _raw_harness(path="/api/lias", app=lying_app)
    start_headers = {
        k.decode().lower(): v.decode() for k, v in sent[0]["headers"]  # pyright: ignore[reportGeneralTypeIssues]
    }
    assert start_headers.get("content-encoding") == "gzip"
    raw = b"".join(m.get("body", b"") for m in sent[1:])  # pyright: ignore[reportGeneralTypeIssues]
    assert gzip.decompress(raw) == _BIG_BODY


def test_non_body_message_while_holding_flushes_untouched() -> None:
    """``http.response.pathsend`` (or anything unexpected) arriving while
    the hold is active: the held messages and the message itself are
    forwarded byte-for-byte — the hold never swallows a frame."""

    async def pathsend_app(scope: Scope, receive: Receive, send: Send) -> None:
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [
                    (b"content-type", b"text/plain"),
                    (b"content-length", b"4"),
                ],
            }
        )
        await send({"type": "http.response.pathsend", "path": "/tmp/file"})

    sent = _raw_harness(path="/assets/file", app=pathsend_app)
    assert [m["type"] for m in sent] == ["http.response.start", "http.response.pathsend"]
    start_headers = {k.decode().lower() for k, _ in sent[0]["headers"]}  # pyright: ignore[reportGeneralTypeIssues]
    assert "content-encoding" not in start_headers


def test_chat_images_prefix_still_compresses() -> None:
    """The exact-match pin: /api/chat-images/x shares the /api/chat
    prefix and MUST go through the gzip path."""
    client = _client()
    response = client.get("/api/chat-images/x", headers={"Accept-Encoding": "gzip"})
    assert response.status_code == 200
    assert response.headers["content-encoding"] == "gzip"
    assert response.content == _BIG_BODY


# ---------------------------------------------------------------------------
# Non-HTTP scopes pass straight through
# ---------------------------------------------------------------------------


def test_non_http_scopes_pass_through_untouched() -> None:
    seen: list[str] = []

    async def probe_app(scope: Scope, receive: Receive, send: Send) -> None:
        seen.append(scope["type"])

    async def receive() -> Message:
        return {"type": "http.disconnect"}

    async def send(message: Message) -> None:  # pragma: no cover - app sends nothing
        raise AssertionError("no ASGI messages expected")

    middleware = SelectiveGZipMiddleware(probe_app, minimum_size=860)
    for scope_type in ("lifespan", "websocket"):
        asyncio.run(
            middleware({"type": scope_type, "path": "/api/anything"}, receive, send)
        )
    assert seen == ["lifespan", "websocket"]


def test_composes_starlette_gzip_with_minimum_size() -> None:
    """The middleware reuses Starlette's GZipMiddleware, it does not fork
    the compression math (D1)."""

    async def app(scope: Scope, receive: Receive, send: Send) -> None: ...

    middleware = SelectiveGZipMiddleware(app, minimum_size=1234)
    assert isinstance(middleware.gzip, GZipMiddleware)
    assert middleware.gzip.minimum_size == 1234
    assert middleware.skip_paths == DEFAULT_SKIP_PATHS


# ---------------------------------------------------------------------------
# Config: the BOR_COMPRESS_MIN_BYTES kill switch and the loud negative
# ---------------------------------------------------------------------------


def _middleware_names(app: Any) -> list[str]:
    """Middleware class names, outermost first (``user_middleware`` order)."""
    return [getattr(mw.cls, "__name__", "") for mw in app.user_middleware]


def test_compress_min_bytes_defaults_to_the_nginx_floor(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    monkeypatch.delenv("BOR_COMPRESS_MIN_BYTES", raising=False)
    assert _settings().compress_min_bytes == 860


def test_compress_min_bytes_env_override(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("BOR_COMPRESS_MIN_BYTES", "2048")
    assert _settings().compress_min_bytes == 2048
    monkeypatch.setenv("BOR_COMPRESS_MIN_BYTES", "0")  # the kill switch is legal
    assert _settings().compress_min_bytes == 0


def test_compress_min_bytes_negative_refuses_startup_naming_the_field() -> None:
    with pytest.raises(ValidationError, match="compress_min_bytes must be >= 0"):
        _settings(compress_min_bytes=-1)


def test_create_app_wires_gzip_inside_security_headers(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """A20 holds: SecurityHeaders stays outermost (index 0 — starlette's
    last add_middleware is the outermost layer), gzip sits between it and
    the caching layer, so an already-rewritten HTML page compresses and
    the /api/* caching pass-through happens before compression.

    Phase 167 task 05 adds ``ResponseBudgetMiddleware`` between gzip and
    caching — INSIDE the compressor on purpose, so its ``bytes=`` counts the
    body the ROUTE produced while its ``enc=`` still reads the encoding the
    compressor put on the wire (see tests/unit/test_response_budget.py)."""
    monkeypatch.setattr(main_mod.settings, "compress_min_bytes", 860)
    app2 = main_mod.create_app()
    assert _middleware_names(app2) == [
        "SecurityHeadersMiddleware",
        "SelectiveGZipMiddleware",
        "ResponseBudgetMiddleware",
        "CachingMiddleware",
        "SessionMiddleware",
    ]
    # The app still serves.
    assert TestClient(app2).get("/api/health").status_code == 200


def test_create_app_adds_no_gzip_layer_when_disabled(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """BOR_COMPRESS_MIN_BYTES=0 removes the middleware entirely (the kill
    switch): the gzip class is absent from app.user_middleware."""
    monkeypatch.setattr(main_mod.settings, "compress_min_bytes", 0)
    app2 = main_mod.create_app()
    names = _middleware_names(app2)
    assert "SelectiveGZipMiddleware" not in names
    assert names == [
        "SecurityHeadersMiddleware",
        # Phase 167 task 05: the byte-budget line is not optional.
        "ResponseBudgetMiddleware",
        "CachingMiddleware",
        "SessionMiddleware",
    ]