test(api): cover task admin endpoints

Author: haasonsaas

app.py

@@ -432,26 +432,6 @@ async def metrics() -> PlainTextResponse:
     return PlainTextResponse(content, media_type="text/plain; version=0.0.4")
 
 
-@app.get("/tasks/{task_id}")
-async def get_task(task_id: str, _admin_key: AdminKeyDep = None) -> dict[str, Any]:
-    """Get task status by ID."""
-    if not _task_queue:
-        raise HTTPException(status_code=503, detail="Task queue not initialized")
-    task = await _task_queue.get_task(task_id)
-    if not task:
-        raise HTTPException(status_code=404, detail="Task not found")
-    return {
-        "task_id": task.task_id,
-        "name": task.name,
-        "status": task.status.value,
-        "created_at": task.created_at.isoformat(),
-        "started_at": task.started_at.isoformat() if task.started_at else None,
-        "completed_at": task.completed_at.isoformat() if task.completed_at else None,
-        "retry_count": task.retry_count,
-        "error": task.error,
-    }
-
-
 @app.get("/tasks")
 async def list_tasks(status: str | None = None, limit: int = 50, _admin_key: AdminKeyDep = None) -> dict[str, Any]:
     """List all tasks with optional status filter."""
@@ -498,6 +478,26 @@ async def worker_status(_admin_key: AdminKeyDep = None) -> dict[str, Any]:
     return {"workers": await _task_queue.worker_heartbeats()}
 
 
+@app.get("/tasks/{task_id}")
+async def get_task(task_id: str, _admin_key: AdminKeyDep = None) -> dict[str, Any]:
+    """Get task status by ID."""
+    if not _task_queue:
+        raise HTTPException(status_code=503, detail="Task queue not initialized")
+    task = await _task_queue.get_task(task_id)
+    if not task:
+        raise HTTPException(status_code=404, detail="Task not found")
+    return {
+        "task_id": task.task_id,
+        "name": task.name,
+        "status": task.status.value,
+        "created_at": task.created_at.isoformat(),
+        "started_at": task.started_at.isoformat() if task.started_at else None,
+        "completed_at": task.completed_at.isoformat() if task.completed_at else None,
+        "retry_count": task.retry_count,
+        "error": task.error,
+    }
+
+
 @app.post("/prd/{plan_id}/versions")
 async def create_prd_version(
     plan_id: str,

tests/test_tasks_api.py

@@ -0,0 +1,101 @@
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+import app as app_module
+from agent_pm.api.auth import AdminKeyDep
+from agent_pm.settings import settings
+from app import app
+
+
+class AppClient:
+    def __init__(self):
+        self.transport = ASGITransport(app=app)
+        self.client = AsyncClient(transport=self.transport, base_url="http://test")
+
+    async def __aenter__(self):
+        await app_module.startup_event()
+        return self.client
+
+    async def __aexit__(self, exc_type, exc, tb):
+        await self.client.aclose()
+        await app_module.shutdown_event()
+
+
+async def _create_client():
+    return AppClient()
+
+
+@pytest.mark.asyncio
+async def test_tasks_admin_endpoints_with_memory_backend(monkeypatch):
+    monkeypatch.setattr(settings, "task_queue_backend", "memory")
+    app.dependency_overrides[AdminKeyDep] = lambda: "admin-test-key"
+
+    try:
+        async with await _create_client() as client:
+            response = await client.get("/tasks/dead-letter")
+            assert response.status_code == 200
+            assert response.json() == {"dead_letter": [], "total": 0}
+
+            worker_resp = await client.get("/tasks/workers")
+            assert worker_resp.status_code == 200
+            assert worker_resp.json() == {"workers": {}}
+    finally:
+        app.dependency_overrides.clear()
+
+
+class StubQueue:
+    def __init__(self):
+        self.limit: int | None = None
+        self.deleted: str | None = None
+
+    async def list_dead_letters(self, limit: int = 100):
+        self.limit = limit
+        return [
+            {
+                "task_id": "dead-1",
+                "name": "explode",
+                "retry_count": 3,
+                "last_error": "boom",
+            }
+        ]
+
+    async def delete_dead_letter(self, task_id: str) -> None:
+        self.deleted = task_id
+
+    async def worker_heartbeats(self) -> dict[str, dict[str, str]]:
+        return {"worker:1": {"status": "ok"}}
+
+    async def get_task(self, task_id: str):  # pragma: no cover - minimal stub for routing
+        return None
+
+
+@pytest.mark.asyncio
+async def test_tasks_admin_endpoints_surface_queue_data(monkeypatch):
+    monkeypatch.setattr(settings, "task_queue_backend", "memory")
+    app.dependency_overrides[AdminKeyDep] = lambda: "admin-test-key"
+
+    try:
+        async with await _create_client() as client:
+            original_backend = settings.task_queue_backend
+            original_queue = app_module._task_queue
+            stub = StubQueue()
+            app_module._task_queue = stub
+            settings.task_queue_backend = "memory"
+            try:
+                dead_resp = await client.get("/tasks/dead-letter", params={"limit": 5})
+                assert dead_resp.status_code == 200
+                assert dead_resp.json()["dead_letter"][0]["task_id"] == "dead-1"
+                assert stub.limit == 5
+
+                del_resp = await client.delete("/tasks/dead-letter/dead-1")
+                assert del_resp.status_code == 200
+                assert stub.deleted == "dead-1"
+
+                worker_resp = await client.get("/tasks/workers")
+                assert worker_resp.status_code == 200
+                assert worker_resp.json() == {"workers": {"worker:1": {"status": "ok"}}}
+            finally:
+                settings.task_queue_backend = original_backend
+                app_module._task_queue = original_queue
+    finally:
+        app.dependency_overrides.clear()