siftlode/backend/tests/test_diagnostics.py

"""E22 S5 — a user's consented diagnostics level, and the browser log it lets in.

Three parts:
  * the level itself: the server computes its expiry, a lapsed one reads as the base, the demo
    account cannot raise it (DB lane, through the real routes);
  * `POST /api/diagnostics/events`: refused at the base level, and nothing in a batch is trusted —
    the account comes from the session, everything is redacted again, events under the consented
    level are dropped, sizes and pace are capped (DB lane);
  * the server side of the elevation: ONE user's requests are logged in detail while everyone
    else's stay at the base level, and the account reaches a handler's own lines at all (pure, an
    in-process app).
"""

import json
import logging
from datetime import datetime, timedelta, timezone

import pytest
from fastapi import Depends, FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import select

from app import diagnostics, logs, logsink
from app.models import LogEvent, User
from app.routes import diagnostics as diagnostics_routes

_PW = "correct horse battery"


# --- the level ------------------------------------------------------------------------------


def test_a_raised_level_counts_until_it_lapses_and_not_a_moment_longer():
    now = datetime.now(timezone.utc)
    assert diagnostics.effective_level(User()) == "INFO"
    raised = User(diagnostics_level="TRACE", diagnostics_until=now + timedelta(minutes=1))
    assert diagnostics.effective_level(raised, now) == "TRACE"
    assert diagnostics.effective_level(raised, now + timedelta(minutes=2)) == "INFO"
    # A level without an expiry is not a level that never ends — it is no level at all.
    assert diagnostics.effective_level(User(diagnostics_level="DEBUG")) == "INFO"


def _signed_in(client, make_user, **kw):
    """A real sign-in: `/api/me` resolves through `optional_current_user`, which the `as_user`
    override does not cover (see test_me_email_flag)."""
    user = make_user(email="diag@example.com", password=_PW, **kw)
    r = client.post("/auth/password-login", json={"email": "diag@example.com", "password": _PW})
    assert r.status_code == 200, r.text
    return user


def test_raising_the_level_sets_the_servers_expiry_whatever_the_client_sends(db, client, make_user):
    user = _signed_in(client, make_user)
    r = client.put("/api/me/diagnostics", json={"level": "DEBUG", "until": "2099-01-01T00:00:00Z"})
    assert r.status_code == 200, r.text
    db.refresh(user)
    expected = datetime.now(timezone.utc) + timedelta(hours=diagnostics.ELEVATION_HOURS)
    assert abs((user.diagnostics_until - expected).total_seconds()) < 60
    assert r.json()["diagnostics"]["level"] == "DEBUG"
    assert client.get("/api/me").json()["diagnostics"]["level"] == "DEBUG"

    assert client.put("/api/me/diagnostics", json={"level": "INFO"}).status_code == 200
    db.refresh(user)
    assert (user.diagnostics_level, user.diagnostics_until) == (None, None)
    assert client.get("/api/me").json()["diagnostics"] == {
        "level": "INFO",
        "until": None,
        "retention_days": 7,
    }


def test_a_lapsed_level_reads_as_the_base_in_me(db, client, make_user):
    _signed_in(
        client,
        make_user,
        diagnostics_level="TRACE",
        diagnostics_until=datetime.now(timezone.utc) - timedelta(minutes=1),
    )
    assert client.get("/api/me").json()["diagnostics"] == {
        "level": "INFO",
        "until": None,
        "retention_days": 7,
    }


def test_only_the_three_levels_can_be_chosen(db, make_user, as_user):
    client = as_user(make_user())
    for bogus in ("WARNING", "debug", "EVERYTHING"):
        assert client.put("/api/me/diagnostics", json={"level": bogus}).status_code == 422


def test_the_demo_account_can_return_to_the_base_but_never_raise_it(db, make_user, as_user):
    client = as_user(make_user(is_demo=True))
    r = client.put("/api/me/diagnostics", json={"level": "TRACE"})
    assert r.status_code == 403 and r.json()["detail"]["code"] == "demo"
    assert client.put("/api/me/diagnostics", json={"level": "INFO"}).status_code == 200


def test_preferences_must_be_a_json_object(db, make_user, as_user):
    client = as_user(make_user())
    assert client.put("/api/me/preferences", json=["not", "an", "object"]).status_code == 422
    r = client.put("/api/me/preferences", json={"theme": "dark"})
    assert r.status_code == 200 and r.json()["preferences"]["theme"] == "dark"


# --- the browser's batches ------------------------------------------------------------------


