cc2ac8da89
All checks were successful
CI / test (push) Successful in 4m5s
GET /api/intelligent-evals 支持 page/page_size（默认不传仍返回全部，向后兼容）： repository 加 count/list_page，lifecycle 加 list_evals_page，router 返回 total。 前端服务端分页：useIntelligentEvalRead 接 page/pageSize，list 存 total， IntelligentEvals 表格 showSizeChanger + 页码切换重新加载；5s 轮询保持当前页。 测试：+3 后端分页 + hook 页码透传/总数断言，895 passed，vitest 19 passed
337 lines
11 KiB
Python
337 lines
11 KiB
Python
"""Integration tests for intelligent eval task queue API."""
|
||
|
||
from datetime import timedelta
|
||
|
||
import pytest
|
||
from agenteval.intelligent_eval.models import IntelligentEvalStatus
|
||
from agenteval.storage.db import IntelligentEvalDB, IntelligentEvalTaskQueueDB, utc_now
|
||
from agenteval.web.app import app
|
||
from agenteval.web.deps import get_db
|
||
from fastapi.testclient import TestClient
|
||
from sqlmodel import Session, SQLModel, create_engine
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(tmp_path):
|
||
"""Create a TestClient with a fresh database."""
|
||
from agenteval.storage.db import ( # noqa: F401
|
||
IntelligentEvalDB,
|
||
IntelligentEvalSessionDB,
|
||
IntelligentEvalTaskQueueDB,
|
||
)
|
||
|
||
engine = create_engine(
|
||
f"sqlite:///{tmp_path / 'test.db'}",
|
||
connect_args={"check_same_thread": False},
|
||
)
|
||
SQLModel.metadata.create_all(engine)
|
||
session = Session(engine)
|
||
|
||
def override_get_db():
|
||
try:
|
||
yield session
|
||
finally:
|
||
pass
|
||
|
||
app.dependency_overrides[get_db] = override_get_db
|
||
client = TestClient(app)
|
||
yield client
|
||
app.dependency_overrides.clear()
|
||
session.close()
|
||
engine.dispose()
|
||
|
||
|
||
@pytest.fixture()
|
||
def db_session(client):
|
||
"""Get the database session from the client fixture."""
|
||
# The session is stored in the dependency override
|
||
return next(app.dependency_overrides[get_db]())
|
||
|
||
|
||
def test_get_next_task_empty(client: TestClient):
|
||
"""Test getting next task when queue is empty."""
|
||
response = client.get("/api/intelligent-evals/tasks/next")
|
||
assert response.status_code == 200
|
||
assert response.json() == {"task": None}
|
||
|
||
|
||
def test_get_next_task_with_pending_task(client: TestClient, db_session: Session):
|
||
"""Test getting next task when there is a pending task."""
|
||
# Create eval
|
||
eval_db = IntelligentEvalDB(
|
||
name="test",
|
||
target_id="target1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
started_at=utc_now() - timedelta(hours=9),
|
||
)
|
||
eval_db.set_plan(
|
||
{
|
||
"time_distribution": [{"time_slot": "8-10h", "sessions": 2}],
|
||
"estimated_sessions": 2,
|
||
}
|
||
)
|
||
db_session.add(eval_db)
|
||
db_session.commit()
|
||
|
||
# Create pending task
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id=eval_db.id,
|
||
status="pending",
|
||
priority=10,
|
||
reason="slot_due",
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
# Get next task
|
||
response = client.get("/api/intelligent-evals/tasks/next")
|
||
assert response.status_code == 200
|
||
|
||
data = response.json()
|
||
assert data["task"] is not None
|
||
assert data["task"]["id"] == task.id
|
||
assert data["task"]["eval_id"] == eval_db.id
|
||
assert data["task"]["priority"] == 10
|
||
assert data["task"]["reason"] == "slot_due"
|
||
assert data["task"]["eval"]["id"] == eval_db.id
|
||
assert data["task"]["eval"]["name"] == "test"
|
||
assert data["task"]["eval"]["status"] == IntelligentEvalStatus.EXECUTING.value
|
||
|
||
|
||
def test_assign_task(client: TestClient, db_session: Session):
|
||
"""Test assigning a task to a cron."""
|
||
# Create pending task
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id="eval1",
|
||
status="pending",
|
||
priority=1,
|
||
reason="slot_due",
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
# Assign task
|
||
response = client.post(f"/api/intelligent-evals/tasks/{task.id}/assign?cron_id=cron1")
|
||
assert response.status_code == 200
|
||
assert response.json() == {"success": True}
|
||
|
||
# Verify assignment
|
||
db_session.refresh(task)
|
||
assert task.status == "assigned"
|
||
assert task.assigned_cron_id == "cron1"
|
||
|
||
|
||
def test_assign_task_not_found(client: TestClient):
|
||
"""Test assigning a non-existent task."""
|
||
response = client.post("/api/intelligent-evals/tasks/nonexistent/assign?cron_id=cron1")
|
||
assert response.status_code == 404
|
||
|
||
|
||
def test_complete_task(client: TestClient, db_session: Session):
|
||
"""Test completing a task."""
|
||
# Create assigned task
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id="eval1",
|
||
status="assigned",
|
||
priority=1,
|
||
reason="slot_due",
|
||
assigned_cron_id="cron1",
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
# Complete task
|
||
response = client.post(f"/api/intelligent-evals/tasks/{task.id}/complete?success=true")
|
||
assert response.status_code == 200
|
||
assert response.json() == {"success": True}
|
||
|
||
# Verify completion
|
||
db_session.refresh(task)
|
||
assert task.status == "completed"
|
||
assert task.completed_at is not None
|
||
|
||
|
||
def test_complete_task_with_error(client: TestClient, db_session: Session):
|
||
"""Test completing a task with error."""
|
||
# Create assigned task
|
||
task = IntelligentEvalTaskQueueDB(
|
||
eval_id="eval1",
|
||
status="assigned",
|
||
priority=1,
|
||
reason="slot_due",
|
||
assigned_cron_id="cron1",
|
||
)
|
||
db_session.add(task)
|
||
db_session.commit()
|
||
|
||
# Complete task with error
|
||
response = client.post(f"/api/intelligent-evals/tasks/{task.id}/complete?success=false&error=test_error")
|
||
assert response.status_code == 200
|
||
assert response.json() == {"success": True}
|
||
|
||
# Verify completion
|
||
db_session.refresh(task)
|
||
assert task.status == "failed"
|
||
assert task.error == "test_error"
|
||
|
||
|
||
def test_end_to_end_task_lifecycle(client: TestClient, db_session: Session):
|
||
"""Test end-to-end task lifecycle: create eval -> scan -> enqueue -> assign -> complete."""
|
||
# Create eval that needs attention
|
||
eval_db = IntelligentEvalDB(
|
||
name="test",
|
||
target_id="target1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
started_at=utc_now() - timedelta(hours=9),
|
||
)
|
||
eval_db.set_plan(
|
||
{
|
||
"time_distribution": [{"time_slot": "8-10h", "sessions": 2}],
|
||
"estimated_sessions": 2,
|
||
}
|
||
)
|
||
db_session.add(eval_db)
|
||
db_session.commit()
|
||
|
||
# Scan and enqueue tasks
|
||
from agenteval.intelligent_eval import task_queue
|
||
|
||
enqueued = task_queue.scan_and_enqueue_tasks(db_session)
|
||
assert enqueued == 1
|
||
|
||
# Get next task
|
||
response = client.get("/api/intelligent-evals/tasks/next")
|
||
assert response.status_code == 200
|
||
task_data = response.json()["task"]
|
||
assert task_data is not None
|
||
assert task_data["eval_id"] == eval_db.id
|
||
|
||
# Assign task
|
||
response = client.post(f"/api/intelligent-evals/tasks/{task_data['id']}/assign?cron_id=cron1")
|
||
assert response.status_code == 200
|
||
|
||
# Complete task
|
||
response = client.post(f"/api/intelligent-evals/tasks/{task_data['id']}/complete?success=true")
|
||
assert response.status_code == 200
|
||
|
||
# Verify task completed
|
||
task = db_session.get(IntelligentEvalTaskQueueDB, task_data["id"])
|
||
assert task.status == "completed"
|
||
|
||
|
||
def test_list_tasks(client: TestClient, db_session: Session):
|
||
"""Task list returns entries (newest first) with eval names and stats."""
|
||
eval_db = IntelligentEvalDB(
|
||
name="list-eval",
|
||
target_id="target1",
|
||
status=IntelligentEvalStatus.EXECUTING.value,
|
||
started_at=utc_now(),
|
||
)
|
||
eval_db.set_plan({"time_distribution": [{"time_slot": "0-1h", "sessions": 1}], "estimated_sessions": 1})
|
||
db_session.add(eval_db)
|
||
db_session.commit()
|
||
|
||
old = IntelligentEvalTaskQueueDB(eval_id=eval_db.id, status="completed", priority=5, reason="done")
|
||
new = IntelligentEvalTaskQueueDB(eval_id=eval_db.id, status="pending", priority=1, reason="slot_due")
|
||
db_session.add_all([old, new])
|
||
db_session.commit()
|
||
# 确保 old 早于 new（created_at 由 default_factory 生成，顺序可能同秒）
|
||
old.created_at = utc_now() - timedelta(seconds=5)
|
||
db_session.commit()
|
||
|
||
response = client.get("/api/intelligent-evals/tasks")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["stats"]["pending"] == 1
|
||
assert data["stats"]["completed"] == 1
|
||
assert data["stats"]["unresolved"] == 1
|
||
# newest first
|
||
assert [t["id"] for t in data["tasks"]] == [new.id, old.id]
|
||
task = data["tasks"][0]
|
||
assert task["eval_id"] == eval_db.id
|
||
assert task["eval_name"] == "list-eval"
|
||
assert task["eval_status"] == IntelligentEvalStatus.EXECUTING.value
|
||
assert task["priority"] == 1
|
||
assert task["reason"] == "slot_due"
|
||
|
||
|
||
def test_list_tasks_status_filter(client: TestClient, db_session: Session):
|
||
"""Status filter narrows the task list."""
|
||
db_session.add(IntelligentEvalTaskQueueDB(eval_id="eval1", status="pending", priority=1, reason="slot_due"))
|
||
db_session.add(IntelligentEvalTaskQueueDB(eval_id="eval1", status="failed", priority=1, reason="slot_due"))
|
||
db_session.commit()
|
||
|
||
response = client.get("/api/intelligent-evals/tasks?status=failed")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert len(data["tasks"]) == 1
|
||
assert data["tasks"][0]["status"] == "failed"
|
||
assert data["stats"]["failed"] == 1
|
||
|
||
|
||
def test_list_tasks_not_shadowed_by_eval_id(client: TestClient):
|
||
"""GET /tasks must hit the task-list endpoint, not /{eval_id} with eval_id="tasks"."""
|
||
response = client.get("/api/intelligent-evals/tasks")
|
||
assert response.status_code == 200
|
||
assert "tasks" in response.json()
|
||
|
||
|
||
def _make_eval(db_session: Session, name: str, days_ago: int) -> IntelligentEvalDB:
|
||
from agenteval.storage.db import utc_now
|
||
|
||
ev = IntelligentEvalDB(
|
||
name=name,
|
||
target_id="target1",
|
||
status=IntelligentEvalStatus.COMPLETED.value,
|
||
created_at=utc_now() - timedelta(days=days_ago),
|
||
started_at=utc_now() - timedelta(days=days_ago),
|
||
)
|
||
db_session.add(ev)
|
||
return ev
|
||
|
||
|
||
def test_list_evals_backward_compatible_no_page(client: TestClient, db_session: Session):
|
||
"""Without page param the endpoint returns everything (no total key)."""
|
||
_make_eval(db_session, "A", days_ago=3)
|
||
_make_eval(db_session, "B", days_ago=2)
|
||
db_session.commit()
|
||
|
||
response = client.get("/api/intelligent-evals")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert len(data["intelligent_evals"]) == 2
|
||
assert "total" not in data
|
||
|
||
|
||
def test_list_evals_pagination(client: TestClient, db_session: Session):
|
||
"""page/page_size returns one slice (newest first) plus total."""
|
||
_make_eval(db_session, "oldest", days_ago=3)
|
||
_make_eval(db_session, "middle", days_ago=2)
|
||
_make_eval(db_session, "newest", days_ago=1)
|
||
db_session.commit()
|
||
|
||
response = client.get("/api/intelligent-evals?page=1&page_size=2")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["total"] == 3
|
||
assert data["page"] == 1
|
||
assert data["page_size"] == 2
|
||
names = [e["name"] for e in data["intelligent_evals"]]
|
||
assert names == ["newest", "middle"] # created_at desc
|
||
|
||
response = client.get("/api/intelligent-evals?page=2&page_size=2")
|
||
names = [e["name"] for e in response.json()["intelligent_evals"]]
|
||
assert names == ["oldest"]
|
||
|
||
|
||
def test_list_evals_page_size_clamped(client: TestClient, db_session: Session):
|
||
"""page_size is clamped to [1, 100]."""
|
||
for i in range(5):
|
||
_make_eval(db_session, f"eval{i}", days_ago=10 - i)
|
||
db_session.commit()
|
||
|
||
response = client.get("/api/intelligent-evals?page=1&page_size=0")
|
||
data = response.json()
|
||
assert data["page_size"] == 1
|
||
assert len(data["intelligent_evals"]) == 1
|