brain-of-reese/tests/integration/test_response_budget_api.py

"""Integration: the byte-budget log line on the REAL app stack (phase 167, task 05).

``tests/unit/test_response_budget.py`` pins the middleware on a bare ASGI
app; this file pins what only the real stack can prove —

* the line describes the REAL ``GET /api/docs/tree``: ``bytes=`` equals the
  JSON the route produced, ``enc=identity`` for an identity client and
  ``enc=gzip`` through the actual ``SelectiveGZipMiddleware`` (the
  reading that works because the middleware sits INSIDE the compressor),
* a conditional revalidation of that route logs ``status=304 bytes=0`` —
  the phase-167 win visible in the log, and
* the level follows ``log_large_response_bytes`` (DEBUG under it, INFO at
  or over it, and ``/api/health`` under the 64 KB default), while
  non-``/api/`` paths (the shell page through the caching + compression
  layers) write nothing.

Uses the real compose Postgres (``db`` fixture) and FastAPI's TestClient.
"""
from __future__ import annotations

import itertools
import logging
import re
from datetime import UTC, datetime
from typing import Any

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import text

import app.core.response_budget as budget_mod
from app.config import Settings
from app.main import app as fastapi_app
from app.models import Chunk, Document

_LINE_RE = re.compile(
    r"^api method=(?P<method>[A-Z]+) path=(?P<path>\S+) status=(?P<status>\d+) "
    r"bytes=(?P<bytes>\d+) enc=(?P<enc>\S+) duration_ms=(?P<ms>\d+)$"
)

_counter = itertools.count()


def _clean(db) -> None:
    db.execute(text("TRUNCATE chunks, documents"))
    db.commit()


def _seed_corpus(db, count: int = 50) -> None:
    """``count`` indexed docs — the tree response lands well above the
    860-byte gzip floor (the corpus that motivates the phase is ~1 k docs)."""
    _clean(db)
    base = datetime.now(UTC)
    for i in range(count):
        doc = Document(
            source="budget-corpus",
            path=f"docs/sub-{i % 5}/file-{i:02d}.md",
            full_path=f"/tmp/budget-corpus/docs/sub-{i % 5}/file-{i:02d}.md",
            title=f"Doc {i:02d}",
            content=f"# Doc {i:02d}\n\nBody of document {i:02d}.",
            content_hash=f"{next(_counter):064d}",
            indexed_at=base,
        )
        db.add(doc)
        db.flush()
        db.add(
            Chunk(
                document_id=doc.id,
                position=0,
                content=f"chunk of doc {i:02d}",
                embedding=[0.01] * 768,
            )
        )
    db.commit()  # the endpoint session reads committed data only


def _budget_lines(
    caplog: pytest.LogCaptureFixture, path: str | None = None
) -> list[tuple[int, str]]:
    """``(levelno, message)`` for the budget logger, optionally narrowed to
    one request path (the ``admin_client`` fixture's own login is a real
    ``/api/*`` response too)."""
    lines = [
        (record.levelno, record.getMessage())
        for record in caplog.records
        if record.name == "app.response_budget"
    ]
    if path is None:
        return lines
    return [(level, message) for level, message in lines if f" path={path} " in message]


def test_tree_response_logs_the_exact_line_for_the_route_bytes(
    admin_client: TestClient, db, caplog: pytest.LogCaptureFixture
) -> None:
    """``bytes=`` is what the route produced — with an identity client that
    is exactly the body the caller received."""
    _seed_corpus(db)
    try:
        with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
            response = admin_client.get(
                "/api/docs/tree?depth=8", headers={"Accept-Encoding": "identity"}
            )
        assert response.status_code == 200

        lines = _budget_lines(caplog, "/api/docs/tree")
        assert len(lines) == 1
        match = _LINE_RE.match(lines[0][1])
        assert match is not None, lines[0][1]
        assert match.group("method") == "GET"
        assert match.group("status") == "200"
        assert int(match.group("bytes")) == len(response.content)
        assert match.group("enc") == "identity"
        assert int(match.group("ms")) >= 0
    finally:
        _clean(db)


