"""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", "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