[moz] docs: §18 API 聚合重构 + 工具链 Tab 设计文档
CI / lint (pull_request) Successful in 7s
CI / test (pull_request) Successful in 28s
CI / notify-on-failure (pull_request) Successful in 1s

- 18-api-refactor-and-toolchain-tab.md: 主设计(9章+实施约束)
  - 后端拆分方案 B(task_routes + task_relation_routes + shared)
  - expand 细粒度聚合(comments/events 带 limit+total_count)
  - 任务列表搜索参数 q
  - 工具链 Tab 设计(仿 MailPanel + 搜索栏)
  - GATE 门控 + 委派原则 + wiki 查询规则
  - 司马懿已审(mail-1781415763066)
- 18-test-design.md: 测试用例详细设计(34 个用例 + CI 集成)
- tests/scripts/verify_api_compat.sh: 路由兼容性验证脚本
This commit is contained in:
cfdaily
2026-06-14 13:53:56 +08:00
parent 923971ad92
commit f55a037c98
3 changed files with 1141 additions and 0 deletions
@@ -0,0 +1,587 @@
# API 聚合重构 + 工具链 Tab 设计
> **编号**: §18
> **状态**: 设计中
> **日期**: 2026-06-14
> **作者**: 庞统(副军师)🐦
> **审查**: 司马懿(mail-1781415763066 已回复,方案 B 调整版确认)
---
## 1. 背景与目标
### 1.1 问题
1. **blackboard_routes.py 膨胀**572 行、22 个路由,task/comment/output/review/event/decision/observation/archive 全堆一个文件,维护困难
2. **前端 N+1 请求**:打开 TaskModal 需要 5 次独立请求(task + events + subtasks + progress + comments),影响前端性能
3. **工具链事件无前端展示**`_toolchain` DB 隔离已完成,但前端无对应 Tab,工具链事件只能通过 Agent 收 Mail 感知
### 1.2 目标
1. 按领域拆分 blackboard_routes.py → 3 个文件
2. 实现细粒度 expand 聚合接口,前端 1-2 次请求拿全任务详情
3. 新增工具链 Tab(列表 + 详情 + 搜索栏)
4. 任务列表支持标题搜索
### 1.3 不做
- checkpoint_routes.py 不纳入拆分(已独立)
- mail_routes / toolchain_routes / project_routes 不动
- SQL JOIN / batch query 性能优化(当前 SQLite 单写下多次查询可接受)
---
## 2. 后端 API 文件拆分
### 2.1 拆分方案(方案 B 调整版,司马懿确认)
| 新文件 | 内容 | 预估行数 |
|--------|------|---------|
| `task_routes.py` | task CRUD + create(含 AI 标题) + patch + progress + claim + status(含广播) + archive + archive-done | ~280 |
| `task_relation_routes.py` | comments + outputs(含文件写入) + reviews + decisions + observations + events + experiences + summary | ~250 |
| `shared.py` | `_bb()` / `_q()` / `_validate_project()` / `_task_to_dict()` / `_init_agent_ids()` / `_extract_mentions()` / 常量导入 | ~30 |
### 2.2 路由分配明细
**task_routes.py**10 个路由):
| 路由 | 方法 | 函数 | 说明 |
|------|------|------|------|
| `/tasks` | GET | `list_tasks` | 列表(新增 `q` 搜索参数) |
| `/tasks` | POST | `create_task` | 创建(含 `_generate_title` |
| `/tasks/{tid}` | GET | `get_task` | 详情(含 expand 聚合) |
| `/tasks/{tid}` | PATCH | `patch_task` | 更新 |
| `/tasks/{tid}/progress` | GET | `task_progress` | 进度 |
| `/tasks/{tid}/claim` | POST | `claim_task` | 认领 |
| `/tasks/{tid}/status` | POST | `update_status` | 状态流转(含广播逻辑) |
| `/tasks/{tid}/archive` | POST | `archive_task` | 归档 |
| `/tasks/archive-done` | POST | `archive_done_tasks` | 批量归档 |
**task_relation_routes.py**13 个路由):
| 路由 | 方法 | 函数 | 说明 |
|------|------|------|------|
| `/tasks/{tid}/comments` | GET | `get_comments` | 评论列表 |
| `/tasks/{tid}/comments` | POST | `add_comment` | 添加评论(含 @mention 提取) |
| `/tasks/{tid}/outputs` | GET | `get_outputs` | 产出列表 |
| `/tasks/{tid}/outputs` | POST | `write_output` | 写入产出(含文件写入逻辑) |
| `/tasks/{tid}/decisions` | GET | `get_decisions` | 决策列表 |
| `/tasks/{tid}/decisions` | POST | `add_decision` | 添加决策 |
| `/tasks/{tid}/observations` | POST | `add_observation` | 添加观察 |
| `/tasks/{tid}/reviews` | GET | `get_reviews` | 审查列表 |
| `/tasks/{tid}/reviews` | POST | `add_review` | 添加审查 |
| `/tasks/{tid}/events` | GET | `get_task_events` | 事件列表 |
| `/tasks/{tid}/experiences` | GET | `get_task_experiences` | 经验列表 |
| `/events` | GET | `get_events` | 项目级事件 |
| `/summary` | GET | `task_summary` | 任务汇总 |
### 2.3 shared.py 共享件
从 blackboard_routes.py 提取到 shared.py
| 符号 | 类型 | 说明 |
|------|------|------|
| `_validate_project()` | function | 项目 ID 校验 |
| `_bb()` | function | Blackboard 实例获取 |
| `_q()` | function | Queries 实例获取 |
| `_task_to_dict()` | function | Task → dict 序列化 |
| `_init_agent_ids()` | function | Agent ID 初始化 |
| `_extract_mentions()` | function | @mention 提取 |
| `VALID_STATUSES` | import | 从 db.py 重导出 |
| `OUTPUT_TYPES` | import | 从 db.py 重导出 |
### 2.4 main.py 路由注册变更
```python
# 拆分前
from src.api.blackboard_routes import router as blackboard_router
app.include_router(blackboard_router)
# 拆分后
from src.api.task_routes import router as task_router
from src.api.task_relation_routes import router as task_relation_router
app.include_router(task_router)
app.include_router(task_relation_router)
```
URL prefix 不变:所有路由仍是 `/api/projects/{pid}/...`,前端 URL 零改动。
### 2.5 向后兼容
- 删除 `blackboard_routes.py`,所有引用指向新文件
- `expand=all` 保持兼容(内部映射为全量 expand)
- 不改变任何 API 的请求/响应格式(仅文件组织变化)
---
## 3. expand 聚合接口
### 3.1 设计
`GET /api/projects/{pid}/tasks/{tid}?expand=comments,outputs,reviews,events,decisions`
支持逗号分隔的细粒度选择,替代当前的 `expand=all`
### 3.2 返回格式
```json
{
"task": { "id": "...", "title": "...", "status": "working", ... },
"comments": {
"items": [...],
"total_count": 45,
"limit": 20
},
"events": {
"items": [...],
"total_count": 120,
"limit": 30
},
"outputs": [...],
"reviews": [...],
"decisions": [...]
}
```
### 3.3 limit 策略
| 关联资源 | expand 返回 | 分页支持 | 理由 |
|----------|------------|---------|------|
| comments | 最新 20 条 + total_count | `GET /comments?limit=50&offset=0` | 高频资源,长任务可能积累几十条 |
| events | 最新 30 条 + total_count | `GET /events?limit=100&offset=0` | 运行几天可能上百条 |
| outputs | 全部 | 不需要 | 通常 <5 条 |
| reviews | 全部 | 不需要 | 通常 <5 条 |
| decisions | 全部 | 不需要 | 通常 <5 条 |
前端拿到 `total_count > items.length` 时显示"还有 N 条",按需点击加载。
### 3.4 实现伪码
```python
@router.get("/tasks/{task_id}")
async def get_task(project_id: str, task_id: str,
expand: Optional[str] = None):
bb = _bb(project_id)
task = bb.get_task(task_id)
if not task:
raise HTTPException(404, f"Task not found: {task_id}")
result = _task_to_dict(task)
if not expand:
return result
expand_list = expand.split(",") if expand != "all" else [
"comments", "outputs", "reviews", "events", "decisions"
]
q = _q(project_id)
if "comments" in expand_list:
all_comments = bb.get_comments(task_id)
result["comments"] = {
"items": [dict(c.__dict__) for c in all_comments[-20:]],
"total_count": len(all_comments),
"limit": 20,
}
if "events" in expand_list:
all_events = q.task_events(task_id)
result["events"] = {
"items": all_events[-30:],
"total_count": len(all_events),
"limit": 30,
}
if "outputs" in expand_list:
result["outputs"] = [dict(o.__dict__) for o in bb.get_outputs(task_id)]
if "reviews" in expand_list:
result["reviews"] = [dict(r.__dict__) for r in bb.get_reviews(task_id)]
if "decisions" in expand_list:
result["decisions"] = [dict(d.__dict__) for d in bb.get_decisions(task_id)]
return result
```
### 3.5 性能分析
| 场景 | 当前(无 expand | expand 后 | 改善 |
|------|-----------------|-----------|------|
| 打开 TaskModal | 5 次 HTTP 请求 | 2 次(task+expand + subtasks | -60% 请求 |
| 单次 expand 响应体 | — | ~5-15KB(典型) | 一次大请求 < 五次小请求 |
| DB 查询次数 | 5 次(各端点独立查) | 5 次(expand 内部循环) | 相同,暂不优化 |
---
## 4. 任务列表搜索
### 4.1 设计
`GET /api/projects/{pid}/tasks?q=关键词`
在现有 `list_tasks` 基础上增加 `q` 查询参数,支持标题模糊搜索(SQL LIKE)。
### 4.2 实现
```python
@router.get("/tasks")
async def list_tasks(project_id: str, q: Optional[str] = None, ...):
bb = _bb(project_id)
tasks = bb.list_tasks(status=status, ...)
if q:
q_lower = q.lower()
tasks = [t for t in tasks if q_lower in (t.title or "").lower()]
return {"tasks": [_task_to_dict(t) for t in tasks]}
```
**设计决策**:过滤在 Python 层做而非 SQL 层。
- 理由:当前 `list_tasks` 已在 Python 层做 status 筛选,加一层 title 过滤一致性更好
- 如果后续任务量大(>1000),再改为 SQL LIKE 查询
---
## 5. 前端:工具链 Tab
### 5.1 Tab 定义
```typescript
// store.ts TabKey 新增
| 'toolchain'
// TAB_DEFS 新增(插在 settings 前面)
{ key: 'toolchain', label: '工具链', icon: '⛓️' },
```
### 5.2 数据加载
```typescript
// store.ts 新增
toolchainTasks: any[];
loadToolchain: async () => {
const res = await fetch('/api/projects/_toolchain/tasks');
const data = await res.json();
set({ toolchainTasks: data.tasks || [] });
}
// Tab 切换时加载
if (tab === 'toolchain') s.loadToolchain();
```
### 5.3 ToolchainPanel 组件
仿 MailPanel 结构,三个区域:
**搜索栏**(顶部):
- 文本输入框,输入关键词实时过滤列表
- 调用 `GET /api/projects/_toolchain/tasks?q=关键词`
**列表区**(左侧):
- 工具链事件列表(时间倒序)
- 每条显示:标题 + 时间 + 状态标签
- 点击选中,高亮当前选中项
**详情区**(右侧):
- 选中事件的完整内容
- 调用 `GET /api/projects/_toolchain/tasks/{tid}?expand=comments` 获取详情
- 展示:标题、描述、状态、评论(action_report 等)
### 5.4 和 Mail 的隔离
| 维度 | Mail Tab | 工具链 Tab |
|------|---------|-----------|
| 数据源 | `_mail` 项目 | `_toolchain` 项目 |
| 事件类型 | Agent 间通信(inform/request | 系统事件(CI/PR/部署/Review |
| 搜索 | 无(邮件量不大) | 有(工具链事件频率高) |
---
## 6. 测试设计
### 6.1 后端 API 拆分测试
**目标**:验证拆分后所有路由 URL 不变、行为不变。
**测试文件**`tests/integration/test_api.py`(扩展现有)+ 新增 `tests/unit/test_task_routes.py`
| 测试类 | 测试用例 | 验证点 |
|--------|---------|--------|
| TestTaskRoutes | test_list_tasks | GET /tasks 返回格式不变 |
| | test_list_tasks_with_search | q 参数过滤正确 |
| | test_list_tasks_empty_q | q 为空时返回全部 |
| | test_get_task | GET /tasks/{tid} 基本详情 |
| | test_get_task_expand_comments | expand=comments 返回带 total_count + limit |
| | test_get_task_expand_events | expand=events 返回带 total_count + limit |
| | test_get_task_expand_outputs | expand=outputs 全量返回 |
| | test_get_task_expand_multiple | expand=comments,outputs,reviews 组合 |
| | test_get_task_expand_all | expand=all 向后兼容 |
| | test_get_task_no_expand | 不传 expand 返回基本 task |
| | test_create_task | POST 格式不变 |
| | test_claim_task | 认领行为不变 |
| | test_update_status | 状态流转不变 |
| | test_patch_task | PATCH 不变 |
| | test_archive_task | 归档不变 |
| 测试类 | 测试用例 | 验证点 |
|--------|---------|--------|
| TestTaskRelationRoutes | test_comments_crud | GET/POST comments 不变 |
| | test_outputs_crud | GET/POST outputs 不变 |
| | test_write_output_file | 文件写入逻辑不变 |
| | test_reviews_crud | GET/POST reviews 不变 |
| | test_decisions_crud | GET/POST decisions 不变 |
| | test_observations_add | POST observations 不变 |
| | test_events_list | GET events 不变 |
| | test_experiences_list | GET experiences 不变 |
| | test_project_events | GET /events 不变 |
| | test_summary | GET /summary 不变 |
**兼容性验证脚本**
```bash
#!/bin/bash
# tests/scripts/verify_api_compat.sh
# 对比拆分前后所有路由 URL 和方法,确保零变化
echo "=== 拆分前路由清单 ==="
# 从 git stash 或 main 分支提取
git stash
python -c "
from src.main import app
for route in app.routes:
if hasattr(route, 'methods') and hasattr(route, 'path'):
for m in sorted(route.methods):
if m in ('GET','POST','PATCH','DELETE','PUT'):
print(f'{m} {route.path}')
" | sort > /tmp/routes_before.txt
git stash pop
echo "=== 拆分后路由清单 ==="
python -c "
from src.main import app
for route in app.routes:
if hasattr(route, 'methods') and hasattr(route, 'path'):
for m in sorted(route.methods):
if m in ('GET','POST','PATCH','DELETE','PUT'):
print(f'{m} {route.path}')
" | sort > /tmp/routes_after.txt
echo "=== Diff ==="
diff /tmp/routes_before.txt /tmp/routes_after.txt
if [ $? -eq 0 ]; then
echo "✅ 路由完全一致"
else
echo "❌ 路由有差异"
exit 1
fi
```
### 6.2 expand 聚合测试
**测试文件**`tests/integration/test_api.py` 新增 `TestExpandAPI`
```python
class TestExpandAPI:
"""expand 聚合接口测试"""
def test_expand_comments_limit(self, client, project_env):
"""comments 返回 limit 20 + total_count"""
# 创建 1 个 task + 25 条 comment
bb = Blackboard(...)
task = bb.create_task(Task(id="t1", ...))
for i in range(25):
bb.add_comment("t1", agent="a1", comment_type="general",
content=f"comment {i}")
resp = client.get(f"/api/projects/test-proj/tasks/t1?expand=comments")
data = resp.json()
assert len(data["comments"]["items"]) == 20 # limit
assert data["comments"]["total_count"] == 25
assert data["comments"]["limit"] == 20
def test_expand_events_limit(self, client, project_env):
"""events 返回 limit 30 + total_count"""
# 创建 1 个 task + 35 条 event
...
def test_expand_outputs_full(self, client, project_env):
"""outputs 全量返回(不限制)"""
...
def test_expand_multiple(self, client, project_env):
"""expand=comments,outputs 组合"""
...
def test_expand_all_compat(self, client, project_env):
"""expand=all 向后兼容(返回全部,格式是 list 不是 {items, total_count}"""
...
def test_no_expand(self, client, project_env):
"""不传 expand 只返回基本 task"""
...
def test_expand_invalid_field_ignored(self, client, project_env):
"""expand=invalid_field 不报错,忽略无效字段"""
...
```
### 6.3 搜索测试
```python
class TestTaskSearch:
def test_search_by_title(self, client, project_env):
"""q 参数标题模糊搜索"""
bb = Blackboard(...)
bb.create_task(Task(id="t1", title="[moz] bug: Mail API 500"))
bb.create_task(Task(id="t2", title="[moz] feat: new dashboard"))
resp = client.get("/api/projects/test-proj/tasks?q=Mail")
data = resp.json()
assert len(data["tasks"]) == 1
assert "Mail" in data["tasks"][0]["title"]
def test_search_case_insensitive(self, client, project_env):
"""大小写不敏感"""
...
def test_search_empty_q(self, client, project_env):
"""q 为空返回全部"""
...
def test_search_no_match(self, client, project_env):
"""无匹配返回空列表"""
...
```
### 6.4 前端测试(手动验证)
| 验证点 | 操作 | 预期 |
|--------|------|------|
| 工具链 Tab 出现 | 打开前端 | Tab 栏有 ⛓️ 工具链 |
| 列表加载 | 点击工具链 Tab | 显示 _toolchain 事件列表 |
| 搜索过滤 | 输入关键词 | 列表实时过滤 |
| 详情展示 | 点击某条事件 | 右侧/弹窗显示完整内容 |
| Tab 切换不丢数据 | 切到其他 Tab 再回来 | 数据保持 |
### 6.5 CI 集成
新增测试运行命令:
```bash
# API 拆分兼容性(CI 必跑)
bash tests/scripts/verify_api_compat.sh
# 新增单元 + 集成测试
pytest tests/unit/test_task_routes.py tests/integration/test_api.py -m "not e2e" -v
```
---
## 7. 实施约束
### 7.1 GATE 门控铁律
本任务为 L3(5+ 文件、跨后端重构 + 前端新增),必须遵守:
1. **需求不清不动手** — 列出假设让用户确认
2. **方案未定不实现** — 先出方案等确认(本文档即为方案产出)
3. **评估过影响范围才动手** — 见 §8 风险评估
4. **使用 plan-act-verify skill** — GATE → PLAN → ACT → VERIFY 全流程
L1 小改动(单文件 <50 行,做错代价低)可跳过 GATE。
### 7.2 委派原则
1. **main session 负责**:理解意图、澄清需求、方案决策、结果审查
2. **需求明确后的所有执行工作 → 通过 subagent-delegation skill 委派**,先阅读 skill 全文再使用
3. **委派时 task 描述必须自包含**:目标、输入、输出路径、验收标准、关键上下文
4. **sub 完成后审查产出质量**,发现问题让 sub 修,不自己动手
5. 保持 main session context 清晰,避免 compact 丢信息
### 7.3 wiki 查询规则
- 做方案前先查 wiki-vault,有 1% 相关就要查
- 路径:`/Volumes/KnowledgeBase/wiki-vault/`
- 检索顺序:index.md → grep 关键词 → summary 字段 → 按需读全文
- 查不到在 `_meta/knowledge-gaps.md` 记录
- 本设计已查询(无 FastAPI 路由拆分相关实践,已记录 gap)
---
## 8. 实施计划
### Phase 1: 后端 API 拆分(不含功能变更)
| 步骤 | 内容 | 验证 |
|------|------|------|
| 1.1 | 创建 `shared.py`,提取共享 helper | import 无报错 |
| 1.2 | 创建 `task_routes.py`,迁移 10 个路由 | 路由注册成功 |
| 1.3 | 创建 `task_relation_routes.py`,迁移 13 个路由 | 路由注册成功 |
| 1.4 | 更新 `main.py` router 注册 | app 启动无报错 |
| 1.5 | 删除 `blackboard_routes.py` | — |
| 1.6 | 运行 `verify_api_compat.sh` | 路由清单 diff = 0 |
| 1.7 | 运行现有测试 | 全量通过 |
### Phase 2: expand 聚合 + 搜索
| 步骤 | 内容 | 验证 |
|------|------|------|
| 2.1 | 重写 `get_task` expand 逻辑(细粒度) | TestExpandAPI 通过 |
| 2.2 | `list_tasks``q` 参数 | TestTaskSearch 通过 |
| 2.3 | 新增测试用例 | 覆盖率达标 |
### Phase 3: 前端工具链 Tab
| 步骤 | 内容 | 验证 |
|------|------|------|
| 3.1 | store.ts 新增 toolchain 数据加载 | — |
| 3.2 | api.ts 新增 expand 调用封装 | — |
| 3.3 | 创建 `ToolchainPanel.tsx` | 组件渲染正常 |
| 3.4 | App.tsx 注册新 Tab | Tab 显示正确 |
| 3.5 | TaskModal 改用 expand 减少 | 请求次数减少 |
### Phase 4: 联调 + 评审
| 步骤 | 内容 |
|------|------|
| 4.1 | 全量测试 `pytest -m "not e2e"` |
| 4.2 | 发评审给司马懿 |
| 4.3 | 前端手动验证 |
---
## 9. 风险评估
| 风险 | 级别 | 缓解 |
|------|------|------|
| 拆分后 import 路径断裂 | 中 | IDE 全局搜索 + 运行时验证 |
| expand 返回体过大 | 低 | comments/events 有 limit |
| 工具链事件量大影响前端 | 低 | 搜索栏 + 分页 |
| expand=all 向后兼容 | 低 | 单独兼容分支处理 |
---
## 10. 评审记录
### 司马懿 mail-17814157630662026-06-14
| 项目 | 结论 |
|------|------|
| 文件拆分 | 方案 B 调整版(task_routes + task_relation_routes + shared |
| expand | 细粒度,events/comments 带 limit+total_count |
| 性能 | 当前 SQLite 多次查询可接受 |
| checkpoint | 不纳入 |
| _generate_title | 留在 task_routes.py |
| write_output | 注意不是简单 CRUD |
---
## 11. 变更记录
| 日期 | 版本 | 内容 |
|------|------|------|
| 2026-06-14 | v1.0 | 初版设计 |
+484
View File
@@ -0,0 +1,484 @@
# §18 测试用例详细设计
> **关联**: `docs/design/18-api-refactor-and-toolchain-tab.md`
> **日期**: 2026-06-14
---
## 1. 测试文件规划
| 文件 | 类型 | 测试数 | 说明 |
|------|------|--------|------|
| `tests/integration/test_api.py` | 集成 | 扩展现有 | 拆分后回归验证 |
| `tests/unit/test_task_routes.py` | 单元 | 14 | task_routes 专项 |
| `tests/unit/test_expand_api.py` | 单元 | 7 | expand 聚合专项 |
| `tests/unit/test_task_search.py` | 单元 | 4 | 搜索专项 |
| `tests/scripts/verify_api_compat.sh` | 脚本 | 1 | CI 路由兼容性 |
**总计**:26 个测试 + 1 个 CI 脚本
---
## 2. 预置 Fixture
```python
# tests/conftest.py 新增(如果不存在则补充)
@pytest.fixture
def expand_env(tmp_path):
"""expand 测试环境:1 个 task + 预置关联数据"""
project_root = tmp_path / "projects"
project_root.mkdir()
os.environ["BLACKBOARD_ROOT"] = str(project_root)
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="Expand Test Task", task_type="coding"))
# 预置 25 条 comment
for i in range(25):
bb.add_comment("t1", agent="agent1", comment_type="general",
content=f"Comment number {i}")
# 预置 5 条 output
for i in range(5):
bb.write_output("t1", agent="agent1",
output_type="code",
content=f"output content {i}",
filename=f"file_{i}.py")
# 预置 3 条 review
for i in range(3):
bb.add_review("t1", reviewer="agent1",
verdict="APPROVE",
confidence=0.9,
risk_level="low",
summary=f"Review {i}")
# 预置 2 条 decision
for i in range(2):
bb.add_decision("t1", agent="agent1",
decision_type="scope",
rationale=f"Decision {i}")
# 预置 35 条 event
from src.blackboard.queries import Queries
q = Queries(project_root / "test-proj" / "blackboard.db")
for i in range(35):
q.add_event("t1", event_type="status_change",
detail=f"Event {i}")
yield project_root
del os.environ["BLACKBOARD_ROOT"]
```
---
## 3. task_routes.py 测试(test_task_routes.py
```python
"""task_routes.py 路由测试 — 验证拆分后行为不变"""
import pytest
from fastapi.testclient import TestClient
from src.main import app
client = TestClient(app)
pytestmark = pytest.mark.integration
class TestTaskListRoutes:
"""GET /tasks + 搜索"""
def test_list_tasks_basic(self, project_env):
"""列表基本返回格式不变"""
resp = client.get("/api/projects/test-proj/tasks")
assert resp.status_code == 200
data = resp.json()
assert "tasks" in data
assert isinstance(data["tasks"], list)
def test_list_tasks_with_search(self, project_env):
"""q 参数搜索标题"""
resp = client.get("/api/projects/test-proj/tasks?q=Existing")
data = resp.json()
assert len(data["tasks"]) == 1
assert "Existing" in data["tasks"][0]["title"]
def test_list_tasks_search_case_insensitive(self, project_env):
"""大小写不敏感"""
resp = client.get("/api/projects/test-proj/tasks?q=existing")
data = resp.json()
assert len(data["tasks"]) == 1
def test_list_tasks_search_no_match(self, project_env):
"""无匹配返回空列表"""
resp = client.get("/api/projects/test-proj/tasks?q=nonexistent_xyz")
data = resp.json()
assert len(data["tasks"]) == 0
def test_list_tasks_search_empty_q(self, project_env):
"""q 为空返回全部"""
resp = client.get("/api/projects/test-proj/tasks?q=")
data = resp.json()
assert len(data["tasks"]) >= 1
class TestTaskDetailRoutes:
"""GET /tasks/{tid} + expand"""
def test_get_task_basic(self, project_env):
"""无 expand 返回基本 task"""
resp = client.get("/api/projects/test-proj/tasks/t1")
assert resp.status_code == 200
data = resp.json()
assert data["id"] == "t1"
assert "comments" not in data # 无 expand 不含关联数据
def test_get_task_404(self, project_env):
"""不存在的 task 返回 404"""
resp = client.get("/api/projects/test-proj/tasks/nonexistent")
assert resp.status_code == 404
class TestTaskActionRoutes:
"""claim/status/patch/archive 行为不变"""
def test_claim_task(self, 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, project_env):
"""状态流转不变"""
client.post("/api/projects/test-proj/tasks/t1/claim",
json={"agent": "agent1"})
resp = client.post("/api/projects/test-proj/tasks/t1/status",
json={"agent": "agent1", "status": "working"})
assert resp.status_code == 200
def test_invalid_status_transition(self, project_env):
"""非法状态转换返回 409"""
resp = client.post("/api/projects/test-proj/tasks/t1/status",
json={"agent": "agent1", "status": "done"})
assert resp.status_code == 409
def test_patch_task(self, project_env):
"""PATCH 更新不变"""
resp = client.patch("/api/projects/test-proj/tasks/t1",
json={"priority": 5})
assert resp.status_code == 200
def test_archive_task(self, project_env):
"""归档不变"""
resp = client.post("/api/projects/test-proj/tasks/t1/archive",
json={"agent": "agent1"})
assert resp.status_code == 200
class TestTaskCreateRoute:
"""POST /tasks 创建行为不变"""
def test_create_task(self, project_env):
"""创建格式不变"""
resp = client.post("/api/projects/test-proj/tasks",
json={"title": "New Task", "description": "test"})
assert resp.status_code == 201
data = resp.json()
assert "id" in data
```
---
## 4. expand 聚合测试(test_expand_api.py
```python
"""expand 聚合接口测试"""
import pytest
from fastapi.testclient import TestClient
from src.main import app
client = TestClient(app)
pytestmark = pytest.mark.integration
class TestExpandComments:
"""expand=comments"""
def test_comments_limit_and_count(self, expand_env):
"""返回最新 20 条 + total_count=25"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=comments")
data = resp.json()
comments = data["comments"]
assert isinstance(comments, dict)
assert len(comments["items"]) == 20
assert comments["total_count"] == 25
assert comments["limit"] == 20
def test_comments_are_latest(self, expand_env):
"""返回的是最新 20 条(Comment 5-24"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=comments")
data = resp.json()
first_content = data["comments"]["items"][0]["content"]
last_content = data["comments"]["items"][-1]["content"]
# 最新 20 条 = index 5 到 24
assert "5" in first_content or "24" in last_content
class TestExpandEvents:
"""expand=events"""
def test_events_limit_and_count(self, expand_env):
"""返回最新 30 条 + total_count=35"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=events")
data = resp.json()
events = data["events"]
assert isinstance(events, dict)
assert len(events["items"]) == 30
assert events["total_count"] == 35
assert events["limit"] == 30
class TestExpandFullResources:
"""outputs/reviews/decisions 全量返回"""
def test_expand_outputs_full(self, expand_env):
"""outputs 全量返回(5 条),格式是 list 不是 dict"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=outputs")
data = resp.json()
outputs = data["outputs"]
assert isinstance(outputs, list)
assert len(outputs) == 5
def test_expand_reviews_full(self, expand_env):
"""reviews 全量返回(3 条)"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=reviews")
data = resp.json()
reviews = data["reviews"]
assert isinstance(reviews, list)
assert len(reviews) == 3
def test_expand_decisions_full(self, expand_env):
"""decisions 全量返回(2 条)"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=decisions")
data = resp.json()
decisions = data["decisions"]
assert isinstance(decisions, list)
assert len(decisions) == 2
class TestExpandCombinations:
"""组合 expand"""
def test_expand_multiple_fields(self, expand_env):
"""expand=comments,outputs,reviews 组合"""
resp = client.get(
"/api/projects/test-proj/tasks/t1?expand=comments,outputs,reviews"
)
data = resp.json()
assert "comments" in data
assert "outputs" in data
assert "reviews" in data
assert "events" not in data # 未请求
assert "decisions" not in data
def test_expand_all_compat(self, expand_env):
"""expand=all 向后兼容"""
resp = client.get("/api/projects/test-proj/tasks/t1?expand=all")
data = resp.json()
# all 返回所有关联资源
assert "comments" in data
assert "outputs" in data
assert "reviews" in data
assert "events" in data
assert "decisions" in data
def test_no_expand(self, expand_env):
"""不传 expand 只返回基本 task"""
resp = client.get("/api/projects/test-proj/tasks/t1")
data = resp.json()
assert "comments" not in data
assert "outputs" not in data
assert data["id"] == "t1"
def test_expand_invalid_field_ignored(self, expand_env):
"""无效 expand 字段静默忽略"""
resp = client.get(
"/api/projects/test-proj/tasks/t1?expand=comments,invalid_field"
)
data = resp.json()
assert "comments" in data
assert "invalid_field" not in data
```
---
## 5. task_relation_routes.py 回归测试
```python
"""task_relation_routes.py 路由回归 — 验证拆分后行为不变"""
import pytest
from fastapi.testclient import TestClient
from src.main import app
client = TestClient(app)
pytestmark = pytest.mark.integration
class TestRelationRoutesRegression:
"""关联路由回归测试"""
def test_comments_crud(self, project_env):
"""GET/POST comments 不变"""
# POST
resp = client.post("/api/projects/test-proj/tasks/t1/comments",
json={"agent": "a1", "comment_type": "general",
"content": "test comment"})
assert resp.status_code == 201
# GET
resp = client.get("/api/projects/test-proj/tasks/t1/comments")
assert resp.status_code == 200
assert len(resp.json()["comments"]) >= 1
def test_outputs_crud(self, project_env):
"""GET/POST outputs 不变"""
resp = client.post("/api/projects/test-proj/tasks/t1/outputs",
json={"agent": "a1", "type": "code",
"content": "print('hello')"})
assert resp.status_code == 201
resp = client.get("/api/projects/test-proj/tasks/t1/outputs")
assert resp.status_code == 200
def test_write_output_with_filename(self, project_env):
"""output 含 filename 的文件写入不变"""
resp = client.post("/api/projects/test-proj/tasks/t1/outputs",
json={"agent": "a1", "type": "code",
"content": "x = 1",
"filename": "test.py"})
assert resp.status_code == 201
def test_write_output_invalid_type(self, project_env):
"""output 无效 type 返回 422"""
resp = client.post("/api/projects/test-proj/tasks/t1/outputs",
json={"agent": "a1", "type": "invalid_type",
"content": "x"})
assert resp.status_code == 422
def test_reviews_crud(self, project_env):
"""GET/POST reviews 不变"""
resp = client.post("/api/projects/test-proj/tasks/t1/reviews",
json={"reviewer": "a1", "verdict": "APPROVE",
"confidence": 0.9, "risk_level": "low",
"summary": "LGTM"})
assert resp.status_code == 201
resp = client.get("/api/projects/test-proj/tasks/t1/reviews")
assert resp.status_code == 200
def test_decisions_crud(self, project_env):
"""GET/POST decisions 不变"""
resp = client.post("/api/projects/test-proj/tasks/t1/decisions",
json={"agent": "a1", "decision_type": "scope",
"rationale": "test"})
assert resp.status_code == 201
resp = client.get("/api/projects/test-proj/tasks/t1/decisions")
assert resp.status_code == 200
def test_observations_add(self, project_env):
"""POST observations 不变"""
resp = client.post("/api/projects/test-proj/tasks/t1/observations",
json={"agent": "a1", "observation_type": "note",
"content": "observed"})
assert resp.status_code == 201
def test_events_list(self, project_env):
"""GET events 不变"""
resp = client.get("/api/projects/test-proj/tasks/t1/events")
assert resp.status_code == 200
assert "events" in resp.json()
def test_project_events(self, project_env):
"""GET /events 项目级事件不变"""
resp = client.get("/api/projects/test-proj/events")
assert resp.status_code == 200
def test_summary(self, project_env):
"""GET /summary 不变"""
resp = client.get("/api/projects/test-proj/summary")
assert resp.status_code == 200
```
---
## 6. CI 集成
### .gitea/workflows/ci.yml 新增步骤
```yaml
test:
runs-on: ubuntu-latest
steps:
# ... 现有步骤 ...
# API 兼容性验证
- name: Verify API Compatibility
run: |
cd src/frontend && npm run build
bash tests/scripts/verify_api_compat.sh
# 新增测试
- name: Run API Tests
run: |
pytest tests/unit/test_task_routes.py \
tests/unit/test_expand_api.py \
tests/integration/test_api.py \
-m "not e2e" -v
```
### 本地开发验证流程
```bash
# 1. 改完代码后先跑兼容性验证
bash tests/scripts/verify_api_compat.sh
# 2. 跑全量测试
pytest -m "not e2e" -v
# 3. 跑新增专项测试
pytest tests/unit/test_task_routes.py tests/unit/test_expand_api.py -v
```
---
## 7. 测试覆盖矩阵
| 设计文档章节 | 测试文件 | 测试类 | 用例数 |
|-------------|---------|--------|--------|
| §2 路由拆分(兼容性) | verify_api_compat.sh | — | 1 |
| §3.1 基本详情 | test_task_routes.py | TestTaskDetailRoutes | 2 |
| §3.2 搜索 | test_task_routes.py | TestTaskListRoutes | 5 |
| §3.3 动作路由 | test_task_routes.py | TestTaskActionRoutes | 5 |
| §3.4 创建 | test_task_routes.py | TestTaskCreateRoute | 1 |
| §4 expand comments | test_expand_api.py | TestExpandComments | 2 |
| §4 expand events | test_expand_api.py | TestExpandEvents | 1 |
| §4 expand 全量 | test_expand_api.py | TestExpandFullResources | 3 |
| §4 expand 组合 | test_expand_api.py | TestExpandCombinations | 4 |
| §5.2 关联回归 | test_api.py | TestRelationRoutesRegression | 10 |
| **合计** | | | **34** |
+70
View File
@@ -0,0 +1,70 @@
#!/bin/bash
# verify_api_compat.sh — 验证 API 拆分前后路由清单完全一致
#
# 用法:
# bash tests/scripts/verify_api_compat.sh
#
# 前置:
# - 当前在开发目录(sanguo_moziplus_v2/
# - git working tree 有拆分改动
# - main 分支是拆分前的基准
#
# 输出:
# ✅ 路由完全一致(exit 0)
# ❌ 路由有差异(exit 1,打印 diff)
set -euo pipefail
BEFORE_FILE="/tmp/routes_before_$$.txt"
AFTER_FILE="/tmp/routes_after_$$.txt"
echo "=== 提取拆分前路由清单(main 分支)==="
# stash 当前改动(如果有 untracked 新文件,--include-untracked
STASHED=0
if ! git diff --quiet || ! git diff --cached --quiet; then
git stash --include-untracked
STASHED=1
fi
python3 -c "
from src.main import app
for route in app.routes:
if hasattr(route, 'methods') and hasattr(route, 'path'):
for m in sorted(route.methods):
if m in ('GET','POST','PATCH','DELETE','PUT'):
print(f'{m} {route.path}')
" | sort > "$BEFORE_FILE"
echo "Routes before: $(wc -l < "$BEFORE_FILE")"
# 恢复改动
if [ "$STASHED" = "1" ]; then
git stash pop
fi
echo "=== 提取拆分后路由清单(当前 working tree==="
python3 -c "
from src.main import app
for route in app.routes:
if hasattr(route, 'methods') and hasattr(route, 'path'):
for m in sorted(route.methods):
if m in ('GET','POST','PATCH','DELETE','PUT'):
print(f'{m} {route.path}')
" | sort > "$AFTER_FILE"
echo "Routes after: $(wc -l < "$AFTER_FILE")"
echo ""
echo "=== Diff ==="
if diff "$BEFORE_FILE" "$AFTER_FILE"; then
echo "✅ 路由完全一致"
rm -f "$BEFORE_FILE" "$AFTER_FILE"
exit 0
else
echo "❌ 路由有差异"
rm -f "$BEFORE_FILE" "$AFTER_FILE"
exit 1
fi