cfdaily
203 lines
7.1 KiB
Python
203 lines
7.1 KiB
Python
"""F5 测试：API 层"""
|
||
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from src.blackboard.operations import Blackboard
|
||
from src.blackboard.models import Task
|
||
from src.blackboard.registry import ProjectRegistry
|
||
from src.main import app
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
|
||
@pytest.fixture
|
||
def project_env(tmp_path):
|
||
"""创建临时项目环境"""
|
||
project_root = tmp_path / "projects"
|
||
project_root.mkdir()
|
||
os.environ["BLACKBOARD_ROOT"] = str(project_root)
|
||
|
||
# Create a test project with registry + DB
|
||
reg = ProjectRegistry(project_root)
|
||
reg.create_project("test-proj", "Test Project", agents=["agent1"])
|
||
|
||
bb = Blackboard(project_root / "test-proj" / "blackboard.db")
|
||
bb.create_task(Task(id="t1", title="Existing Task", task_type="coding"))
|
||
|
||
yield project_root
|
||
del os.environ["BLACKBOARD_ROOT"]
|
||
|
||
|
||
@pytest.fixture
|
||
def client():
|
||
return TestClient(app)
|
||
|
||
|
||
# ===================================================================
|
||
# Daemon API
|
||
# ===================================================================
|
||
|
||
class TestDaemonAPI:
|
||
def test_status(self, client):
|
||
resp = client.get("/api/daemon/status")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["status"] == "running"
|
||
|
||
def test_manual_tick(self, client):
|
||
resp = client.post("/api/daemon/tick")
|
||
assert resp.status_code == 200
|
||
|
||
|
||
# ===================================================================
|
||
# Project API
|
||
# ===================================================================
|
||
|
||
class TestProjectAPI:
|
||
def test_list_projects(self, client, project_env):
|
||
resp = client.get("/api/projects")
|
||
assert resp.status_code == 200
|
||
assert "test-proj" in resp.json()["projects"]
|
||
|
||
def test_create_project(self, client, tmp_path):
|
||
root = tmp_path / "new_root"
|
||
root.mkdir()
|
||
os.environ["BLACKBOARD_ROOT"] = str(root)
|
||
try:
|
||
resp = client.post("/api/projects", json={
|
||
"id": "new-proj", "name": "New",
|
||
"agents": ["a1"], "description": "test",
|
||
})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["ok"]
|
||
finally:
|
||
del os.environ["BLACKBOARD_ROOT"]
|
||
|
||
def test_create_duplicate(self, client, project_env):
|
||
resp = client.post("/api/projects", json={
|
||
"id": "test-proj", "name": "Dup",
|
||
})
|
||
assert resp.status_code == 409
|
||
|
||
def test_get_project(self, client, project_env):
|
||
resp = client.get("/api/projects/test-proj")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["name"] == "Test Project"
|
||
|
||
def test_get_nonexistent(self, client, project_env):
|
||
resp = client.get("/api/projects/nope")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# ===================================================================
|
||
# Blackboard API
|
||
# ===================================================================
|
||
|
||
class TestBlackboardAPI:
|
||
def test_list_tasks(self, client, project_env):
|
||
resp = client.get("/api/projects/test-proj/tasks")
|
||
assert resp.status_code == 200
|
||
assert len(resp.json()["tasks"]) == 1
|
||
|
||
def test_get_task(self, client, project_env):
|
||
resp = client.get("/api/projects/test-proj/tasks/t1")
|
||
assert resp.status_code == 200
|
||
assert resp.json()["title"] == "Existing Task"
|
||
|
||
def test_get_task_404(self, client, project_env):
|
||
resp = client.get("/api/projects/test-proj/tasks/nope")
|
||
assert resp.status_code == 404
|
||
|
||
def test_create_task(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks", json={
|
||
"id": "t2", "title": "New Task", "task_type": "review",
|
||
})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["task_id"] == "t2"
|
||
|
||
def test_claim_task(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/claim",
|
||
json={"agent": "agent1"})
|
||
assert resp.status_code == 200
|
||
|
||
def test_update_status(self, client, project_env):
|
||
# Claim first
|
||
client.post("/api/projects/test-proj/tasks/t1/claim",
|
||
json={"agent": "agent1"})
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/status",
|
||
json={"status": "working", "agent": "agent1"})
|
||
assert resp.status_code == 200
|
||
|
||
def test_invalid_status_transition(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/status",
|
||
json={"status": "done"})
|
||
assert resp.status_code == 409
|
||
|
||
def test_add_comment(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/comments", json={
|
||
"author": "pangtong", "body": "Nice", "comment_type": "general",
|
||
"mentions": ["agent1"],
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
def test_get_comments(self, client, project_env):
|
||
client.post("/api/projects/test-proj/tasks/t1/comments", json={
|
||
"author": "a", "body": "Hello",
|
||
})
|
||
resp = client.get("/api/projects/test-proj/tasks/t1/comments")
|
||
assert resp.status_code == 200
|
||
assert len(resp.json()["comments"]) == 1
|
||
|
||
def test_write_output(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/outputs", json={
|
||
"agent": "agent1", "type": "code", "title": "main.py",
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
def test_add_decision(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/decisions", json={
|
||
"decider": "pangtong", "decision": "Use X",
|
||
"rationale": "Better",
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
def test_add_observation(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/observations", json={
|
||
"observer": "simayi", "body": "Warning!", "severity": "warning",
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
def test_add_review(self, client, project_env):
|
||
resp = client.post("/api/projects/test-proj/tasks/t1/reviews", json={
|
||
"id": "rev-1", "reviewer": "simayi",
|
||
"review_type": "output_review", "verdict": "approved",
|
||
"summary": "LGTM", "confidence": 0.9,
|
||
})
|
||
assert resp.status_code == 200
|
||
|
||
def test_get_events(self, client, project_env):
|
||
resp = client.get("/api/projects/test-proj/events")
|
||
assert resp.status_code == 200
|
||
assert "events" in resp.json()
|
||
|
||
def test_summary(self, client, project_env):
|
||
resp = client.get("/api/projects/test-proj/summary")
|
||
assert resp.status_code == 200
|
||
assert "pending" in resp.json()["summary"]
|
||
|
||
|
||
# ===================================================================
|
||
# SSE
|
||
# ===================================================================
|
||
|
||
class TestSSE:
|
||
def test_sse_endpoint_registered(self):
|
||
"""SSE 路由已注册（详细测试见 test_sse.py）"""
|
||
from src.main import app
|
||
routes = [r.path for r in app.routes]
|
||
assert "/api/events" in routes
|