auto-sync: 2026-06-05 23:19:24

tests/e2e/conftest.py

@@ -0,0 +1,112 @@
+"""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()