diff --git a/tests/test_api_system_activity.py b/tests/test_api_system_activity.py new file mode 100644 index 0000000..c13ae79 --- /dev/null +++ b/tests/test_api_system_activity.py @@ -0,0 +1,267 @@ +"""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 import create_app +from backend.app.api import system_activity as activity_module +from backend.app.models import TaskRun + +pytestmark = pytest.mark.integration + + +@pytest.fixture +async def app(): + return create_app() + + +@pytest.fixture +async def client(app): + async with app.test_client() as c: + yield c + + +@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": {q: 0 for q in activity_module._QUEUE_NAMES}, + "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_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 = {q: 0 for q in activity_module._QUEUE_NAMES} + 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_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