siftlode/backend/tests/test_logs.py

"""E22 S2 — the one logging configuration, its redaction, and the request line that replaced
uvicorn's access log.

Redaction is the part worth guarding hardest: it is applied at the HANDLER, so these cases are the
only thing standing between a future call site and a token in a log we attach to a Gitea issue.
The request-line cases pin what uvicorn's access log got wrong — the query string (`?account=1`,
`?token=…`) was printed on every request, and nothing correlated two lines of the same request.

Pure logic + an in-process ASGI app: no database.
"""

import json
import logging

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app import logs
from app.config import LogLevel, choices_of


class _Sink(logging.Handler):
    """Captures records AND their formatted JSON, with the same filter/formatter the real stdout
    handler uses — so a test sees exactly the bytes a deployment would write."""

    def __init__(self):
        super().__init__(level=logging.NOTSET)
        self.lines: list[str] = []
        self.setFormatter(logs.JsonFormatter())
        self.addFilter(logs.ContextFilter())

    def emit(self, record):
        self.lines.append(self.format(record))

    def objects(self) -> list[dict]:
        return [json.loads(line) for line in self.lines]


@pytest.fixture
def sink():
    """Attached to `siftlode` (not root) so a `configure_logging` call inside a test — which
    replaces the ROOT handlers — can't silently detach it mid-test."""
    logger = logging.getLogger("siftlode")
    handler = _Sink()
    previous = logger.level
    logger.addHandler(handler)
    logger.setLevel(logs.TRACE)
    try:
        yield handler
    finally:
        logger.removeHandler(handler)
        logger.setLevel(previous)


# --- redaction ------------------------------------------------------------------------------


@pytest.mark.parametrize(
    "text, must_not_contain",
    [
        ("user signed in: someone@example.com", "someone@example.com"),
        ("callback https://app/auth/callback?code=4/0AY&state=xyz", "4/0AY"),
        ("mail link https://app/#verify=abc123def", "abc123def"),
        ("share https://app/s?token=SHARETOKEN", "SHARETOKEN"),
        ("reset https://app/#reset=RESETTOKEN", "RESETTOKEN"),
    ],
)
def test_free_text_secrets_never_reach_the_line(text, must_not_contain):
    assert must_not_contain not in logs.redact_text(text)


def test_a_key_that_names_a_secret_is_dropped_whole():
    # Not truncated to a prefix: a prefix is still a leak and is useless for triage anyway.
    out = logs.redact(
        {
            "password": "hunter2",
            "current_password": "hunter2",
            "plex_server_token": "abc",
            "Cookie": "session=xyz",
            "authorization": "Bearer abc",
            "nested": {"refresh_token": "rt", "keep": "visible"},
            "list": [{"api_key": "k"}],
            "count": 7,
        }
    )
    assert out["password"] == logs.REDACTED
    assert out["current_password"] == logs.REDACTED
    assert out["plex_server_token"] == logs.REDACTED
    assert out["Cookie"] == logs.REDACTED
    assert out["authorization"] == logs.REDACTED
    assert out["nested"]["refresh_token"] == logs.REDACTED
    assert out["nested"]["keep"] == "visible"  # only the named keys go
    assert out["list"][0]["api_key"] == logs.REDACTED
    assert out["count"] == 7


def test_an_email_in_an_attribute_value_is_scrubbed_too():
    assert logs.redact({"to": "a.b+tag@mail.example.org"})["to"] == logs.REDACTED_EMAIL


def test_redaction_runs_on_the_message_the_attrs_and_the_traceback(sink):
    log = logging.getLogger("siftlode.test")
    try:
        raise RuntimeError("boom for someone@example.com")
    except RuntimeError:
        log.exception(
            "failed for %s",
            "someone@example.com",
            extra=logs.event("test.failed", password="hunter2", url="https://x/?token=T0K3N"),
        )
    line = sink.lines[-1]
    assert "someone@example.com" not in line
    assert "hunter2" not in line
    assert "T0K3N" not in line
    payload = json.loads(line)
    assert payload["attrs"]["password"] == logs.REDACTED
    assert "RuntimeError" in payload["exc"]  # the diagnosis survives the scrubbing


# --- event shape ----------------------------------------------------------------------------


def test_every_line_carries_the_agreed_fields(sink):
    logging.getLogger("siftlode.test").info("hello", extra=logs.event("test.hello", n=1))
    payload = sink.objects()[-1]
    for field in ("ts", "level", "sev", "source", "logger", "msg", "event", "version"):
        assert field in payload, field
    assert payload["level"] == "INFO"
    assert payload["sev"] == 9
    assert payload["event"] == "test.hello"
    assert payload["attrs"] == {"n": 1}


def test_a_plain_log_call_still_produces_a_well_formed_line(sink):
    """An event code is optional — 125 existing call sites have none yet (E22.5 adds them), and they
    must not be worse off than before this module."""
    logging.getLogger("siftlode.test").warning("just a message")
    payload = sink.objects()[-1]
    assert payload["msg"] == "just a message"
    assert payload["level"] == "WARNING"
    assert "event" not in payload