def _event(**over) -> dict:
    return {
        "ts": datetime.now(timezone.utc).isoformat(),
        "level": "INFO",
        "event": "nav.view",
        "msg": "view",
        "tab_id": "a1b2c3",
        "version": "0.88.0",
        **over,
    }


@pytest.fixture
def elevated(db, make_user, as_user):
    """A signed-in client whose user consented to `level` a minute ago, and a reader for its rows."""
    diagnostics_routes._batches.reset()

    def _make(level: str = "DEBUG"):
        user = make_user(
            diagnostics_level=level,
            diagnostics_until=datetime.now(timezone.utc) + timedelta(hours=23),
        )

        def rows() -> list[LogEvent]:
            db.expire_all()
            return list(db.execute(select(LogEvent).order_by(LogEvent.id)).scalars())

        return user, as_user(user), rows

    yield _make
    diagnostics_routes._batches.reset()


def test_nothing_is_accepted_at_the_base_level(db, make_user, as_user):
    client = as_user(make_user())
    r = client.post("/api/diagnostics/events", json={"events": [_event()]})
    assert r.status_code == 409 and r.json()["detail"]["code"] == "diagnostics_off"
    assert not db.execute(select(LogEvent)).first()


def test_an_anonymous_caller_is_refused(client):
    assert client.post("/api/diagnostics/events", json={"events": [_event()]}).status_code == 401


def test_a_batch_is_stored_as_the_sessions_web_events_and_redacted_again(elevated):
    user, client, rows = elevated("DEBUG")
    r = client.post(
        "/api/diagnostics/events",
        json={
            "events": [
                _event(
                    level="WARNING",
                    event="ui.notify",
                    msg="sent to someone@example.com",
                    attrs={"password": "hunter2", "url": "/?token=abc123", "user_id": 999},
                    request_id="0123456789abcdef",
                ),
                _event(level="DEBUG", event="ui.click", attrs={"control": "feed.refresh"}),
            ],
            "user_id": 999,  # ignored: the account is the session's
        },
    )
    assert r.json() == {"ok": True, "stored": 2}
    warning, click = rows()
    assert (warning.source, warning.logger, warning.user_id) == ("web", "web", user.id)
    assert warning.msg == "sent to [email]"
    assert warning.attrs["password"] == logs.REDACTED
    assert "abc123" not in json.dumps(warning.attrs)
    assert warning.attrs["tab_id"] == "a1b2c3"
    assert warning.request_id == "0123456789abcdef"
    assert (warning.sev, click.level, click.version) == (13, "DEBUG", "0.88.0")


def test_an_event_under_the_consented_level_is_dropped(elevated):
    _user, client, rows = elevated("DEBUG")
    r = client.post(
        "/api/diagnostics/events",
        json={"events": [_event(level="TRACE", msg="typed: lofi"), _event(level="DEBUG")]},
    )
    assert r.json()["stored"] == 1
    assert [row.level for row in rows()] == ["DEBUG"]


def test_a_timestamp_outside_the_consent_window_is_clamped(elevated):
    _user, client, rows = elevated("DEBUG")
    now = datetime.now(timezone.utc)
    future = (now + timedelta(days=365)).isoformat()
    past = (now - timedelta(days=30)).isoformat()
    client.post("/api/diagnostics/events", json={"events": [_event(ts=future), _event(ts=past)]})
    late, early = rows()
    assert late.ts <= datetime.now(timezone.utc)
    assert early.ts >= now - timedelta(hours=diagnostics.ELEVATION_HOURS, minutes=1)


def test_sizes_are_capped(elevated):
    _user, client, rows = elevated("DEBUG")
    too_many = [_event() for _ in range(diagnostics_routes.EVENTS_MAX + 1)]
    assert client.post("/api/diagnostics/events", json={"events": too_many}).status_code == 422

    huge = "x" * (diagnostics_routes.BODY_MAX_BYTES + 1)
    r = client.post(
        "/api/diagnostics/events",
        content=json.dumps({"events": [_event(attrs={"blob": huge})]}),
        headers={"content-type": "application/json"},
    )
    assert r.status_code == 413

    big_attrs = {"blob": "y" * (diagnostics_routes.ATTRS_MAX_BYTES + 10)}
    assert (
        client.post("/api/diagnostics/events", json={"events": [_event(attrs=big_attrs)]}).json()[
            "stored"
        ]
        == 1
    )
    (row,) = rows()
    assert "blob" not in row.attrs and row.attrs["attrs_dropped_bytes"] > 0

    bad_code = client.post("/api/diagnostics/events", json={"events": [_event(event="Nav View")]})
    assert bad_code.status_code == 422


