fc3i: integration tests for /api/system/activity/* endpoints

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-24 21:06:20 -04:00
parent e523d0ac94
commit 48ef22445a
1 changed files with 267 additions and 0 deletions
@@ -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