FabledCurator/tests/test_api_system_activity.py

"""FC-3i: /api/system/activity/* endpoint tests.

Mocks Redis LLEN + celery inspect via monkeypatch on the module-level
_read_queues_sync / _read_workers_sync helpers, so tests don't need
a live broker or running workers.
"""

from datetime import UTC, datetime, timedelta

import pytest
import pytest_asyncio

from backend.app.api import system_activity as activity_module
from backend.app.models import TaskRun

pytestmark = pytest.mark.integration


@pytest.fixture(autouse=True)
def _reset_caches(monkeypatch):
    """Clear the module-level caches between tests so cache state from
    one test doesn't leak into the next."""
    monkeypatch.setitem(activity_module._QUEUE_CACHE, "data", None)
    monkeypatch.setitem(activity_module._QUEUE_CACHE, "ts", 0.0)
    monkeypatch.setitem(activity_module._WORKER_CACHE, "data", None)
    monkeypatch.setitem(activity_module._WORKER_CACHE, "ts", 0.0)


# --- /queues -------------------------------------------------------


@pytest.mark.asyncio
async def test_queues_returns_all_known_queues(client, monkeypatch):
    def _fake():
        return {
            "queues": dict.fromkeys(activity_module._QUEUE_NAMES, 0),
            "fetched_at": datetime.now(UTC).isoformat(),
        }
    monkeypatch.setattr(activity_module, "_read_queues_sync", _fake)

    resp = await client.get("/api/system/activity/queues")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert set(body["queues"].keys()) == set(activity_module._QUEUE_NAMES)


@pytest.mark.asyncio
async def test_summary_returns_rollup_shape(client, monkeypatch):
    def _fake():
        return {
            "queues": {**dict.fromkeys(activity_module._QUEUE_NAMES, 0), "ml": 3},
            "fetched_at": datetime.now(UTC).isoformat(),
        }
    monkeypatch.setattr(activity_module, "_read_queues_sync", _fake)

    resp = await client.get("/api/system/activity/summary")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert set(body.keys()) == {"scheduler", "queues", "queued_total", "running", "failing"}
    assert body["queued_total"] == 3  # only ml has a non-zero depth
    assert isinstance(body["running"], int)
    assert isinstance(body["failing"], int)
    assert set(body["scheduler"]) == {
        "last_tick_at", "next_due_at", "due_now", "auto_sources",
        "failing_sources", "no_access_sources",
        "platform_cooldowns",
    }


@pytest.mark.asyncio
async def test_queues_cached_within_ttl(client, monkeypatch):
    call_count = {"n": 0}

    def _counting():
        call_count["n"] += 1
        return {"queues": {}, "fetched_at": "x"}
    monkeypatch.setattr(activity_module, "_read_queues_sync", _counting)

    await client.get("/api/system/activity/queues")
    await client.get("/api/system/activity/queues")
    await client.get("/api/system/activity/queues")
    assert call_count["n"] == 1  # cached after the first call


@pytest.mark.asyncio
async def test_queues_redis_failure_returns_null_per_failing_queue(client, monkeypatch):
    """_read_queues_sync's per-queue try/except returns None on failure."""
    def _partial():
        out = dict.fromkeys(activity_module._QUEUE_NAMES, 0)
        out["ml"] = None  # simulate broker hiccup on the ml queue
        return {"queues": out, "fetched_at": "x"}
    monkeypatch.setattr(activity_module, "_read_queues_sync", _partial)

    resp = await client.get("/api/system/activity/queues")
    body = await resp.get_json()
    assert body["queues"]["ml"] is None
    assert body["queues"]["import"] == 0


# --- /workers ------------------------------------------------------


@pytest.mark.asyncio
async def test_workers_returns_per_worker_queue_membership(client, monkeypatch):
    def _fake():
        return {
            "workers": {
                "worker@host1": {"queues": ["import", "thumbnail"], "active_count": 2},
                "worker@host2": {"queues": ["ml"], "active_count": 0},
            },
            "fetched_at": "x",
        }
    monkeypatch.setattr(activity_module, "_read_workers_sync", _fake)

    resp = await client.get("/api/system/activity/workers")
    assert resp.status_code == 200
    body = await resp.get_json()
    assert "worker@host1" in body["workers"]
    assert "import" in body["workers"]["worker@host1"]["queues"]


@pytest.mark.asyncio
async def test_workers_cached_within_ttl(client, monkeypatch):
    call_count = {"n": 0}

    def _counting():
        call_count["n"] += 1
        return {"workers": {}, "fetched_at": "x"}
    monkeypatch.setattr(activity_module, "_read_workers_sync", _counting)

    await client.get("/api/system/activity/workers")
    await client.get("/api/system/activity/workers")
    assert call_count["n"] == 1


# --- /runs ---------------------------------------------------------


@pytest_asyncio.fixture
async def _seed_runs(db):
    """Insert 5 task_run rows for paging tests."""
    now = datetime.now(UTC)
    for i in range(5):
        db.add(TaskRun(
            celery_task_id=f"tid-{i}",
            queue="import" if i % 2 == 0 else "ml",
            task_name=f"backend.app.tasks.fake.task_{i}",
            target_id=i,
            started_at=now - timedelta(seconds=10 - i),
            finished_at=now - timedelta(seconds=9 - i),
            duration_ms=1000,
            status="ok" if i < 3 else "error",
            error_type="ValueError" if i >= 3 else None,
            error_message="boom" if i >= 3 else None,
        ))
    await db.commit()


