siftlode/backend/tests/test_admin_logs.py

"""The admin diagnostics console's query API (E22.6 S1), DB-lane."""

from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import insert

from app.models import LogEvent

T0 = datetime(2026, 9, 19, 12, 0, tzinfo=timezone.utc)


def _row(**kw) -> dict:
    row = {
        "ts": T0,
        "level": "INFO",
        "sev": 9,
        "source": "api",
        "logger": "siftlode.http",
        "event": "http.request",
        "msg": "GET /api/feed 200",
        "attrs": None,
        "exc": None,
        "request_id": None,
        "user_id": None,
        "job_id": None,
        "version": "0.90.0",
    }
    row.update(kw)
    return row


def _add(db, *rows: dict) -> None:
    db.execute(insert(LogEvent), list(rows))
    db.commit()


@pytest.fixture
def admin(db, make_user, as_user):
    return as_user(make_user(email="admin@example.com", role="admin"))


def _msgs(resp) -> list[str]:
    assert resp.status_code == 200, resp.text
    return [it["msg"] for it in resp.json()["items"]]


def test_only_an_admin_reads_the_log(db, make_user, as_user, client):
    _add(db, _row())
    assert as_user(make_user()).get("/api/admin/logs").status_code == 403
    assert as_user(make_user()).get("/api/admin/logs/facets").status_code == 403
    assert client.get("/api/admin/logs").status_code == 401


def test_newest_first_and_every_row_once_across_pages_even_on_equal_timestamps(db, admin):
    # A browser row's `ts` is clamped, so several rows share one: the id breaks the tie.
    _add(db, *[_row(msg=f"same {i}") for i in range(5)], _row(ts=T0 - timedelta(1), msg="older"))

    seen, cursor = [], None
    while True:
        resp = admin.get(
            "/api/admin/logs", params={"limit": 2, "before": cursor} if cursor else {"limit": 2}
        )
        seen += _msgs(resp)
        cursor = resp.json()["next_cursor"]
        if cursor is None:
            break

    assert seen == ["same 4", "same 3", "same 2", "same 1", "same 0", "older"]


@pytest.mark.parametrize(
    ("params", "expected"),
    [
        ({"levels": ["ERROR", "WARNING"]}, {"err", "warn"}),
        ({"sources": ["worker"]}, {"err"}),
        ({"user": "1"}, {"mine"}),
        ({"user": "none"}, {"err", "warn", "plain"}),
        ({"event": "download.job.failed"}, {"err"}),
        ({"event": "none"}, {"warn"}),
        ({"request_id": "req-abcdef12"}, {"mine"}),
        ({"job_id": 42}, {"err"}),
        ({"q": "FORBIDDEN"}, {"err"}),  # in attrs, any case
        ({"q": "KeyError"}, {"warn"}),  # in the stack
        ({"q": "100%"}, {"plain"}),  # a literal, not a wildcard
        ({"since": (T0 - timedelta(hours=1)).isoformat()}, {"mine", "err", "warn"}),
        ({"until": (T0 - timedelta(hours=1)).isoformat()}, {"plain", "decoy"}),
        ({"levels": ["INFO"], "user": "none"}, {"plain"}),
    ],
)
def test_each_filter_selects_exactly_its_rows(db, admin, make_user, params, expected):
    # user 1 is the admin the fixture made
    _add(
        db,
        _row(msg="mine", user_id=1, request_id="req-abcdef12"),
        _row(
            msg="err",
            level="ERROR",
            sev=17,
            source="worker",
            event="download.job.failed",
            job_id=42,
            attrs={"code": "forbidden"},
        ),
        _row(msg="warn", level="WARNING", sev=13, event=None, exc="Traceback …\nKeyError: 'x'"),
        _row(msg="plain", ts=T0 - timedelta(hours=2), attrs={"note": "100% done"}),
        # Matched by no filter above but `until`: another account, and `100%` read as a pattern.
        _row(msg="decoy", ts=T0 - timedelta(hours=3), attrs={"note": "1000 done"}, user_id=99),
    )
    assert set(_msgs(admin.get("/api/admin/logs", params=params))) == expected


def test_a_page_names_its_accounts_and_leaves_an_erased_one_bare(db, admin):
    _add(db, _row(user_id=1), _row(user_id=77))

    assert admin.get("/api/admin/logs").json()["accounts"] == {"1": "admin@example.com"}


@pytest.mark.parametrize(
    "params",
    [
        {"before": "not-a-cursor"},
        {"before": "WyJ4IiwgMV0="},  # ["x", 1]: decodes, not a timestamp
    ],
)
def test_a_bad_cursor_is_a_400_with_a_code(db, admin, params):
    resp = admin.get("/api/admin/logs", params=params)
    assert resp.status_code == 400
    assert resp.json()["detail"]["code"] == "bad_cursor"


def test_facets_count_the_window_and_name_the_rows_without_a_value(db, admin):
    _add(
        db,
        _row(user_id=1),
        _row(user_id=1, level="ERROR", sev=17, event="sync.rss.failed"),
        _row(event=None),
        _row(ts=T0 - timedelta(days=2), source="worker"),
    )

    body = admin.get(
        "/api/admin/logs/facets", params={"since": (T0 - timedelta(days=1)).isoformat()}
    ).json()

    assert body["total"] == 3
    assert body["levels"] == [{"value": "INFO", "count": 2}, {"value": "ERROR", "count": 1}]
    assert body["sources"] == [{"value": "api", "count": 3}]
    assert body["events"] == [
        {"value": "http.request", "count": 1},
        {"value": "sync.rss.failed", "count": 1},
        {"value": "none", "count": 1},
    ]
    assert body["users"] == [{"value": "1", "count": 2}, {"value": "none", "count": 1}]
    assert body["accounts"] == {"1": "admin@example.com"}