cfdaily
106 lines
3.1 KiB
Python
106 lines
3.1 KiB
Python
import os
|
||
import uuid
|
||
import atexit
|
||
import pytest
|
||
from pathlib import Path
|
||
from fastapi.testclient import TestClient
|
||
|
||
|
||
def pytest_configure(config):
|
||
markers = {
|
||
"unit": "单元测试：纯逻辑，mock 外部依赖",
|
||
"integration": "集成测试：API 端点 + 真实/临时 DB",
|
||
"e2e": "端到端测试：真实 daemon + Agent（手动触发）",
|
||
"slow": "慢测试（>5s）",
|
||
"broadcast": "广播认领相关",
|
||
"mail": "邮件系统相关",
|
||
"state_machine": "状态机转换",
|
||
"classify": "Classify Outcome 相关",
|
||
"review": "审查/Rebuttal 相关",
|
||
}
|
||
for name, desc in markers.items():
|
||
config.addinivalue_line("markers", f"{name}: {desc}")
|
||
|
||
# When RUN_INTEGRATION=1, remove default marker filter so integration/e2e tests run
|
||
if os.environ.get("RUN_INTEGRATION"):
|
||
config.known_args_namespace.markexpr = ""
|
||
|
||
|
||
@pytest.fixture
|
||
def isolated_data_root(tmp_path):
|
||
"""隔离的 data_root，测试结束自动清理"""
|
||
data_root = tmp_path / "test_data"
|
||
data_root.mkdir()
|
||
return data_root
|
||
|
||
|
||
@pytest.fixture
|
||
def isolated_registry(isolated_data_root):
|
||
"""隔离的 registry.db"""
|
||
from src.blackboard.registry import ProjectRegistry
|
||
registry = ProjectRegistry(isolated_data_root)
|
||
return registry
|
||
|
||
|
||
@pytest.fixture
|
||
def client_with_isolation(isolated_data_root):
|
||
"""带数据隔离的 TestClient"""
|
||
import src.utils as utils
|
||
original = utils.get_data_root
|
||
utils.get_data_root = lambda: isolated_data_root
|
||
|
||
from src.main import app
|
||
client = TestClient(app)
|
||
|
||
yield client
|
||
|
||
utils.get_data_root = original
|
||
|
||
|
||
# ── E2E gate ──
|
||
|
||
def pytest_collection_modifyitems(config, items):
|
||
if not os.environ.get("RUN_INTEGRATION"):
|
||
skip = pytest.mark.skip(reason="needs RUN_INTEGRATION=1")
|
||
for item in items:
|
||
if "integration" in item.keywords or "e2e" in item.keywords:
|
||
item.add_marker(skip)
|
||
|
||
|
||
skip_no_integration = pytest.mark.skipif(
|
||
not os.environ.get("RUN_INTEGRATION"),
|
||
reason="Set RUN_INTEGRATION=1 to run E2E tests against real daemon",
|
||
)
|
||
|
||
|
||
# ── atexit 兜底清理 ──
|
||
|
||
def _emergency_cleanup():
|
||
"""进程退出时的最后防线：清理所有 e2e- 前缀的测试数据"""
|
||
try:
|
||
import requests
|
||
api_base = os.environ.get("API_BASE", "http://localhost:8083")
|
||
# 清理 e2e 前缀项目
|
||
resp = requests.get(f"{api_base}/api/projects", timeout=5)
|
||
if resp.ok:
|
||
for proj in resp.json():
|
||
pid = proj.get("id", "")
|
||
if pid.startswith("e2e-"):
|
||
try:
|
||
requests.delete(
|
||
f"{api_base}/api/projects/{pid}?physical=true",
|
||
timeout=5,
|
||
)
|
||
except Exception:
|
||
pass
|
||
# 清理 e2e 前缀邮件
|
||
try:
|
||
requests.delete(f"{api_base}/api/mail?prefix=e2e-", timeout=5)
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
atexit.register(_emergency_cleanup)
|