"""The activity page's reads: the audit log and the background runs, filtered on the server and paged by a keyset, and the quota breakdown over a time range and one account. Each list is asked for the rows a filter must keep AND must drop, and a page boundary is crossed with rows that share a timestamp — the case an offset or a timestamp-only cursor gets wrong. """ from datetime import datetime, timedelta, timezone from app.models import AuditLog, Channel, JobRun, LogEvent, QuotaEvent T0 = datetime(2026, 10, 1, 12, 0, tzinfo=timezone.utc) def _audit(db, action, actor=None, at=T0, summary=None, request_id=None) -> int: row = AuditLog( actor_id=actor, action=action, summary=summary, created_at=at, request_id=request_id ) db.add(row) db.flush() return row.id def _run(db, job, status="ok", actor=None, at=T0, target=None, trigger="scheduled") -> int: row = JobRun( job=job, trigger=trigger, actor_id=actor, target=target, status=status, started_at=at, finished_at=None if status == "running" else at + timedelta(seconds=4), quota_units=0, ) db.add(row) db.flush() return row.id def _ids(body) -> list[int]: return [r["id"] for r in body["items"]] def test_the_audit_list_filters_on_the_server_and_pages_across_a_shared_timestamp( db, make_user, as_user ): admin = make_user(role="admin") a = _audit(db, "config_set", admin.id, T0, "quota_daily_budget = 9000") b = _audit(db, "config_set", admin.id, T0, "log_level = DEBUG") c = _audit(db, "config_set", admin.id, T0, "smtp_host = mail") bg = _audit(db, "user_purged", None, T0 - timedelta(days=3)) db.commit() client = as_user(admin) first = client.get("/api/admin/audit", params={"limit": 2, "action": "config_set"}).json() assert _ids(first) == [c, b] second = client.get( "/api/admin/audit", params={"limit": 2, "action": "config_set", "before": first["next_cursor"]}, ).json() assert (_ids(second), second["next_cursor"]) == ([a], None) # A page that ends exactly at the last row says there is nothing older. whole = client.get("/api/admin/audit", params={"limit": 3, "action": "config_set"}).json() assert (_ids(whole), whole["next_cursor"]) == ([c, b, a], None) assert first["accounts"] == {str(admin.id): admin.email} def ids(**params) -> list[int]: return _ids(client.get("/api/admin/audit", params=params).json()) assert ids(user="none") == [bg] assert ids(user=str(admin.id)) == [c, b, a] assert ids(q="DEBUG") == [b] assert ids(since=(T0 - timedelta(days=1)).isoformat()) == [c, b, a] assert ids(until=(T0 - timedelta(days=1)).isoformat()) == [bg] # The action menu counts every action under the other filters, its own pick included. facets = client.get("/api/admin/audit/facets", params={"action": "config_set"}).json() assert facets["actions"] == [ {"value": "config_set", "count": 3}, {"value": "user_purged", "count": 1}, ] def test_an_admin_action_keeps_the_request_it_was_made_in(db, make_user, as_user): client = as_user(make_user(role="admin")) r = client.patch("/api/admin/config/quota_daily_budget", json={"value": 9500}) assert r.status_code == 200, r.text (row,) = client.get("/api/admin/audit", params={"action": "config_set"}).json()["items"] assert row["request_id"] == r.headers["x-request-id"] def test_the_run_list_filters_pages_and_names_the_channel_and_the_log_lines(db, make_user, as_user): admin = make_user(role="admin") db.add(Channel(id="UCchan", title="Aileen Wang")) ok1 = _run(db, "rss_poll", at=T0) ok2 = _run(db, "rss_poll", at=T0) bad = _run(db, "backfill", "error", admin.id, T0, trigger="manual") adhoc = _run( db, "channel_backfill", "ok", admin.id, T0 - timedelta(hours=1), "UCchan", "ad_hoc" ) for _ in range(2): db.add( LogEvent(ts=T0, level="WARNING", sev=30, source="api", logger="t", msg="m", run_id=bad) ) db.commit() client = as_user(admin) first = client.get("/api/admin/runs", params={"limit": 2}).json() rest = client.get("/api/admin/runs", params={"limit": 2, "before": first["next_cursor"]}).json() assert _ids(first) + _ids(rest) == [bad, ok2, ok1, adhoc] assert rest["next_cursor"] is None rows = {r["id"]: r for r in first["items"] + rest["items"]} assert (rows[bad]["log_lines"], rows[ok2]["log_lines"]) == (2, 0) assert (rows[adhoc]["target_title"], rows[ok1]["target_title"]) == ("Aileen Wang", None) def ids(**params) -> list[int]: return _ids(client.get("/api/admin/runs", params=params).json()) assert ids(status="error") == [bad] assert ids(job="rss_poll") == [ok2, ok1] assert ids(user="none") == [ok2, ok1] assert ids(user=str(admin.id)) == [bad, adhoc] assert ids(q="UCchan") == [adhoc] assert ids(since=(T0 - timedelta(minutes=30)).isoformat()) == [bad, ok2, ok1] assert client.get("/api/admin/runs", params={"status": "finished"}).status_code == 422 facets = client.get("/api/admin/runs/facets", params={"status": "error"}).json() assert facets["statuses"] == [{"value": "ok", "count": 3}, {"value": "error", "count": 1}] assert facets["jobs"] == [{"value": "backfill", "count": 1}] assert facets["total"] == 4 def test_the_quota_breakdown_covers_its_range_and_one_account(db, make_user, as_user): admin = make_user(role="admin") now = datetime.now(timezone.utc) db.add_all( [ QuotaEvent(user_id=admin.id, action="search", units=100, created_at=now), QuotaEvent(user_id=None, action="backfill", units=7, created_at=now), QuotaEvent( user_id=admin.id, action="search", units=50, created_at=now - timedelta(days=10) ), ] ) db.commit() client = as_user(admin) def totals(**params) -> dict: body = client.get("/api/quota/admin", params=params).json() return {r["user_id"]: r["total"] for r in body["rows"]} assert totals() == {admin.id: 150, None: 7} # the last 30 days by default assert totals(since=(now - timedelta(days=1)).isoformat()) == {admin.id: 100, None: 7} assert totals(until=(now - timedelta(days=1)).isoformat()) == {admin.id: 50} assert totals(user="none") == {None: 7} assert totals(user=str(admin.id)) == {admin.id: 150}