43ec5a4a71
## 后端变更 ### 任务管理核心 - 新增 tasker 服务，支持异步任务队列和工作线程池 - 添加任务执行时间追踪（started_at, completed_at） - 实现任务状态持久化（JSON文件存储） - 支持任务取消和进度更新 ### 知识库导入优化 - 将知识库文档导入改为异步任务处理 - 逐个处理文档并实时更新进度（5% → 95%） - 支持任务取消检测 ### API接口 - 新增任务列表、详情、取消等REST API - 任务API需要管理员权限 ### 测试 - 添加任务路由集成测试 ## 前端变更 ### 任务中心UI - 新增任务中心抽屉组件，采用简洁的卡片式设计 - 支持任务状态筛选（全部/进行中/已完成/失败） - 进行中任务显示实时进度条 - 已完成任务显示执行时长和完成状态徽章 - 支持相对时间显示（如：5分钟前） ### 状态管理 - 新增 tasker store，管理任务列表和轮询 - 支持任务注册、刷新、取消操作 ### 交互优化 - 在状态栏和侧边栏添加任务中心入口 - 显示活跃任务数量徽章 - 知识库导入自动注册到任务中心
94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
"""
|
|
Integration tests for the task management router.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
pytestmark = [pytest.mark.asyncio, pytest.mark.integration]
|
|
|
|
|
|
async def test_task_routes_require_admin(test_client, standard_user):
|
|
"""Non-admin users should be blocked from accessing task APIs."""
|
|
headers = standard_user["headers"]
|
|
|
|
list_response = await test_client.get("/api/tasks", headers=headers)
|
|
assert list_response.status_code == 403
|
|
|
|
detail_response = await test_client.get("/api/tasks/some-task", headers=headers)
|
|
assert detail_response.status_code == 403
|
|
|
|
cancel_response = await test_client.post("/api/tasks/some-task/cancel", headers=headers)
|
|
assert cancel_response.status_code == 403
|
|
|
|
|
|
async def test_admin_can_list_tasks(test_client, admin_headers):
|
|
"""Admin should receive a well-formed task list payload."""
|
|
response = await test_client.get("/api/tasks", headers=admin_headers)
|
|
assert response.status_code == 200, response.text
|
|
|
|
payload = response.json()
|
|
assert "tasks" in payload
|
|
assert isinstance(payload["tasks"], list)
|
|
|
|
|
|
async def test_cancel_unknown_task_returns_client_error(test_client, admin_headers):
|
|
"""Cancelling a non-existent task should surface a 400 response."""
|
|
response = await test_client.post("/api/tasks/not-real/cancel", headers=admin_headers)
|
|
assert response.status_code == 400, response.text
|
|
|
|
|
|
async def test_enqueue_document_creates_task(
|
|
test_client,
|
|
admin_headers,
|
|
knowledge_database,
|
|
):
|
|
"""Trigger knowledge ingestion to ensure a task record is materialised."""
|
|
db_id = knowledge_database["db_id"]
|
|
|
|
enqueue_response = await test_client.post(
|
|
f"/api/knowledge/databases/{db_id}/documents",
|
|
json={
|
|
"items": [],
|
|
"params": {"content_type": "file"},
|
|
},
|
|
headers=admin_headers,
|
|
)
|
|
assert enqueue_response.status_code == 200, enqueue_response.text
|
|
|
|
enqueue_payload = enqueue_response.json()
|
|
assert enqueue_payload.get("status") == "queued"
|
|
task_id = enqueue_payload.get("task_id")
|
|
assert task_id, "Knowledge ingestion did not return a task_id"
|
|
|
|
# The task should be queryable immediately after enqueueing.
|
|
detail_response = await test_client.get(f"/api/tasks/{task_id}", headers=admin_headers)
|
|
assert detail_response.status_code == 200, detail_response.text
|
|
detail_payload = detail_response.json().get("task", {})
|
|
assert detail_payload.get("id") == task_id
|
|
assert detail_payload.get("status") in {"queued", "pending", "running", "failed", "success", "cancelled"}
|
|
|
|
# Ensure the task surfaces in the list endpoint within a short window.
|
|
for _ in range(10):
|
|
list_response = await test_client.get("/api/tasks", headers=admin_headers)
|
|
assert list_response.status_code == 200, list_response.text
|
|
all_tasks = list_response.json().get("tasks", [])
|
|
if any(entry.get("id") == task_id for entry in all_tasks):
|
|
break
|
|
await asyncio.sleep(0.2)
|
|
else:
|
|
pytest.fail("Task did not appear in list endpoint within timeout window")
|
|
|
|
# Poll for terminal state to validate worker bookkeeping.
|
|
for _ in range(20):
|
|
detail_response = await test_client.get(f"/api/tasks/{task_id}", headers=admin_headers)
|
|
task_status = detail_response.json().get("task", {}).get("status")
|
|
if task_status in {"success", "failed", "cancelled"}:
|
|
break
|
|
await asyncio.sleep(0.5)
|
|
else:
|
|
pytest.fail("Task did not reach a terminal status within timeout window")
|