def test_the_pace_is_capped_per_user(elevated):
    _user, client, _rows = elevated("DEBUG")
    codes = [
        client.post("/api/diagnostics/events", json={"events": []}).status_code for _ in range(31)
    ]
    assert codes[:30] == [200] * 30 and codes[30] == 429


# --- the server side of an elevation ---------------------------------------------------------


class _Capture(logging.Handler):
    """The stdout handler's filters (context + gate), capturing records."""

    def __init__(self):
        super().__init__(level=logging.NOTSET)
        self.records: list[logging.LogRecord] = []
        self.addFilter(logs.ContextFilter())
        self.addFilter(logs.LevelGate())

    def emit(self, record):
        self.records.append(record)


@pytest.fixture
def gated():
    """A process configured at INFO with a gated handler on the ROOT (where the real one sits), and
    every piece of module state the elevation touches restored afterwards."""
    root = logging.getLogger()
    saved = (logs._base_level, dict(logs._elevations), logs._root_applied, root.level)
    handler = _Capture()
    root.addHandler(handler)
    logs._elevations.clear()
    logs.set_level("INFO")
    try:
        yield handler
    finally:
        root.removeHandler(handler)
        logs._base_level, logs._root_applied = saved[0], saved[2]
        logs._elevations.clear()
        logs._elevations.update(saved[1])
        root.setLevel(saved[3])


def _app() -> TestClient:
    """What a real request does: the request middleware, then a SYNC auth dependency that resolves
    the account (run by FastAPI in a threadpool, on a copy of the context), then a handler."""
    app = FastAPI()
    app.middleware("http")(logs.log_requests)
    handler_log = logging.getLogger("siftlode.difftest")

    def account(user_id: int, level: str = "INFO"):
        until = datetime.now(timezone.utc) + timedelta(hours=1)
        logs.set_user(user_id, level, until)

    @app.get("/api/thing/{user_id}")
    def _thing(user_id: int, _=Depends(account)):
        handler_log.debug("detail only an elevated request keeps")
        handler_log.info("an ordinary line")
        return {}

    return TestClient(app)


def test_the_account_reaches_the_handlers_own_lines(gated):
    _app().get("/api/thing/41")
    line = next(r for r in gated.records if r.getMessage() == "an ordinary line")
    assert getattr(line, "user_id", None) == 41


def test_one_users_elevation_logs_their_requests_in_detail_and_nobody_elses(gated):
    client = _app()
    client.get("/api/thing/7", params={"level": "DEBUG"})
    client.get("/api/thing/8")
    detail = [r for r in gated.records if r.getMessage().startswith("detail only")]
    assert [r.user_id for r in detail] == [7]
    assert getattr(detail[0], "elevated", False) is True
    # Every line of the elevated request is marked for storage — the sink keeps it at any level.
    ordinary = [r for r in gated.records if r.getMessage() == "an ordinary line"]
    assert [logsink.should_store(r) for r in ordinary] == [True, False]
    # The process's configured level is still what it reports, however low the root went.
    assert logs.current_level() == "INFO" and logging.getLogger().level == logging.DEBUG


def test_the_root_level_returns_to_the_base_when_the_elevation_ends(gated):
    client = _app()
    client.get("/api/thing/7", params={"level": "TRACE"})
    assert logging.getLogger().level == logs.TRACE
    client.get("/api/thing/7")  # the same user, back at the base
    assert logging.getLogger().level == logging.INFO
    logs._track_elevation(9, logging.DEBUG, datetime.now(timezone.utc) - timedelta(seconds=1))
    assert logging.getLogger().level == logging.INFO  # an already-expired one is pruned at once


def test_the_query_string_is_logged_only_for_a_traced_request_and_redacted(gated):
    client = _app()
    client.get("/api/thing/7", params={"level": "DEBUG", "q": "lofi"})
    client.get("/api/thing/7", params={"level": "TRACE", "q": "lofi", "token": "SECRET"})
    lines = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
    assert "query" not in lines[0].attrs
    assert "q=lofi" in lines[1].attrs["query"] and "SECRET" not in lines[1].attrs["query"]


def test_a_real_sign_in_carries_the_consented_level_into_the_process(gated, client, db, make_user):
    user = _signed_in(client, make_user)
    assert client.put("/api/me/diagnostics", json={"level": "DEBUG"}).status_code == 200
    logs._elevations.clear()  # only a request resolving the account through auth may put it back
    client.get("/api/me")
    assert logs._elevations[user.id][0] == logging.DEBUG
    assert client.put("/api/me/diagnostics", json={"level": "INFO"}).status_code == 200
    assert user.id not in logs._elevations