def test_gzipped_tree_logs_enc_gzip_with_the_uncompressed_bytes(
    admin_client: TestClient, db, caplog: pytest.LogCaptureFixture
) -> None:
    """The D5 reading, end to end: the client received gzip and the log says
    ``enc=gzip`` while ``bytes=`` stays the ROUTE's size — which is why the
    middleware is wired inside the compression layer."""
    _seed_corpus(db)
    try:
        with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
            identity = admin_client.get(
                "/api/docs/tree?depth=8", headers={"Accept-Encoding": "identity"}
            )
            response = admin_client.get(
                "/api/docs/tree?depth=8", headers={"Accept-Encoding": "gzip"}
            )
        assert response.status_code == 200
        assert response.headers["content-encoding"] == "gzip"
        assert response.content == identity.content

        lines = _budget_lines(caplog, "/api/docs/tree")
        assert len(lines) == 2
        identity_line, gzip_line = (
            _LINE_RE.match(lines[0][1]),
            _LINE_RE.match(lines[1][1]),
        )
        assert identity_line is not None and identity_line.group("enc") == "identity"
        assert gzip_line is not None and gzip_line.group("enc") == "gzip"
        # Same route bytes in both lines…
        assert identity_line.group("bytes") == gzip_line.group("bytes")
        # …and that is the ROUTE's size, not the compressed transfer.
        assert int(gzip_line.group("bytes")) == len(identity.content)
    finally:
        _clean(db)


def test_conditional_revalidation_logs_zero_bytes(
    admin_client: TestClient, db, caplog: pytest.LogCaptureFixture
) -> None:
    """The phase-167 win is visible, not hidden: the 304 costs one line with
    ``bytes=0``."""
    _seed_corpus(db)
    try:
        fresh = admin_client.get("/api/docs/tree?depth=1")
        assert fresh.status_code == 200
        etag = fresh.headers["etag"]

        with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
            revalidated = admin_client.get(
                "/api/docs/tree?depth=1", headers={"If-None-Match": etag}
            )
        assert revalidated.status_code == 304

        lines = _budget_lines(caplog, "/api/docs/tree")
        assert len(lines) == 1
        match = _LINE_RE.match(lines[0][1])
        assert match is not None
        assert match.group("status") == "304"
        assert match.group("bytes") == "0"
    finally:
        _clean(db)


def test_level_follows_the_configured_budget(
    admin_client: TestClient,
    monkeypatch: pytest.MonkeyPatch,
    caplog: pytest.LogCaptureFixture,
) -> None:
    """``/api/health`` sits under the 65536 default → DEBUG; the same
    response with ``log_large_response_bytes=0`` is INFO (the loud mode). The
    middleware reads the setting per response, so this is the operator's
    runtime knob, not a boot-time one."""
    with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
        assert admin_client.get("/api/health").status_code == 200
    assert [level for level, _ in _budget_lines(caplog, "/api/health")] == [logging.DEBUG]

    monkeypatch.setattr(
        budget_mod,
        "get_settings",
        lambda: Settings(_env_file=None, log_large_response_bytes=0),  # pyright: ignore[reportCallIssue]
    )
    caplog.clear()
    with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
        assert admin_client.get("/api/health").status_code == 200
    assert [level for level, _ in _budget_lines(caplog, "/api/health")] == [logging.INFO]


def test_html_page_writes_no_budget_line(
    admin_client: TestClient, caplog: pytest.LogCaptureFixture
) -> None:
    """The shell page rides the caching rewrite AND the compression layer
    and still writes nothing — the budget measures ``/api/*`` only."""
    with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
        response = admin_client.get("/sources.html", headers={"Accept-Encoding": "gzip"})
        assert response.status_code == 200
        assert response.headers["content-encoding"] == "gzip"

    assert not any(" path=/sources.html " in message for _, message in _budget_lines(caplog))


def test_anonymous_403_is_measured_like_any_other_response(
    client: TestClient, caplog: pytest.LogCaptureFixture
) -> None:
    """The gate is a response too — an unauthenticated catalogue read logs
    its status (and the byte line is what makes a chatty 403 visible)."""
    with caplog.at_level(logging.DEBUG, logger="app.response_budget"):
        assert client.get("/api/docs").status_code == 403

    lines = _budget_lines(caplog, "/api/docs")
    assert len(lines) == 1
    match: Any = _LINE_RE.match(lines[0][1])
    assert match is not None
    assert match.group("status") == "403"


def test_the_module_level_app_carries_the_middleware() -> None:
    """Smoke: the wiring pin in ``tests/unit/test_response_budget.py`` covers
    the ORDER on a fresh ``create_app()``; this proves the module-level ``app``
    the whole suite (and the dev server) runs against has the layer too."""
    names = [getattr(mw.cls, "__name__", "") for mw in fastapi_app.user_middleware]
    assert "ResponseBudgetMiddleware" in names