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 (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
import threading
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, sysconfig
from app.config import settings
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,
        "session_minutes": 60,
    }


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,
        "session_minutes": 60,
    }


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_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_erased_accounts_batch_is_stored_without_the_account(elevated):
    """The sink's writer is not the only path into `log_events` — this route inserts too, and an
    account erased while its browser was mid-flush must not get its rows back through it. Faked at
    the `forget_user` seam the purge itself calls: once the User row is actually gone the session
    no longer resolves, so the reachable window is exactly a request already past `current_user`.
    (`logs._erased` is cleared before every test by conftest's `_reset_log_erasures`.)"""
    user, client, rows = elevated("DEBUG")
    logs.forget_user(user.id)
    r = client.post("/api/diagnostics/events", json={"events": [_event(level="WARNING")]})
    assert r.json() == {"ok": True, "stored": 1}
    (stored,) = rows()
    assert (stored.source, stored.user_id) == ("web", None)


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()
    assert (
        client.post(
            "/api/diagnostics/events", json={"events": [_event(ts=future), _event(ts=past)]}
        ).status_code
        == 200
    )
    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


def test_what_an_account_stores_per_hour_is_capped_and_said_once(elevated, monkeypatch):
    user, client, rows = elevated("DEBUG")
    monkeypatch.setattr(
        diagnostics_routes, "_stored_bytes", diagnostics_routes.RateLimiter(3_000, 3600)
    )
    batch = {"events": [_event(msg="m" * 1_000)]}
    stored = [client.post("/api/diagnostics/events", json=batch).json()["stored"] for _ in range(4)]
    assert stored == [1, 1, 0, 0]
    events = [row.event for row in rows()]
    # Two batches, then ONE notice for the hour, not one per refused batch.
    assert events == ["nav.view", "nav.view", "diagnostics.budget_spent"]
    assert rows()[-1].user_id == user.id


def test_only_a_json_body_is_read(elevated):
    _user, client, rows = elevated("DEBUG")
    # What a same-site `<form enctype="text/plain">` can send without a preflight.
    r = client.post(
        "/api/diagnostics/events",
        content=json.dumps({"events": [_event()]}),
        headers={"content-type": "text/plain"},
    )
    assert r.status_code == 415 and r.json()["detail"]["code"] == "json_required"
    assert rows() == []
    ok = client.post(
        "/api/diagnostics/events",
        content=json.dumps({"events": [_event()]}),
        headers={"content-type": "application/json; charset=utf-8"},
    )
    assert ok.json()["stored"] == 1


