cfdaily
113 lines
3.0 KiB
Python
113 lines
3.0 KiB
Python
"""E2E conftest：清理 API fixture + session 兜底 + manifest"""
|
|
|
|
import os
|
|
import uuid
|
|
import atexit
|
|
import pytest
|
|
import requests as http_requests
|
|
|
|
API_BASE = os.environ.get("API_BASE", "http://localhost:8083")
|
|
POLL_INTERVAL = 5
|
|
MAX_WAIT_DISPATCH = 120
|
|
MAX_WAIT_AGENT = 300
|
|
E2E_PREFIX = "e2e-v27-"
|
|
|
|
|
|
def _check_environment():
|
|
"""前置检查：daemon 是否运行"""
|
|
try:
|
|
resp = http_requests.get(f"{API_BASE}/api/projects", timeout=5)
|
|
assert resp.status_code == 200, f"daemon 未响应: {resp.status_code}"
|
|
except Exception as e:
|
|
pytest.skip(f"daemon 未运行: {e}")
|
|
|
|
|
|
def _pid(prefix: str = E2E_PREFIX) -> str:
|
|
"""生成唯一测试项目ID"""
|
|
return f"{prefix}{uuid.uuid4().hex[:8]}"
|
|
|
|
|
|
def _tid() -> str:
|
|
"""生成唯一任务ID"""
|
|
return f"e2e-task-{uuid.uuid4().hex[:8]}"
|
|
|
|
|
|
def _cleanup_project(pid: str):
|
|
"""清理单个项目"""
|
|
try:
|
|
http_requests.delete(
|
|
f"{API_BASE}/api/projects/{pid}?physical=true", timeout=10
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _get_db_path(pid: str):
|
|
"""获取项目 blackboard.db 路径"""
|
|
from src.utils import get_data_root
|
|
return get_data_root() / pid / "blackboard.db"
|
|
|
|
|
|
def _create_project(tracked_list, name_prefix="E2E", agents=None):
|
|
"""创建测试项目并记录到 manifest"""
|
|
pid = _pid()
|
|
config = {}
|
|
if agents:
|
|
config["agents"] = agents
|
|
resp = http_requests.post(
|
|
f"{API_BASE}/api/projects",
|
|
json={"id": pid, "name": f"{name_prefix}-{pid}", "config": config},
|
|
timeout=10,
|
|
)
|
|
assert resp.status_code == 200, f"创建项目失败: {resp.text}"
|
|
tracked_list.append(pid)
|
|
return pid
|
|
|
|
|
|
def _create_task(pid, **kwargs):
|
|
"""创建测试任务"""
|
|
tid = kwargs.get("id") or _tid()
|
|
body = {"id": tid, "status": "pending", "priority": 5, **kwargs}
|
|
resp = http_requests.post(
|
|
f"{API_BASE}/api/projects/{pid}/tasks", json=body, timeout=10
|
|
)
|
|
assert resp.status_code == 200, f"创建任务失败: {resp.text}"
|
|
return tid
|
|
|
|
|
|
def _poll_task(pid, tid, timeout=MAX_WAIT_AGENT, terminal_states=("done", "failed", "cancelled")):
|
|
"""轮询任务状态直到终态"""
|
|
import time
|
|
start = time.time()
|
|
while time.time() - start < timeout:
|
|
resp = http_requests.get(
|
|
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
status = data.get("status", "")
|
|
if status in terminal_states:
|
|
return data
|
|
time.sleep(POLL_INTERVAL)
|
|
return {"status": "timeout", "tid": tid}
|
|
|
|
|
|
# ── Session 级 fixture ──
|
|
|
|
@pytest.fixture(scope="session")
|
|
def e2e_session_prefix():
|
|
"""Session 级唯一前缀"""
|
|
return f"e2e-v27-{uuid.uuid4().hex[:6]}-"
|
|
|
|
|
|
_session_manifest = []
|
|
|
|
|
|
@pytest.fixture(scope="session", autouse=True)
|
|
def e2e_session_cleanup():
|
|
"""Session 结束时兜底清理所有 e2e- 前缀项目"""
|
|
yield
|
|
for pid in list(_session_manifest):
|
|
_cleanup_project(pid)
|
|
_session_manifest.clear()
|