def test_a_broken_format_call_reports_itself_instead_of_raising():
    """A record is formatted directly rather than logged: a genuinely broken call ALSO breaks
    pytest's own capture handler (stdlib formatting raises there too), which would fail the test for
    the wrong reason. What is under test is our formatter's refusal to raise."""
    record = logging.LogRecord(
        "siftlode.test", logging.INFO, __file__, 0, "two args %s %s", ("only-one",), None
    )
    payload = json.loads(logs.JsonFormatter().format(record))
    assert "log formatting failed" in payload["msg"]


def test_bound_correlation_appears_and_is_released(sink):
    log = logging.getLogger("siftlode.test")
    with logs.bind(job_id=42):
        log.info("inside")
    log.info("outside")
    assert sink.objects()[-2]["job_id"] == 42
    assert "job_id" not in sink.objects()[-1]


def test_every_selectable_level_resolves_to_a_real_logging_level():
    # The picker's options come from `LogLevel`; each must be a level `logging` knows, TRACE included.
    assert [logs.resolve_level(n) for n in choices_of(LogLevel)] == [
        logs.TRACE,
        logging.DEBUG,
        logging.INFO,
        logging.WARNING,
        logging.ERROR,
    ]
    with pytest.raises(ValueError):
        logs.resolve_level("nonsense")


def test_configure_logging_silences_the_uvicorn_access_log_and_keeps_ours_propagating():
    logs.configure_logging("api")
    assert logging.getLogger("uvicorn.access").getEffectiveLevel() >= logging.CRITICAL
    # propagate stays on so every line reaches the ONE root handler — and so pytest's caplog works.
    assert logging.getLogger("siftlode.anything").propagate is True
    assert len(logging.getLogger().handlers) == 1


# --- the request line -----------------------------------------------------------------------


def _client() -> TestClient:
    app = FastAPI()
    app.middleware("http")(logs.log_requests)

    @app.get("/api/videos/{video_id}")
    def _video(video_id: str, q: str | None = None):
        return {"ok": video_id}

    @app.get("/healthz")
    def _health():
        return {"ok": True}

    @app.get("/boom")
    def _boom():
        raise RuntimeError("kaboom")

    return TestClient(app, raise_server_exceptions=False)


def test_the_request_line_has_the_route_template_and_no_query_string(sink):
    logging.getLogger("siftlode.http").setLevel(logs.TRACE)
    r = _client().get("/api/videos/abc123?q=cats&token=SECRET")
    assert r.status_code == 200
    payload = sink.objects()[-1]
    assert payload["event"] == "http.request"
    # The TEMPLATE, so every video request groups into one queryable path...
    assert payload["attrs"]["path"] == "/api/videos/{video_id}"
    # ...and the query string — uvicorn's access log printed it, `?account=1` and all — is gone.
    assert "SECRET" not in sink.lines[-1]
    assert "cats" not in sink.lines[-1]
    assert payload["attrs"]["status"] == 200
    assert isinstance(payload["attrs"]["duration_ms"], float)


def test_the_request_id_is_echoed_and_shared_by_every_line_of_that_request(sink):
    client = _client()
    r = client.get("/api/videos/abc123")
    rid = r.headers["X-Request-Id"]
    assert rid and sink.objects()[-1]["request_id"] == rid


def test_a_client_supplied_id_is_reused_only_when_it_is_well_formed(sink):
    client = _client()
    good = client.get("/api/videos/x", headers={"X-Request-Id": "web-0123456789ab"})
    assert good.headers["X-Request-Id"] == "web-0123456789ab"
    # An unbounded header would be a way to write arbitrary text into the log; short/odd ones are
    # replaced with our own id rather than trusted.
    for bogus in ("short", "has spaces and \n newline", "x" * 200):
        r = client.get("/api/videos/x", headers={"X-Request-Id": bogus})
        assert r.headers["X-Request-Id"] != bogus


def test_the_healthcheck_is_logged_at_debug_so_it_cannot_drown_the_log(sink):
    http = logging.getLogger("siftlode.http")
    http.setLevel(logs.TRACE)
    _client().get("/healthz")
    assert sink.objects()[-1]["level"] == "DEBUG"
    # ...but it is still logged when something is wrong with it: the floor drops, nothing is hidden.
    http.setLevel(logging.INFO)
    before = len(sink.lines)
    _client().get("/healthz")
    assert len(sink.lines) == before


def test_a_failing_request_is_logged_by_status_and_an_unhandled_error_is_logged_at_all(sink):
    client = _client()
    client.get("/api/videos/missing/deeper")  # 404 from the router
    assert sink.objects()[-1]["level"] == "WARNING"
    client.get("/boom")
    payload = sink.objects()[-1]
    # The request that most needs a line is the one that raised: uvicorn's access log wrote it, but
    # our middleware sees the exception instead of a response, so it has to log it explicitly.
    assert payload["level"] == "ERROR"
    assert payload["event"] == "http.request"
    assert "RuntimeError" in payload["exc"]