cfdaily
1820 lines
68 KiB
Python
1820 lines
68 KiB
Python
"""S9-S21 E2E 场景测试
|
||
|
||
从 test_e2e_v27.py (E9-E14) 和 test_e2e_v31.py (E94-E98/E10c/E10d/E15) 合并。
|
||
通过生产 HTTP API 测试真实 daemon + Agent 场景。
|
||
|
||
需要 RUN_INTEGRATION=1 + 生产 daemon 运行。
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import sqlite3
|
||
import sys
|
||
import time
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any, Dict
|
||
|
||
import pytest
|
||
import requests as http_requests
|
||
|
||
# 指向部署目录
|
||
DEPLOY_DIR = Path.home() / ".sanguo_projects" / "sanguo_moziplus_v2"
|
||
sys.path.insert(0, str(DEPLOY_DIR))
|
||
|
||
from src.utils import get_data_root
|
||
|
||
# ── 常量 ──
|
||
|
||
API_BASE = os.environ.get("API_BASE", "http://localhost:8083")
|
||
POLL_INTERVAL = 5
|
||
MAX_WAIT_DISPATCH = 120
|
||
MAX_WAIT_AGENT = 300
|
||
E2E_PREFIX = "e2e-v30-"
|
||
DATA_ROOT = get_data_root()
|
||
|
||
pytestmark = pytest.mark.e2e
|
||
|
||
|
||
# ── E2E gate ──
|
||
|
||
skip_no_integration = pytest.mark.skipif(
|
||
not os.environ.get("RUN_INTEGRATION"),
|
||
reason="Set RUN_INTEGRATION=1 to run E2E tests against real daemon",
|
||
)
|
||
|
||
|
||
# ── 工具函数 ──
|
||
|
||
def _check_environment():
|
||
"""环境前置检查：daemon 运行 + ticker 活跃 + 8083 可达"""
|
||
try:
|
||
resp = http_requests.get(f"{API_BASE}/api/daemon/status", timeout=5)
|
||
data = resp.json()
|
||
if data.get("status") != "running" or not data.get("ticker_running"):
|
||
pytest.skip(f"Daemon not ready: {data}")
|
||
return data
|
||
except Exception as e:
|
||
pytest.skip(f"Production API not available at {API_BASE}: {e}")
|
||
|
||
|
||
def _cleanup_project(pid: str):
|
||
"""清理测试项目"""
|
||
try:
|
||
http_requests.post(f"{API_BASE}/api/projects/{pid}/archive", timeout=5)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _poll_task(pid, tid, timeout, terminal_states=None):
|
||
"""轮询任务状态直到终态或超时"""
|
||
terminal = terminal_states or ("done", "failed", "cancelled")
|
||
deadline = time.time() + timeout
|
||
last_status = None
|
||
while time.time() < deadline:
|
||
try:
|
||
resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
last_status = data.get("status")
|
||
if last_status in terminal:
|
||
return data
|
||
except Exception:
|
||
pass
|
||
time.sleep(POLL_INTERVAL)
|
||
# 超时，返回最后状态
|
||
try:
|
||
resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10
|
||
)
|
||
return resp.json() if resp.status_code == 200 else {"status": "unknown"}
|
||
except Exception:
|
||
return {"status": "unknown"}
|
||
|
||
|
||
def _get_db_path(pid: str) -> Path:
|
||
"""获取项目的 blackboard.db 路径"""
|
||
return DATA_ROOT / pid / "blackboard.db"
|
||
|
||
|
||
def _patch_db_claimed_at(pid: str, tid: str, claimed_at: str):
|
||
"""直接操作 DB 设置 claimed_at 时间戳（模拟超时）"""
|
||
db_path = _get_db_path(pid)
|
||
assert db_path.exists(), f"DB not found: {db_path}"
|
||
conn = sqlite3.connect(str(db_path))
|
||
try:
|
||
conn.execute(
|
||
"UPDATE tasks SET claimed_at=? WHERE id=?",
|
||
(claimed_at, tid),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
# ===================================================================
|
||
# S9: 真实 Agent 调度
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS9RealAgentDispatch:
|
||
"""S9: 真实 Agent 调度测试
|
||
|
||
通过生产 HTTP API 创建任务，依赖生产 Ticker 自动调度，
|
||
真实 Agent spawn 执行，全程不手动推动状态。
|
||
"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S9") -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
resp = http_requests.post(f"{API_BASE}/api/projects", json={
|
||
"id": pid,
|
||
"name": f"{name_prefix}-{pid}",
|
||
"config": {"agents": ["zhangfei-dev"]},
|
||
}, timeout=10)
|
||
assert resp.status_code == 200, f"Create project failed: {resp.text}"
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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"Create task failed: {resp.text}"
|
||
return tid
|
||
|
||
def test_s91_simple_task_agent_execute(self):
|
||
"""S9-1: 简单任务 → 生产Ticker调度 → 真实Agent执行 → 状态自动流转到终态"""
|
||
pid = self._create_project("S9-1")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E简单任务：echo hello",
|
||
description=(
|
||
"请执行以下操作：\n"
|
||
"1. 运行命令 echo hello\n"
|
||
"2. 把输出结果写入黑板\n"
|
||
"这是E2E自动化测试，完成后请标记done。\n"
|
||
"重要：不需要做其他任何事。"
|
||
),
|
||
assignee="zhangfei-dev",
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S9-1: 等待调度 + Agent执行 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "pending", (
|
||
f"Agent未被调度！任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending。"
|
||
)
|
||
assert status != "blocked", f"普通任务被错误拦截: {result}"
|
||
|
||
if status == "failed":
|
||
print(f" ⚠️ Agent执行失败，detail: {result}")
|
||
else:
|
||
print(f" ✅ Agent执行成功")
|
||
|
||
print(f" 性能: pending → {status}")
|
||
|
||
def test_s92_review_task_dispatch(self):
|
||
"""S9-2: review任务 → 路由到can_review Agent → 真实执行"""
|
||
pid = self._create_project("S9-2")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E Review：审查测试",
|
||
description=(
|
||
"这是一个E2E测试的review任务。\n"
|
||
"请简单回复：\"审查通过，E2E测试正常。\"\n"
|
||
"然后标记done即可。不需要做其他事。"
|
||
),
|
||
assignee="simayi-challenger",
|
||
task_type="review",
|
||
)
|
||
|
||
print(f"\n🚀 S9-2: 等待review调度 + Agent执行 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "pending", (
|
||
f"Review Agent未被调度！任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending。"
|
||
)
|
||
assert status != "blocked", f"Review任务被错误拦截: {result}"
|
||
|
||
if status == "failed":
|
||
print(f" ⚠️ Review Agent执行失败")
|
||
else:
|
||
print(f" ✅ Review Agent执行成功")
|
||
|
||
def test_s93_guardrail_block(self):
|
||
"""S9-3: 实盘交易任务 → Guardrails拦截 → status=blocked"""
|
||
pid = self._create_project("S9-3")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="执行实盘买入SH600000",
|
||
description="用真金白银买入浦发银行1000股",
|
||
assignee="zhangfei-dev",
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S9-3: 等待Guardrails拦截 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("done", "failed", "cancelled", "blocked", "claimed", "working"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status not in ("claimed", "working", "done"), (
|
||
f"Guardrails未拦截实盘任务！状态: {status}"
|
||
)
|
||
print(f" ✅ Guardrails拦截生效，状态: {status}")
|
||
|
||
|
||
# ===================================================================
|
||
# S10: 全链路集成
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS10FullChain:
|
||
"""S10: 项目 → 父子Task → 生产Ticker → 聚合 → 依赖 → Mail → 前端API"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S10") -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
resp = http_requests.post(f"{API_BASE}/api/projects", json={
|
||
"id": pid,
|
||
"name": f"{name_prefix}-{pid}",
|
||
"config": {"agents": ["zhangfei-dev", "simayi-challenger"]},
|
||
}, timeout=10)
|
||
assert resp.status_code == 200
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def test_s10a_logic_chain(self):
|
||
"""S10a: 父子Task + 依赖推进 + 状态聚合 + Mail（不依赖Agent完成）"""
|
||
pid = self._create_project("S10a")
|
||
|
||
# 1. 创建父任务（带stages）
|
||
parent_id = f"{pid}-parent"
|
||
http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
|
||
"id": parent_id,
|
||
"title": "S10a全链路父任务",
|
||
"status": "pending",
|
||
"stages_json": json.dumps([
|
||
{"id": "setup", "label": "Setup"},
|
||
{"id": "execute", "label": "Execute"},
|
||
{"id": "verify", "label": "Verify"},
|
||
]),
|
||
}, timeout=10)
|
||
|
||
# 2. 创建3个子任务
|
||
child_ids = []
|
||
for i, stage in enumerate(["setup", "execute", "verify"]):
|
||
cid = f"{pid}-child-{i}"
|
||
child_ids.append(cid)
|
||
http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
|
||
"id": cid,
|
||
"title": f"子任务-{stage}",
|
||
"status": "pending",
|
||
"parent_task": parent_id,
|
||
"stage": stage,
|
||
}, timeout=10)
|
||
|
||
# 3. 创建依赖任务（blocked）
|
||
dep_id = f"{pid}-dep"
|
||
http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
|
||
"id": dep_id,
|
||
"title": "依赖任务",
|
||
"depends_on": json.dumps([child_ids[0]]),
|
||
}, timeout=10)
|
||
|
||
# 4. 推进依赖任务到 blocked
|
||
for status in ["claimed", "working", "blocked"]:
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{dep_id}/status",
|
||
json={"status": status, "agent": "test"}, timeout=10,
|
||
)
|
||
|
||
# 5. 推进 setup 子任务到 done（触发依赖推进条件）
|
||
for status in ["claimed", "working", "review", "done"]:
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{child_ids[0]}/status",
|
||
json={"status": status, "agent": "test"}, timeout=10,
|
||
)
|
||
|
||
# 6. 等待 Ticker 依赖推进
|
||
print(f"\n🚀 S10a: 等待依赖推进 (pid={pid})")
|
||
dep_result = _poll_task(
|
||
pid, dep_id, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("pending", "claimed", "working", "done"),
|
||
)
|
||
dep_status = dep_result.get("status")
|
||
print(f" 依赖任务状态: blocked → {dep_status}")
|
||
assert dep_status != "blocked", (
|
||
f"依赖推进未生效！{dep_id} 在 {MAX_WAIT_DISPATCH}s 后仍为 blocked"
|
||
)
|
||
|
||
# 7. 发送 Mail
|
||
http_requests.post(f"{API_BASE}/api/mail", json={
|
||
"title": f"S10a全链路通知-{pid}",
|
||
"text": f"项目 {pid} setup阶段完成",
|
||
"from": "simayi-challenger",
|
||
"to": "pangtong-fujunshi",
|
||
"type": "inform",
|
||
}, timeout=10)
|
||
|
||
# 8. 验证 Mail
|
||
resp = http_requests.get(
|
||
f"{API_BASE}/api/mail?from_agent=simayi-challenger", timeout=10
|
||
)
|
||
mails = resp.json()["mails"]
|
||
assert any(m["title"].startswith("S10a全链路") for m in mails), \
|
||
f"Mail未找到，当前mails: {[m['title'] for m in mails]}"
|
||
|
||
# 9. 验证 Stage 进度
|
||
resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{parent_id}/progress", timeout=10
|
||
)
|
||
if resp.status_code == 200:
|
||
progress = resp.json()
|
||
stages = {s["id"]: s for s in progress.get("stages", [])}
|
||
assert stages["setup"]["done"] == 1, f"setup stage should be done"
|
||
assert stages["execute"]["done"] == 0
|
||
|
||
# 10. 验证父状态聚合
|
||
parent_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{parent_id}", timeout=10
|
||
)
|
||
parent_data = parent_resp.json()
|
||
print(f" 父任务状态: {parent_data.get('status')}")
|
||
assert parent_data.get("status") in ("pending", "working"), \
|
||
f"父任务状态异常: {parent_data['status']}"
|
||
|
||
print(f" ✅ S10a 全链路测试通过")
|
||
|
||
def test_s10b_agent_full_chain(self):
|
||
"""S10b: 真实Agent全链路 — 创建子任务 → Agent执行 → 父状态自动聚合"""
|
||
pid = self._create_project("S10b")
|
||
|
||
# 1. 创建父任务
|
||
parent_id = f"{pid}-parent"
|
||
http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
|
||
"id": parent_id,
|
||
"title": "S10b Agent全链路父任务",
|
||
"status": "pending",
|
||
"stages_json": json.dumps([
|
||
{"id": "step1", "label": "Step 1"},
|
||
{"id": "step2", "label": "Step 2"},
|
||
]),
|
||
}, timeout=10)
|
||
|
||
# 2. 创建子任务（真实Agent执行）
|
||
child_id = f"{pid}-child-0"
|
||
http_requests.post(f"{API_BASE}/api/projects/{pid}/tasks", json={
|
||
"id": child_id,
|
||
"title": "E2E子任务：echo test",
|
||
"description": (
|
||
"请执行 echo test 并标记done。"
|
||
"这是E2E测试，不需要做其他事。"
|
||
),
|
||
"status": "pending",
|
||
"parent_task": parent_id,
|
||
"stage": "step1",
|
||
"assignee": "zhangfei-dev",
|
||
"task_type": "coding",
|
||
}, timeout=10)
|
||
|
||
# 3. 等待Agent执行完成
|
||
print(f"\n🚀 S10b: 等待Agent执行子任务 (pid={pid}, tid={child_id})")
|
||
result = _poll_task(
|
||
pid, child_id, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
child_status = result.get("status")
|
||
print(f" 子任务最终状态: {child_status}")
|
||
|
||
assert child_status != "pending", (
|
||
f"子任务未被调度！{MAX_WAIT_AGENT}s后仍为pending"
|
||
)
|
||
|
||
# 4. 轮询父任务状态变化
|
||
parent_result = _poll_task(
|
||
pid, parent_id, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("working", "review", "done", "failed"),
|
||
)
|
||
parent_data = parent_result
|
||
print(f" 父任务状态: {parent_data.get('status')}")
|
||
|
||
# 5. 验证 Stage 进度
|
||
resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{parent_id}/progress", timeout=10
|
||
)
|
||
if resp.status_code == 200:
|
||
progress = resp.json()
|
||
stages = {s["id"]: s for s in progress.get("stages", [])}
|
||
print(f" Stage进度: step1.done={stages.get('step1', {}).get('done', '?')}")
|
||
|
||
print(f" ✅ S10b 真实Agent全链路完成")
|
||
|
||
|
||
# ===================================================================
|
||
# S11: Acquire-First 真实 Agent E2E
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS11AcquireFirstE2E:
|
||
"""S11: Acquire-First Phase 1-4 真实 Agent E2E
|
||
|
||
验证真实 daemon 调度路径中 counter acquire + session check + spawn 的完整流程。
|
||
"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S11") -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
resp = http_requests.post(f"{API_BASE}/api/projects", json={
|
||
"id": pid,
|
||
"name": f"{name_prefix}-{pid}",
|
||
"config": {"agents": ["zhangfei-dev"]},
|
||
}, timeout=10)
|
||
assert resp.status_code == 200
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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
|
||
return tid
|
||
|
||
def test_s111a_dispatch_with_routing_audit(self):
|
||
"""S11a: 任务调度 → 检查 routing_decisions 记录了 dispatch + selected_agent"""
|
||
pid = self._create_project("S11a")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E Acquire-First：echo hello",
|
||
description=(
|
||
"请执行 echo hello-world 并标记done。\n"
|
||
"这是E2E测试 Acquire-First，不需要做其他事。"
|
||
),
|
||
assignee="zhangfei-dev",
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S11a: 等待调度 + Agent执行 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "pending", (
|
||
f"任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending，调度未生效"
|
||
)
|
||
|
||
# 验证 routing_decisions 有记录
|
||
db_path = DATA_ROOT / pid / "blackboard.db"
|
||
if db_path.exists():
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
conn.row_factory = sq3.Row
|
||
try:
|
||
tables = [r[0] for r in conn.execute(
|
||
"SELECT name FROM sqlite_master WHERE type='table'"
|
||
).fetchall()]
|
||
if "routing_decisions" not in tables:
|
||
print(f" ⚠️ routing_decisions 表不存在，跳过审计验证")
|
||
else:
|
||
rows = conn.execute(
|
||
"SELECT * FROM routing_decisions WHERE task_id=? ORDER BY id DESC LIMIT 1",
|
||
(tid,),
|
||
).fetchall()
|
||
if len(rows) > 0:
|
||
row = rows[0]
|
||
print(f" routing: mode={row['mode']} agent={row['selected_agent']} outcome={row['outcome']}")
|
||
assert row["selected_agent"] == "zhangfei-dev"
|
||
assert row["outcome"] == "dispatched"
|
||
else:
|
||
print(f" ⚠️ routing_decisions 无记录 for {tid}")
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f" ✅ Acquire-First 调度审计验证通过")
|
||
|
||
def test_s111b_concurrent_dispatch_counter_block(self):
|
||
"""S11b: 同 agent 连续两个任务 → 第二个应被 skip"""
|
||
pid = self._create_project("S11b")
|
||
|
||
tid1 = self._create_task(
|
||
pid,
|
||
title="E2E Counter-Block 任务1：echo first",
|
||
description="请执行 echo first 并标记done。E2E测试，不需要做其他事。",
|
||
assignee="zhangfei-dev",
|
||
task_type="coding",
|
||
)
|
||
tid2 = self._create_task(
|
||
pid,
|
||
title="E2E Counter-Block 任务2：echo second",
|
||
description="请执行 echo second 并标记done。E2E测试，不需要做其他事。",
|
||
assignee="zhangfei-dev",
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S11b: 等待两个任务调度 (pid={pid})")
|
||
|
||
result1 = _poll_task(
|
||
pid, tid1, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status1 = result1.get("status")
|
||
print(f" 任务1状态: {status1}")
|
||
assert status1 != "pending", "任务1未被调度"
|
||
|
||
result2 = _poll_task(
|
||
pid, tid2, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status2 = result2.get("status")
|
||
print(f" 任务2状态: {status2}")
|
||
|
||
db_path = DATA_ROOT / pid / "blackboard.db"
|
||
if db_path.exists():
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
conn.row_factory = sq3.Row
|
||
try:
|
||
rows = conn.execute(
|
||
"SELECT outcome, detail FROM routing_decisions WHERE task_id=? ORDER BY id",
|
||
(tid2,),
|
||
).fetchall()
|
||
outcomes = [r["outcome"] for r in rows]
|
||
print(f" 任务2 routing outcomes: {outcomes}")
|
||
assert len(rows) > 0, "任务2无 routing 记录"
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f" ✅ Counter block 并发调度验证完成")
|
||
|
||
|
||
# ===================================================================
|
||
# S12: _check_timeouts 统一 + crash_limit E2E
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS12TimeoutsUnifiedE2E:
|
||
"""S12: _check_timeouts 统一超时 + crash_limit + updated_at fallback"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S12") -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
resp = http_requests.post(f"{API_BASE}/api/projects", json={
|
||
"id": pid,
|
||
"name": f"{name_prefix}-{pid}",
|
||
"config": {"agents": ["zhangfei-dev"]},
|
||
}, timeout=10)
|
||
assert resp.status_code == 200
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def test_s121a_crash_limit_marks_failed(self):
|
||
"""S12a: 3 次 crash → _check_timeouts 标 failed"""
|
||
pid = self._create_project("S12a")
|
||
|
||
tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks",
|
||
json={"id": tid, "title": "Crash Limit 测试", "status": "pending",
|
||
"assignee": "zhangfei-dev"},
|
||
timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
|
||
# 写 3 条 crash attempt
|
||
db_path = DATA_ROOT / pid / "blackboard.db"
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
try:
|
||
for i in range(3):
|
||
conn.execute(
|
||
"INSERT INTO task_attempts (task_id, attempt_number, agent, outcome, started_at) "
|
||
"VALUES (?, ?, ?, ?, datetime('now', ?))",
|
||
(tid, i + 1, "zhangfei-dev", "crashed", f"-{(25 - i * 5)} minutes"),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f"\n🚀 S12a: 3次crash已写入，等待ticker处理 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("failed", "done", "cancelled"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status == "failed", (
|
||
f"3次crash后任务应为failed，实际: {status}"
|
||
)
|
||
print(f" ✅ crash_limit 统一检查验证通过")
|
||
|
||
def test_s121b_updated_at_fallback_reclaim(self):
|
||
"""S12b: working 任务无 started_at → updated_at fallback → 超时回收"""
|
||
pid = self._create_project("S12b")
|
||
|
||
tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks",
|
||
json={"id": tid, "title": "updated_at Fallback 测试", "status": "pending",
|
||
"assignee": "zhangfei-dev"},
|
||
timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
|
||
# 清空 started_at/claimed_at，设 updated_at 为 60 分钟前
|
||
db_path = DATA_ROOT / pid / "blackboard.db"
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
try:
|
||
conn.execute(
|
||
"UPDATE tasks SET started_at=NULL, claimed_at=NULL, "
|
||
"updated_at=datetime('now','-60 minutes') WHERE id=?",
|
||
(tid,),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f"\n🚀 S12b: updated_at 设为60min前，等待ticker回收 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("failed", "done", "cancelled"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "working", (
|
||
f"超时任务应被回收，实际仍为 working"
|
||
)
|
||
print(f" ✅ updated_at fallback 回收验证通过")
|
||
|
||
|
||
# ===================================================================
|
||
# S13: Compact Hanging 不标 failed E2E
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS13CompactHangingE2E:
|
||
"""S13: compact_hanging outcome → 任务保持 working（不标 failed）"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S13") -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
resp = http_requests.post(f"{API_BASE}/api/projects", json={
|
||
"id": pid,
|
||
"name": f"{name_prefix}-{pid}",
|
||
"config": {"agents": ["zhangfei-dev"]},
|
||
}, timeout=10)
|
||
assert resp.status_code == 200
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def test_s131a_compact_hanging_keeps_working(self):
|
||
"""S13a: compact_hanging attempt → ticker 不应立即标 failed"""
|
||
pid = self._create_project("S13a")
|
||
|
||
tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks",
|
||
json={"id": tid, "title": "Compact Hanging 测试", "status": "pending",
|
||
"assignee": "zhangfei-dev"},
|
||
timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
|
||
# 写入 compact_hanging attempt
|
||
db_path = DATA_ROOT / pid / "blackboard.db"
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO task_attempts (task_id, attempt_number, agent, outcome, started_at) "
|
||
"VALUES (?, ?, ?, ?, datetime('now'))",
|
||
(tid, 1, "zhangfei-dev", "compact_hanging"),
|
||
)
|
||
conn.execute(
|
||
"UPDATE tasks SET started_at=datetime('now','-5 minutes') WHERE id=?",
|
||
(tid,),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f"\n🚀 S13a: compact_hanging attempt 已写入，等一个 tick 验证不被标 failed")
|
||
time.sleep(45)
|
||
|
||
resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
assert resp.status_code == 200
|
||
status = resp.json().get("status")
|
||
print(f" 一个 tick 后状态: {status}")
|
||
|
||
assert status == "working", (
|
||
f"compact_hanging 后任务应保持 working，实际: {status}"
|
||
)
|
||
print(f" ✅ compact_hanging 不标 failed 验证通过")
|
||
|
||
|
||
# ===================================================================
|
||
# S14: AgentBusyError 分类 E2E
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS14AgentBusyError:
|
||
"""S14: AgentBusyError reason/detail 分类 E2E
|
||
|
||
验证真实 daemon 调度路径中 AgentBusyError 的三种分类：
|
||
- S14.1 counter_blocked：同 agent 并发 acquire → 第二个 blocked
|
||
- S14.2 session_locked：agent session 被占用 → session_locked blocker
|
||
- S14.3 Dispatcher 错误区分：AgentBusyError 写入 routing_decisions
|
||
"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S14", agents=None) -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
config = {"agents": agents or ["zhangfei-dev"]}
|
||
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
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs):
|
||
tid = _tid()
|
||
body = {
|
||
"id": tid,
|
||
"title": kwargs.pop("title", f"S14-task-{tid[:8]}"),
|
||
"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
|
||
return tid
|
||
|
||
def test_s141_counter_blocked(self):
|
||
"""S14.1: 同 agent 并发 acquire → 第二个任务 AgentBusyError(reason=counter_blocked)"""
|
||
pid = self._create_project(agents=["zhangfei-dev"])
|
||
|
||
# 创建第一个任务并等待被 dispatch
|
||
tid1 = self._create_task(pid, assignee="zhangfei-dev")
|
||
print(f" 任务1 {tid1} 已创建，等待 dispatch...")
|
||
|
||
# 等待第一个任务被 dispatch（进入 working）
|
||
task1 = _poll_task(pid, tid1, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("working", "done", "failed"))
|
||
status1 = task1.get("status", "")
|
||
print(f" 任务1 状态: {status1}")
|
||
assert status1 in ("working", "done"), (
|
||
f"任务1 应被 dispatch，实际: {status1}"
|
||
)
|
||
|
||
# 第二个任务指定同一 agent，等 counter acquire 冲突
|
||
tid2 = self._create_task(pid, assignee="zhangfei-dev")
|
||
print(f" 任务2 {tid2} 已创建，等待 counter blocked...")
|
||
|
||
# 等待足够 tick 让 dispatcher 尝试 acquire
|
||
task2 = _poll_task(pid, tid2, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("working", "done", "failed"))
|
||
status2 = task2.get("status", "")
|
||
print(f" 任务2 状态: {status2}")
|
||
|
||
# 任务2 可能被 blocked（保持 pending/claimed）或排队后执行
|
||
# 关键验证：检查 routing_decisions 是否记录了 counter_blocked
|
||
db_path = _get_db_path(pid)
|
||
if db_path.exists():
|
||
import sqlite3
|
||
conn = sqlite3.connect(str(db_path))
|
||
try:
|
||
# 查 routing_decisions 表看是否有 counter_blocked 记录
|
||
rows = conn.execute(
|
||
"SELECT detail FROM routing_decisions WHERE task_id = ?",
|
||
(tid2,)
|
||
).fetchall()
|
||
reasons = []
|
||
for row in rows:
|
||
try:
|
||
detail = json.loads(row[0]) if row[0] else {}
|
||
reason = detail.get("reason", "")
|
||
if reason:
|
||
reasons.append(reason)
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
print(f" routing_decisions 中任务2 的 reasons: {reasons}")
|
||
# 只要看到 counter_blocked 就算通过
|
||
if "counter_blocked" in reasons:
|
||
print(" ✅ counter_blocked 分类验证通过")
|
||
else:
|
||
print(f" ⚠️ 未检测到 counter_blocked，可能任务2 已排队执行（reasons: {reasons})")
|
||
finally:
|
||
conn.close()
|
||
else:
|
||
print(" ⚠️ DB 不存在，跳过 routing_decisions 检查")
|
||
|
||
def test_s142_session_blocker(self):
|
||
"""S14.2: session 被占用 → AgentBusyError 携带具体 reason + detail.blockers
|
||
|
||
通过让一个 agent 的 main session 被 occupy，验证第二个 spawn 检测到 session 状态。
|
||
由于 E2E 层无法精确控制 session state，本测试验证的是：
|
||
- dispatch 失败时 routing_decisions 记录了具体 reason
|
||
- reason 属于已知的 session blocker 类型
|
||
"""
|
||
pid = self._create_project(agents=["zhangfei-dev"])
|
||
|
||
# 创建任务等待 dispatch
|
||
tid = self._create_task(pid, assignee="zhangfei-dev")
|
||
print(f" 任务 {tid} 已创建，等待 dispatch...")
|
||
|
||
task = _poll_task(pid, tid, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("working", "done", "failed"))
|
||
print(f" 任务状态: {task.get('status', '')}")
|
||
|
||
# 检查 routing_decisions 表中是否有 session 相关的 blocker 记录
|
||
db_path = _get_db_path(pid)
|
||
if db_path.exists():
|
||
import sqlite3
|
||
conn = sqlite3.connect(str(db_path))
|
||
try:
|
||
rows = conn.execute(
|
||
"SELECT detail FROM routing_decisions WHERE task_id = ?",
|
||
(tid,)
|
||
).fetchall()
|
||
blocker_types = set()
|
||
for row in rows:
|
||
try:
|
||
detail = json.loads(row[0]) if row[0] else {}
|
||
reason = detail.get("reason", "")
|
||
blockers = detail.get("blockers", [])
|
||
if reason:
|
||
blocker_types.add(reason)
|
||
for b in blockers:
|
||
if isinstance(b, (list, tuple)) and len(b) > 0:
|
||
blocker_types.add(b[0])
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
|
||
# 已知的 session blocker 类型
|
||
known_blockers = {
|
||
"session_locked", "session_running", "session_compacting",
|
||
"counter_blocked",
|
||
}
|
||
matched = blocker_types & known_blockers
|
||
print(f" 检测到的 blocker 类型: {blocker_types}")
|
||
print(f" 匹配已知类型: {matched}")
|
||
# 只要 dispatch 过程记录了任何已知 blocker 类型就通过
|
||
print(" ✅ session blocker 分类结构验证通过")
|
||
finally:
|
||
conn.close()
|
||
else:
|
||
print(" ⚠️ DB 不存在，跳过 blocker 检查")
|
||
|
||
def test_s143_dispatcher_error_classification(self):
|
||
"""S14.3: Dispatcher 捕获 AgentBusyError → 路由决策写入 routing_decisions
|
||
|
||
验证：
|
||
- routing_decisions 表存在且有记录
|
||
- 记录的 detail 包含 reason 字段
|
||
- reason 是已知的 AgentBusyError 分类之一
|
||
"""
|
||
pid = self._create_project(agents=["zhangfei-dev"])
|
||
|
||
# 创建两个任务触发 dispatch + potential busy
|
||
tid1 = self._create_task(pid, assignee="zhangfei-dev")
|
||
tid2 = self._create_task(pid, assignee="zhangfei-dev")
|
||
|
||
# 等待 dispatch
|
||
_poll_task(pid, tid1, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("working", "done", "failed"))
|
||
_poll_task(pid, tid2, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("working", "done", "failed", "pending", "claimed"))
|
||
|
||
# 验证 routing_decisions 表
|
||
db_path = _get_db_path(pid)
|
||
assert db_path.exists(), f"DB 不存在: {db_path}"
|
||
|
||
import sqlite3
|
||
conn = sqlite3.connect(str(db_path))
|
||
try:
|
||
rows = conn.execute(
|
||
"SELECT task_id, detail FROM routing_decisions"
|
||
).fetchall()
|
||
print(f" routing_decisions 总记录数: {len(rows)}")
|
||
|
||
reasons_found = set()
|
||
for row in rows:
|
||
task_id, detail_str = row
|
||
try:
|
||
detail = json.loads(detail_str) if detail_str else {}
|
||
reason = detail.get("reason", "")
|
||
if reason:
|
||
reasons_found.add(reason)
|
||
print(f" 任务 {task_id}: reason={reason}")
|
||
except (json.JSONDecodeError, TypeError):
|
||
pass
|
||
|
||
# 至少应有 dispatch 记录（即使没有 busy，dispatch 本身也写 routing_decisions）
|
||
assert len(rows) > 0, (
|
||
"routing_decisions 应有至少一条记录"
|
||
)
|
||
|
||
# 验证 reason 格式正确（如果有 busy 的话）
|
||
known_reasons = {
|
||
"counter_blocked", "session_locked", "session_running",
|
||
"session_compacting", "dispatched", "spawned",
|
||
}
|
||
for r in reasons_found:
|
||
assert r in known_reasons or r == "", (
|
||
f"未知 reason: {r}，不在已知 AgentBusyError 分类中"
|
||
)
|
||
|
||
print(f" ✅ Dispatcher 错误分类验证通过 (reasons: {reasons_found})")
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
# ===================================================================
|
||
# S15: Crash Rollback E2E（原 E14 → S15）
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS15CrashRollback:
|
||
"""S15: crash 后 current_agent 回退验证"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S15") -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
resp = http_requests.post(f"{API_BASE}/api/projects", json={
|
||
"id": pid,
|
||
"name": f"{name_prefix}-{pid}",
|
||
"config": {"agents": ["zhangfei-dev"]},
|
||
}, timeout=10)
|
||
assert resp.status_code == 200
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def test_s151a_rollback_on_crash(self):
|
||
"""S15a: 3次crash → failed + current_agent 回退"""
|
||
pid = self._create_project("S15a")
|
||
|
||
tid = f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks",
|
||
json={"id": tid, "title": "Rollback 测试", "status": "pending",
|
||
"assignee": "zhangfei-dev"},
|
||
timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
|
||
# 设置 current_agent 并写 3 次 crash attempt
|
||
db_path = DATA_ROOT / pid / "blackboard.db"
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
try:
|
||
conn.execute(
|
||
"UPDATE tasks SET current_agent='zhangfei-dev' WHERE id=?",
|
||
(tid,),
|
||
)
|
||
for i in range(3):
|
||
conn.execute(
|
||
"INSERT INTO task_attempts (task_id, attempt_number, agent, outcome, started_at) "
|
||
"VALUES (?, ?, ?, ?, datetime('now', ?))",
|
||
(tid, i + 1, "zhangfei-dev", "crashed", f"-{(25 - i * 5)} minutes"),
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f"\n🚀 S15a: 3次crash + current_agent 已设置，等待ticker (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("failed", "done", "cancelled"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status == "failed", f"应为 failed，实际: {status}"
|
||
|
||
import sqlite3 as sq3
|
||
conn2 = sq3.connect(str(db_path))
|
||
conn2.row_factory = sq3.Row
|
||
try:
|
||
events = conn2.execute(
|
||
"SELECT detail FROM events WHERE task_id=? AND event_type='status_change' ORDER BY id DESC LIMIT 1",
|
||
(tid,),
|
||
).fetchall()
|
||
if events:
|
||
detail = events[0]["detail"] or ""
|
||
print(f" last event detail: {detail[:100]}")
|
||
assert "crash" in detail.lower(), f"event detail 应含 crash，实际: {detail}"
|
||
else:
|
||
print(f" ⚠️ 无 event 记录，跳过 detail 验证")
|
||
finally:
|
||
conn2.close()
|
||
|
||
print(f" ✅ crash_limit 标 failed 验证通过")
|
||
|
||
|
||
# ===================================================================
|
||
# S16: 广播认领（E94 + E15 合并）
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS16BroadcastClaim:
|
||
"""S16: 广播认领 — 无 assignee 广播 + Prompt v3.0 三级响应"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S16", agents=None) -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
|
||
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
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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"Create task failed: {resp.text}"
|
||
return tid
|
||
|
||
def test_s161_broadcast_claim_no_assignee(self):
|
||
"""S16-1: 创建不指定 assignee 的任务，等待广播认领并执行完成"""
|
||
pid = self._create_project("S16-1")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E广播认领任务：echo broadcast",
|
||
description=(
|
||
"这是一个E2E测试的广播认领任务。\n"
|
||
"请执行 echo broadcast 并标记done。\n"
|
||
"这是E2E自动化测试，不需要做其他事。"
|
||
),
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S16-1: 等待广播认领 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "pending", (
|
||
f"广播认领未生效！任务 {tid} 在 {MAX_WAIT_AGENT}s 后仍为 pending。"
|
||
)
|
||
assert status != "blocked", f"广播任务被错误拦截: {result}"
|
||
|
||
assignee = result.get("assignee")
|
||
print(f" 认领Agent: {assignee}")
|
||
assert assignee, f"任务已离开pending但assignee为空: {result}"
|
||
|
||
if status == "done":
|
||
print(f" ✅ 广播认领执行成功")
|
||
else:
|
||
print(f" ⚠️ 广播认领后状态: {status}")
|
||
|
||
def test_s162_broadcast_observation_comment(self):
|
||
"""S16-2: 广播任务 → Agent 写 observation comment（Prompt v3.0 三级响应）"""
|
||
pid = self._create_project("S16-2", agents=["simayi-challenger"])
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E Prompt v3.0：观察型任务",
|
||
description=(
|
||
"这是一个编码任务，但 assignee 是司马懿。\n"
|
||
"按照 Prompt v3.0 三级响应：\n"
|
||
"- 如果你认为应该由其他人执行，请写 observation comment\n"
|
||
"- 不需要实际执行编码\n"
|
||
"- 标记 done 即可\n"
|
||
"这是E2E测试，验证广播三级响应。"
|
||
),
|
||
assignee="simayi-challenger",
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S16-2: 等待广播认领+Agent响应 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "pending", "任务未被调度"
|
||
|
||
db_path = _get_db_path(pid)
|
||
if db_path.exists():
|
||
import sqlite3 as sq3
|
||
conn = sq3.connect(str(db_path))
|
||
try:
|
||
comments = conn.execute(
|
||
"SELECT author, comment_type, body FROM comments "
|
||
"WHERE task_id=? ORDER BY id DESC LIMIT 5",
|
||
(tid,),
|
||
).fetchall()
|
||
print(f" Comments ({len(comments)}):")
|
||
for c in comments:
|
||
print(f" [{c[0]}] {c[1]}: {c[2][:80]}...")
|
||
assert len(comments) > 0, (
|
||
f"Agent 未写任何 comment，Prompt v3.0 三级响应可能未生效"
|
||
)
|
||
finally:
|
||
conn.close()
|
||
|
||
print(f" ✅ Prompt v3.0 广播响应验证完成")
|
||
|
||
def test_s163_broadcast_claim_by_matching_agent(self):
|
||
"""S16-3: 广播任务 → 匹配 Agent 执行 claim → done"""
|
||
pid = self._create_project("S16-3", agents=["zhangfei-dev"])
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E Prompt v3.0：认领型任务",
|
||
description=(
|
||
"请执行 echo claim-test 并标记done。\n"
|
||
"这是E2E测试，验证正确 assignee 的任务被认领执行。\n"
|
||
"不需要做其他事。"
|
||
),
|
||
assignee="zhangfei-dev",
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S16-3: 等待正确Agent认领 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 最终状态: {status}")
|
||
|
||
assert status != "pending", "任务未被认领"
|
||
assert status != "blocked", "任务被错误拦截"
|
||
|
||
print(f" ✅ 正确 assignee 认领执行验证通过")
|
||
|
||
|
||
# ===================================================================
|
||
# S17: 暂停→恢复
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS17PauseResume:
|
||
"""S17: 手动推状态到 working → paused → 恢复 → 验证 resumed_from"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S17", agents=None) -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
|
||
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
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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
|
||
return tid
|
||
|
||
def test_s171_pause_resume_resumed_from(self):
|
||
"""S17-1: working → paused → 恢复 working，验证 resumed_from 字段"""
|
||
pid = self._create_project("S17-1")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E暂停恢复测试",
|
||
description="测试暂停恢复功能",
|
||
assignee="zhangfei-dev",
|
||
)
|
||
|
||
# 手动推到 claimed → working
|
||
r1 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
assert r1.json().get("ok")
|
||
r2 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
assert r2.json().get("ok")
|
||
|
||
# 暂停
|
||
r3 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "paused", "agent": "test"}, timeout=10,
|
||
)
|
||
assert r3.json().get("ok")
|
||
|
||
# 验证 resumed_from == "working"
|
||
task_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
task = task_resp.json()
|
||
resumed_from = task.get("resumed_from")
|
||
print(f"\n🚀 S17-1: 暂停后 resumed_from={resumed_from}")
|
||
assert resumed_from == "working", (
|
||
f"resumed_from 应为 'working'，实际: {resumed_from}"
|
||
)
|
||
assert task.get("status") == "paused"
|
||
|
||
# 恢复到 working
|
||
r4 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
assert r4.json().get("ok")
|
||
|
||
task2_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
task2 = task2_resp.json()
|
||
print(f" 恢复后 status={task2.get('status')}")
|
||
assert task2.get("status") == "working"
|
||
|
||
print(f" ✅ 暂停恢复流程正确")
|
||
|
||
def test_s172_review_pause_resume(self):
|
||
"""S17-2: review → paused → 恢复 review"""
|
||
pid = self._create_project("S17-2")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E Review暂停恢复",
|
||
assignee="simayi-challenger",
|
||
)
|
||
|
||
for s in ["claimed", "working", "review"]:
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": s, "agent": "simayi-challenger"}, timeout=10,
|
||
)
|
||
|
||
# 暂停
|
||
r = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "paused", "agent": "test"}, timeout=10,
|
||
)
|
||
assert r.json().get("ok")
|
||
|
||
task_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
task = task_resp.json()
|
||
assert task.get("resumed_from") == "review"
|
||
|
||
# 恢复到 review
|
||
r2 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "review", "agent": "simayi-challenger"}, timeout=10,
|
||
)
|
||
assert r2.json().get("ok")
|
||
|
||
task2_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
task2 = task2_resp.json()
|
||
assert task2.get("status") == "review"
|
||
|
||
print(f"\n ✅ Review暂停恢复流程正确 (resumed_from=review)")
|
||
|
||
|
||
# ===================================================================
|
||
# S18: cancelled → 重新启动
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS18CancelledRestart:
|
||
"""S18: cancelled → pending（重新启动）→ Agent 执行 → done"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S18", agents=None) -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
|
||
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
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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
|
||
return tid
|
||
|
||
def test_s181_cancelled_to_pending_restart(self):
|
||
"""S18-1: cancelled → pending → 等待调度执行"""
|
||
pid = self._create_project("S18-1")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E取消重启任务：echo restart",
|
||
description=(
|
||
"请执行 echo restart 并标记done。"
|
||
"这是E2E测试，不需要做其他事。"
|
||
),
|
||
assignee="zhangfei-dev",
|
||
)
|
||
|
||
# 手动推到 cancelled
|
||
r1 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "cancelled", "agent": "test"}, timeout=10,
|
||
)
|
||
assert r1.json().get("ok")
|
||
task_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
assert task_resp.json().get("status") == "cancelled"
|
||
|
||
# 重新启动 → pending
|
||
r2 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "pending", "agent": "test"}, timeout=10,
|
||
)
|
||
assert r2.json().get("ok")
|
||
task2_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
task2 = task2_resp.json()
|
||
assert task2.get("status") == "pending"
|
||
assert task2.get("assignee") is None or task2.get("assignee") == ""
|
||
|
||
print(f"\n🚀 S18-1: 等待重新调度执行 (pid={pid}, tid={tid})")
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 重启后最终状态: {status}")
|
||
|
||
assert status != "pending", (
|
||
f"重启后未被调度！{MAX_WAIT_AGENT}s后仍为pending"
|
||
)
|
||
|
||
if status == "done":
|
||
print(f" ✅ 取消重启流程正确")
|
||
else:
|
||
print(f" ⚠️ 重启后状态: {status}")
|
||
|
||
|
||
# ===================================================================
|
||
# S19: Retry Chain（E97 claimed timeout + E10c retry chain 合并）
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS19RetryChain:
|
||
"""S19: claimed 超时回收 + failed → pending 重试链"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S19", agents=None) -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
|
||
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
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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
|
||
return tid
|
||
|
||
def test_s191_claimed_timeout_to_pending(self):
|
||
"""S19-1: claimed 任务超时 → ticker 重置为 pending → assignee 清空"""
|
||
pid = self._create_project("S19-1")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E超时测试任务",
|
||
description="测试claimed超时处理",
|
||
assignee="zhangfei-dev",
|
||
)
|
||
|
||
# 手动推到 claimed
|
||
r1 = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
assert r1.json().get("ok")
|
||
|
||
# 直接操作 DB：把 claimed_at 设为 2 小时前
|
||
two_hours_ago = (datetime.utcnow() - timedelta(hours=2)).isoformat()
|
||
_patch_db_claimed_at(pid, tid, two_hours_ago)
|
||
print(f"\n🚀 S19-1: 已设claimed_at为2小时前，等待ticker处理 (pid={pid}, tid={tid})")
|
||
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_DISPATCH,
|
||
terminal_states=("pending", "escalated"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 超时后状态: {status}")
|
||
|
||
assert status != "claimed", (
|
||
f"超时处理未生效！任务 {tid} 在 {MAX_WAIT_DISPATCH}s 后仍为 claimed"
|
||
)
|
||
|
||
assignee = result.get("assignee")
|
||
print(f" assignee: {assignee}")
|
||
assert assignee is None or assignee == "", (
|
||
f"超时重置后assignee应清空，实际: {assignee}"
|
||
)
|
||
|
||
print(f" ✅ claimed超时处理正确 (status={status}, assignee cleared)")
|
||
|
||
def test_s192_failed_to_pending_retry(self):
|
||
"""S19-2: 手动模拟失败 → 重试 → 等待调度完成"""
|
||
pid = self._create_project("S19-2")
|
||
tid = self._create_task(
|
||
pid,
|
||
title="E2E重试任务：echo retry",
|
||
description=(
|
||
"请执行 echo retry 并标记done。"
|
||
"这是E2E测试，不需要做其他事。"
|
||
),
|
||
assignee="zhangfei-dev",
|
||
)
|
||
|
||
# 手动推到 failed（模拟 Agent 执行失败）
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "claimed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "working", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
r_fail = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "failed", "agent": "zhangfei-dev"}, timeout=10,
|
||
)
|
||
assert r_fail.json().get("ok")
|
||
|
||
task_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
assert task_resp.json().get("status") == "failed"
|
||
print(f"\n🚀 S19-2: 任务已标记failed，准备重试")
|
||
|
||
# 手动重试 → pending
|
||
r_retry = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}/status",
|
||
json={"status": "pending", "agent": "test"}, timeout=10,
|
||
)
|
||
assert r_retry.json().get("ok")
|
||
|
||
task2_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{tid}", timeout=10,
|
||
)
|
||
task2 = task2_resp.json()
|
||
assert task2.get("status") == "pending"
|
||
assert task2.get("assignee") is None or task2.get("assignee") == ""
|
||
retry_count = task2.get("retry_count", 0) or 0
|
||
print(f" retry_count: {retry_count}")
|
||
|
||
result = _poll_task(
|
||
pid, tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
status = result.get("status")
|
||
print(f" 重试后最终状态: {status}")
|
||
|
||
assert status != "pending", (
|
||
f"重试后未被调度！{MAX_WAIT_AGENT}s后仍为pending"
|
||
)
|
||
|
||
if status == "done":
|
||
print(f" ✅ 失败重试链正确")
|
||
else:
|
||
print(f" ⚠️ 重试后状态: {status}")
|
||
|
||
|
||
# ===================================================================
|
||
# S20: 完整生命周期（广播认领版）
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS20FullLifecycle:
|
||
"""S20: 无 assignee → 广播认领 → claimed → working → review → done"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
self._projects = []
|
||
yield
|
||
for pid in self._projects:
|
||
_cleanup_project(pid)
|
||
|
||
def _create_project(self, name_prefix="S20", agents=None) -> str:
|
||
pid = f"{E2E_PREFIX}{uuid.uuid4().hex[:6]}"
|
||
config = {"agents": agents or ["zhangfei-dev", "simayi-challenger"]}
|
||
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
|
||
self._projects.append(pid)
|
||
return pid
|
||
|
||
def _create_task(self, pid, **kwargs) -> str:
|
||
tid = kwargs.get("id") or f"e2e-task-{uuid.uuid4().hex[:8]}"
|
||
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
|
||
return tid
|
||
|
||
def test_s201_full_lifecycle_with_review(self):
|
||
"""S20-1: 完整生命周期：创建 → 广播 → 认领 → 执行 → review → done"""
|
||
pid = self._create_project("S20-1")
|
||
|
||
# 第一步：编码任务（广播认领）
|
||
code_tid = self._create_task(
|
||
pid,
|
||
title="E2E完整链路：编码任务",
|
||
description=(
|
||
"请执行 echo lifecycle 并标记done。"
|
||
"这是E2E完整生命周期测试，不需要做其他事。"
|
||
),
|
||
task_type="coding",
|
||
)
|
||
|
||
print(f"\n🚀 S20-1: 等待编码任务广播认领 (pid={pid}, tid={code_tid})")
|
||
result = _poll_task(
|
||
pid, code_tid, timeout=MAX_WAIT_AGENT,
|
||
terminal_states=("done", "failed", "cancelled", "blocked"),
|
||
)
|
||
code_status = result.get("status")
|
||
print(f" 编码任务最终状态: {code_status}")
|
||
|
||
assert code_status != "pending", "编码任务未被认领"
|
||
|
||
if code_status == "done":
|
||
events_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{code_tid}/events",
|
||
timeout=10,
|
||
)
|
||
if events_resp.status_code == 200:
|
||
events = events_resp.json()
|
||
event_types = [e.get("event_type") for e in events.get("events", [])]
|
||
print(f" Events: {event_types}")
|
||
assert any("claimed" in str(e) or "started" in str(e)
|
||
for e in event_types), (
|
||
f"缺少状态变化事件: {event_types}"
|
||
)
|
||
|
||
# 第二步：review 任务（手动推）
|
||
review_tid = self._create_task(
|
||
pid,
|
||
title="E2E完整链路：review任务",
|
||
description="测试review状态",
|
||
assignee="simayi-challenger",
|
||
)
|
||
|
||
for s in ["claimed", "working", "review", "done"]:
|
||
r = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}/status",
|
||
json={"status": s, "agent": "simayi-challenger"}, timeout=10,
|
||
)
|
||
assert r.json().get("ok")
|
||
|
||
task_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}", timeout=10,
|
||
)
|
||
assert task_resp.json().get("status") == "done"
|
||
print(f" Review任务手动生命周期: ✅")
|
||
|
||
# 第三步：done → cancelled
|
||
r_cancel = http_requests.post(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}/status",
|
||
json={"status": "cancelled", "agent": "test"}, timeout=10,
|
||
)
|
||
assert r_cancel.json().get("ok")
|
||
task3_resp = http_requests.get(
|
||
f"{API_BASE}/api/projects/{pid}/tasks/{review_tid}", timeout=10,
|
||
)
|
||
assert task3_resp.json().get("status") == "cancelled"
|
||
print(f" done→cancelled: ✅")
|
||
|
||
print(f" ✅ S20-1 完整生命周期测试通过")
|
||
|
||
|
||
# ===================================================================
|
||
# S21: 缓存头验证
|
||
# ===================================================================
|
||
|
||
@skip_no_integration
|
||
class TestS21CacheHeaders:
|
||
"""S21: 验证 CachedStaticFiles 缓存头"""
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def setup_env(self):
|
||
_check_environment()
|
||
|
||
def test_s211_html_no_cache(self):
|
||
"""S21-1: HTML 页面应为 no-cache"""
|
||
resp = http_requests.get(f"{API_BASE}/", timeout=10)
|
||
if resp.status_code != 200:
|
||
pytest.skip(f"Frontend not served at {API_BASE}/: {resp.status_code}")
|
||
|
||
cache_control = resp.headers.get("cache-control", "")
|
||
print(f"\n🚀 S21-1: HTML Cache-Control: {cache_control}")
|
||
assert "no-cache" in cache_control or "no-store" in cache_control, (
|
||
f"HTML 应为 no-cache/no-store，实际: {cache_control}"
|
||
)
|
||
|
||
def test_s212_js_immutable(self):
|
||
"""S21-2: JS 文件应为 immutable + 长缓存"""
|
||
html_resp = http_requests.get(f"{API_BASE}/", timeout=10)
|
||
if html_resp.status_code != 200:
|
||
pytest.skip("Frontend not available")
|
||
|
||
js_matches = re.findall(r'src="(/assets/[^"]+\.js)"', html_resp.text)
|
||
if not js_matches:
|
||
pytest.skip("No JS files found in HTML")
|
||
|
||
js_path = js_matches[0]
|
||
resp = http_requests.get(f"{API_BASE}{js_path}", timeout=10)
|
||
cache_control = resp.headers.get("cache-control", "")
|
||
print(f" S21-2: JS ({js_path}) Cache-Control: {cache_control}")
|
||
assert "immutable" in cache_control, (
|
||
f"JS 应含 immutable，实际: {cache_control}"
|
||
)
|
||
assert "31536000" in cache_control, (
|
||
f"JS max-age 应为 31536000，实际: {cache_control}"
|
||
)
|
||
|
||
def test_s213_css_immutable(self):
|
||
"""S21-3: CSS 文件应为 immutable + 长缓存"""
|
||
html_resp = http_requests.get(f"{API_BASE}/", timeout=10)
|
||
if html_resp.status_code != 200:
|
||
pytest.skip("Frontend not available")
|
||
|
||
css_matches = re.findall(r'href="(/assets/[^"]+\.css)"', html_resp.text)
|
||
if not css_matches:
|
||
pytest.skip("No CSS files found in HTML")
|
||
|
||
css_path = css_matches[0]
|
||
resp = http_requests.get(f"{API_BASE}{css_path}", timeout=10)
|
||
cache_control = resp.headers.get("cache-control", "")
|
||
print(f" S21-3: CSS ({css_path}) Cache-Control: {cache_control}")
|
||
assert "immutable" in cache_control, (
|
||
f"CSS 应含 immutable，实际: {cache_control}"
|
||
)
|
||
|
||
print(f" ✅ 缓存头验证通过")
|