# --- 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):
    assert _app().get("/api/thing/41").status_code == 200
    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()
    assert client.get("/api/thing/7", params={"level": "DEBUG"}).status_code == 200
    assert client.get("/api/thing/8").status_code == 200
    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()
    assert client.get("/api/thing/7", params={"level": "TRACE"}).status_code == 200
    assert logging.getLogger().level == logs.TRACE
    assert client.get("/api/thing/7").status_code == 200  # 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()
    assert client.get("/api/thing/7", params={"level": "DEBUG", "q": "lofi"}).status_code == 200
    assert (
        client.get(
            "/api/thing/7", params={"level": "TRACE", "q": "lofi", "token": "SECRET"}
        ).status_code
        == 200
    )
    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_an_address_typed_into_a_search_is_redacted_though_the_browser_encodes_it(gated):
    assert (
        _app()
        .get("/api/thing/7?level=TRACE&q=someone%40example.com&new_password=hunter2")
        .status_code
        == 200
    )
    (line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
    assert line.attrs["query"] == "level=TRACE&q=[email]&new_password=[redacted]"


@pytest.mark.parametrize(
    "raw",
    ["state=a%20b", "code=%20SECRET", "sig=a%22b", "key=a%26b", "reset=a+b"],
    ids=["space", "leading-space", "quote", "ampersand", "plus"],
)
def test_a_credential_param_goes_whole_however_its_value_decodes(raw):
    """Decoding is what lets `_EMAIL_RE` see `%40`, but it also puts spaces and separators into a
    value that `_SECRET_PARAM_RE`'s value class stops at — so the name rule decides, not the text."""
    name = raw.split("=", 1)[0]
    assert logs.redact_query(f"{raw}&q=lofi") == f"{name}={logs.REDACTED}&q=lofi"


def test_an_admin_level_of_trace_does_not_log_other_users_query_strings(gated):
    logs.set_level("TRACE")
    assert _app().get("/api/thing/7", params={"q": "something private"}).status_code == 200
    (line,) = [r for r in gated.records if getattr(r, "event", None) == "http.request"]
    assert "query" not in line.attrs


def test_the_admin_level_applies_to_the_running_process_when_set_and_when_reset(
    gated, db, make_user, as_user
):
    admin = as_user(make_user(role="admin"))
    r = admin.patch("/api/admin/config/log_level", json={"value": "DEBUG"})
    assert r.status_code == 200, r.text
    assert logs.current_level() == "DEBUG"
    assert admin.delete("/api/admin/config/log_level").status_code == 200
    assert logs.current_level() == settings.log_level


def test_erasing_the_only_elevated_account_puts_the_process_level_back(gated):
    """The root level is the lowest of the base and every unexpired elevation. `forget_user` drops
    an elevation like any other path that drops one, so it owes the same recompute — or the process
    keeps building every TRACE record for an account that no longer exists, to throw them away."""
    until = datetime.now(timezone.utc) + timedelta(hours=1)
    try:
        logs.set_user(7, "TRACE", until)
        assert logging.getLogger().level == logs.TRACE
        logs.forget_user(7)
        assert logging.getLogger().level == logging.INFO
    finally:
        # `set_user` writes the request scope, and outside a request that scope is this process's
        # for good — a later test would store its INFO lines as an elevated session's.
        logs.set_user(None)


def test_the_stored_admin_level_is_applied_at_boot(gated, db, monkeypatch):
    import asyncio

    import app.main as app_main

    sysconfig.set_value(db, "log_level", "DEBUG")
    db.commit()
    monkeypatch.setattr(app_main.state, "is_configured", lambda db: True)
    monkeypatch.setattr(app_main, "start_scheduler", lambda: None)
    monkeypatch.setattr(app_main, "shutdown_scheduler", lambda: None)
    monkeypatch.setattr(app_main.plex_subtitles, "shutdown", lambda: None)

    async def _no_sweep():
        return None

    monkeypatch.setattr(app_main, "_sweep_hls_orphans", _no_sweep)
    levels: list[str] = []

    async def _cycle():
        async with app_main.lifespan(app_main.app):
            levels.append(logs.current_level())

    asyncio.run(_cycle())
    assert levels == ["DEBUG"]


def test_the_root_level_is_applied_in_the_order_the_elevations_changed(gated, monkeypatch):
    """Two requests change the elevations at once: A's user leaves (the root may go back to INFO),
    B's user arrives (it must go to DEBUG). B's change is the later one, so DEBUG must be what stays
    — even when A is paused between working out its level and applying it."""
    root = logging.getLogger()
    logs._track_elevation(1, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1))
    real_set = root.setLevel
    other: list[threading.Thread] = []

    def paused_set(level):
        if level == logging.INFO and not other:
            b = threading.Thread(
                target=logs._track_elevation,
                args=(2, logging.DEBUG, datetime.now(timezone.utc) + timedelta(hours=1)),
            )
            other.append(b)
            b.start()
            b.join(0.3)  # B runs to completion here unless A's level change holds it off
        real_set(level)

    monkeypatch.setattr(root, "setLevel", paused_set)
    logs._track_elevation(1, None, None)  # A
    other[0].join(5)
    assert root.level == logging.DEBUG and 2 in logs._elevations


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
    assert client.get("/api/me").status_code == 200
    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