@pytest.mark.asyncio
async def test_runs_paginated_descending_by_id(client, _seed_runs):
    resp = await client.get("/api/system/activity/runs?limit=3")
    body = await resp.get_json()
    assert len(body["runs"]) == 3
    # Descending: latest id first.
    ids = [r["id"] for r in body["runs"]]
    assert ids == sorted(ids, reverse=True)
    assert body["next_cursor"] is not None  # 5 total, asked for 3 → more


@pytest.mark.asyncio
async def test_runs_filter_by_queue(client, _seed_runs):
    resp = await client.get("/api/system/activity/runs?queue=ml")
    body = await resp.get_json()
    assert all(r["queue"] == "ml" for r in body["runs"])
    assert len(body["runs"]) == 2  # i=1, i=3


@pytest.mark.asyncio
async def test_runs_filter_by_status(client, _seed_runs):
    resp = await client.get("/api/system/activity/runs?status=error")
    body = await resp.get_json()
    assert all(r["status"] == "error" for r in body["runs"])
    assert len(body["runs"]) == 2  # i=3, i=4


@pytest.mark.asyncio
async def test_runs_filter_by_task_substring(client, _seed_runs):
    # Case-insensitive substring across the full task_name.
    resp = await client.get("/api/system/activity/runs?task=FAKE")
    body = await resp.get_json()
    assert len(body["runs"]) == 5
    assert all("fake" in r["task_name"] for r in body["runs"])


@pytest.mark.asyncio
async def test_runs_filter_by_task_escapes_underscore(client, _seed_runs):
    # The literal "_" in "task_3" must match one row, not act as a
    # single-char wildcard matching every task_N.
    resp = await client.get("/api/system/activity/runs?task=task_3")
    body = await resp.get_json()
    assert len(body["runs"]) == 1
    assert body["runs"][0]["task_name"].endswith("task_3")


@pytest.mark.asyncio
async def test_runs_keyset_cursor(client, _seed_runs):
    page1 = await (await client.get("/api/system/activity/runs?limit=2")).get_json()
    assert len(page1["runs"]) == 2
    assert page1["next_cursor"] is not None

    page2 = await (await client.get(
        f"/api/system/activity/runs?limit=2&before_id={page1['next_cursor']}"
    )).get_json()
    assert len(page2["runs"]) == 2
    page1_ids = {r["id"] for r in page1["runs"]}
    page2_ids = {r["id"] for r in page2["runs"]}
    assert page1_ids.isdisjoint(page2_ids)


@pytest.mark.asyncio
async def test_runs_invalid_limit_400(client):
    resp = await client.get("/api/system/activity/runs?limit=not-a-number")
    assert resp.status_code == 400


# --- /failures -----------------------------------------------------


@pytest_asyncio.fixture
async def _seed_failures(db):
    """Mix of ok / error / timeout rows; some old, some recent."""
    now = datetime.now(UTC)
    recent = [
        ("error", "OperationalError"),
        ("error", "OperationalError"),
        ("error", "OperationalError"),
        ("timeout", "TimeoutError"),
        ("timeout", "TimeoutError"),
        ("error", "OSError"),
        ("ok", None),  # success — should NOT appear in failures
    ]
    for i, (status, etype) in enumerate(recent):
        db.add(TaskRun(
            celery_task_id=f"f-{i}",
            queue="ml",
            task_name="backend.app.tasks.fake.x",
            target_id=i,
            started_at=now - timedelta(seconds=120 - i),
            finished_at=now - timedelta(seconds=60 - i),
            duration_ms=1000,
            status=status,
            error_type=etype,
            error_message="boom" if status != "ok" else None,
        ))
    # One ancient failure (>24h) that should NOT appear.
    db.add(TaskRun(
        celery_task_id="f-old",
        queue="ml",
        task_name="backend.app.tasks.fake.x",
        target_id=999,
        started_at=now - timedelta(hours=30),
        finished_at=now - timedelta(hours=29),
        duration_ms=1000,
        status="error",
        error_type="OldError",
        error_message="ancient",
    ))
    await db.commit()


@pytest.mark.asyncio
async def test_failures_returns_recent_errors_and_timeouts(client, _seed_failures):
    resp = await client.get("/api/system/activity/failures")
    body = await resp.get_json()
    statuses = {r["status"] for r in body["recent"]}
    assert statuses == {"error", "timeout"}
    # No ok rows.
    assert "ok" not in statuses


@pytest.mark.asyncio
async def test_failures_count_by_type_groups_correctly(client, _seed_failures):
    resp = await client.get("/api/system/activity/failures")
    body = await resp.get_json()
    counts = body["count_by_type"]
    assert counts.get("OperationalError") == 3
    assert counts.get("TimeoutError") == 2
    assert counts.get("OSError") == 1
    assert "OldError" not in counts  # outside 24h window


@pytest.mark.asyncio
async def test_failures_only_within_24h_window(client, _seed_failures):
    resp = await client.get("/api/system/activity/failures")
    body = await resp.get_json()
    ids = {r["error_type"] for r in body["recent"]}
    assert "OldError" not in ids