Files
effekt-interface/.plan/1ZFcjkLpHmrHbluVoOtA4.md
2026-05-11 14:29:16 +08:00

2052 lines
55 KiB
Markdown
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# pytest + Jenkins 自动化执行接入后端详细方案MVC / 可直接交付 AI 编码)
## 1. 目标与边界
### 1.1 目标
基于当前项目 `D:\zhyy\effekt-interface` 的现有 MVC 结构,为“计划中的自动化用例执行”设计一套可直接落地的后端方案,支持:
1. 单条功能用例触发自动化执行。
2. 测试计划维度批量触发自动化执行。
3. 自动化项目基于 `pytest` 执行指定 `case_id` 对应的用例。
4. Jenkins 作为统一调度入口。
5. 平台记录执行任务、执行明细、状态流转、Jenkins 构建信息、回调结果。
6. 方案需适配当前项目的 MVC 分层:
- controller: `app/api/controller/*.py`
- service: `app/api/service/*.py`
- dao: `app/api/dao/*.py`
- model: `app/api/model/*.py`
- route: `app/api/views.py`
### 1.2 已知现状
经代码结构确认:
- 当前项目是 Flask + SQLAlchemy Session 的轻量 MVC。
- 已有测试计划、计划用例、测试用例表:
- `test_plan`
- `plan_case`
- `test_case`
- 当前计划执行 `plan_case_execute` 仅支持手工回填结果,不支持自动化编排。
- 功能用例与自动化用例的桥梁是 `case_id`
- `test_case.is_auto` 已可区分是否自动化。
- 当前 controller/service/dao 的风格是:
- controller 做参数校验和聚合返回
- service 主要转发 dao 或封装少量业务
- dao 负责数据库 CRUD
- 当前项目已有 HTTP 请求封装:`common/getRequest.py -> Request.go(...)`
- 当前有 `common/cronRequest.py` 作为外部系统请求示例,但不适合直接复用为 Jenkins 客户端,建议新增独立 Jenkins 请求封装。
### 1.3 本期边界
只设计后端逻辑,不做前端实现,不写自动化项目代码,但会定义自动化项目与平台的接口契约。
---
## 2. 总体架构设计
推荐采用:
**平台编排 + Jenkins 参数化触发 + pytest 项目按 execution_id 拉取执行清单 + 回调平台写回结果**
### 2.1 责任划分
#### 平台(本项目)
负责:
- 识别单条/计划执行请求
- 校验用例是否可自动化执行
- 创建执行记录
- 调用 Jenkins Job
- 提供执行清单查询接口给 pytest 项目拉取
- 接收执行中/执行完成回调
- 聚合执行结果并更新计划状态
#### Jenkins
负责:
- 接收平台触发请求
- 以参数化方式启动自动化项目
- 记录构建号、日志、报告地址
- 失败兜底回调平台
#### pytest 自动化项目
负责:
- 接收 Jenkins 参数
- 通过 `execution_id` 调平台接口获取待执行 case 列表
-`case_id` 映射执行 pytest 用例
- 回调平台单条结果和最终结果
- 输出 junit/allure/json 等结果
### 2.2 为什么不直接把 case_id 列表塞进 Jenkins 参数
不推荐直接传长列表,理由:
- 计划内用例多时参数长度不可控。
- URL/参数转义复杂。
- 后续扩展环境、标签、重跑等字段时耦合变高。
- 执行审计不清晰。
推荐只传 `execution_id`,由 pytest 项目回平台拉取清单。
---
## 3. 与当前 MVC 框架的集成方式
### 3.1 新增模块建议
在现有 `case / plan / report` 域外,新增一个独立“自动化执行域”:
- controller: `app/api/controller/automationController.py`
- service: `app/api/service/automationService.py`
- dao: `app/api/dao/automationDao.py`
- model: `app/api/model/automationModel.py`
- util/client: `common/jenkinsRequest.py`
### 3.2 为什么独立成 automation 域
不要把自动化执行逻辑继续塞进 `planController.py``caseController.py`,原因:
- 自动化执行有独立生命周期,不只是计划或用例的附属字段。
- 涉及外部系统 Jenkins 调用、结果回写、状态机、执行明细聚合。
- 拆成独立域后controller/service/dao 责任更清晰。
### 3.3 与现有域的关系
#### 与 Case 域的关系
- `test_case.id` 作为桥梁。
- 通过 `test_case.is_auto=1` 判断可自动化。
- 后续如需补充自动化元信息,可扩展 `test_case` 字段或新增自动化映射表。
#### 与 Plan 域的关系
- 批量执行来源于 `plan_case`
- 自动化执行结果需要同步回写 `plan_case.status / actual_result / executed_time / execution_duration`
- 自动化执行完成后,要复用或增强当前 `_update_plan_status` 的逻辑,更新 `test_plan.status`
---
## 4. 数据库表结构设计
本方案采用:
- 尽量复用现有 `test_case / plan_case / test_plan`
- 新增执行主表和执行明细表
- 可选新增执行事件表(如要完整审计)
---
## 5. 现有表改造建议
### 5.1 `test_case` 表改造
当前已有:
- `id`
- `case_key`
- `is_auto`
由于你说明桥梁就是 `case_id`pytest 项目也按 `case_id` 执行,因此**本期不强制新增自动化用例定位字段**。
但建议增加以下可选字段,便于后续扩展:
```sql
ALTER TABLE test_case
ADD COLUMN auto_status SMALLINT DEFAULT 0,
ADD COLUMN auto_updated_time TIMESTAMP,
ADD COLUMN auto_meta JSONB DEFAULT '{}'::jsonb;
```
字段说明:
- `auto_status`: 0未接入 1已接入 2已失效
- `auto_updated_time`: 自动化信息最近更新时间
- `auto_meta`: 预留字段,后续可存 pytest marker、目录、owner 等
> 如果你要严格最小改动,这 3 个字段可以不做。
---
## 6. 新增核心表结构
### 6.1 自动化执行主表 `auto_execution`
用途:记录一次自动化执行任务,支持单条执行和计划执行。
```sql
CREATE TABLE auto_execution (
id BIGSERIAL PRIMARY KEY,
execution_no VARCHAR(64) NOT NULL UNIQUE,
trigger_type SMALLINT NOT NULL,
project_id BIGINT NOT NULL,
plan_id BIGINT,
plan_round_no INTEGER,
source_case_id BIGINT,
env_code VARCHAR(32) NOT NULL,
run_mode SMALLINT DEFAULT 1,
status SMALLINT NOT NULL DEFAULT 0,
jenkins_job_name VARCHAR(128),
jenkins_queue_id BIGINT,
jenkins_build_number BIGINT,
jenkins_build_url VARCHAR(512),
console_url VARCHAR(512),
report_url VARCHAR(512),
total_count INTEGER DEFAULT 0,
pending_count INTEGER DEFAULT 0,
running_count INTEGER DEFAULT 0,
passed_count INTEGER DEFAULT 0,
failed_count INTEGER DEFAULT 0,
blocked_count INTEGER DEFAULT 0,
skipped_count INTEGER DEFAULT 0,
not_found_count INTEGER DEFAULT 0,
trigger_by BIGINT,
trigger_source VARCHAR(32) DEFAULT 'platform',
trigger_message TEXT,
start_time TIMESTAMP,
end_time TIMESTAMP,
duration_seconds INTEGER,
callback_token VARCHAR(128),
ext JSONB DEFAULT '{}'::jsonb,
created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```
字段定义:
- `execution_no`: 平台执行编号,建议格式 `AE202502120001`
- `trigger_type`:
- 1 单条用例执行
- 2 计划批量执行
- `project_id`: 冗余存项目,方便查询
- `plan_id`: 计划执行时必填,单条执行可空
- `plan_round_no`: 若按轮次执行可记录
- `source_case_id`: 单条执行来源 case_id
- `env_code`: 执行环境编码,如 `st/dev/pre`
- `run_mode`:
- 1 串行
- 2 并行
- `status`: 执行主状态,见后文状态机
- `jenkins_job_name / queue_id / build_number / build_url`
- `console_url / report_url`: 用于前端直接跳转
- `*_count`: 执行汇总
- `trigger_by`: 触发用户
- `trigger_message`: 触发失败或备注
- `callback_token`: Jenkins/pytest 回调鉴权
- `ext`: 预留字段,记录 branch、pytest args、report meta 等
索引建议:
```sql
CREATE INDEX idx_auto_execution_plan_id ON auto_execution(plan_id);
CREATE INDEX idx_auto_execution_project_id ON auto_execution(project_id);
CREATE INDEX idx_auto_execution_status ON auto_execution(status);
CREATE INDEX idx_auto_execution_created_time ON auto_execution(created_time DESC);
CREATE INDEX idx_auto_execution_trigger_by ON auto_execution(trigger_by);
```
---
### 6.2 自动化执行明细表 `auto_execution_case`
用途:记录本次执行下每个 `case_id` 的状态和结果。
```sql
CREATE TABLE auto_execution_case (
id BIGSERIAL PRIMARY KEY,
execution_id BIGINT NOT NULL,
plan_case_id BIGINT,
case_id BIGINT NOT NULL,
case_key VARCHAR(64),
case_title VARCHAR(255),
run_order INTEGER DEFAULT 0,
status SMALLINT NOT NULL DEFAULT 0,
pytest_nodeid VARCHAR(512),
result_message TEXT,
error_message TEXT,
stack_trace TEXT,
report_url VARCHAR(512),
duration_seconds INTEGER,
started_time TIMESTAMP,
finished_time TIMESTAMP,
retry_count INTEGER DEFAULT 0,
ext JSONB DEFAULT '{}'::jsonb,
created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```
字段定义:
- `execution_id`: 对应 `auto_execution.id`
- `plan_case_id`: 如果来源于计划执行,则关联 `plan_case.id`
- `case_id`: 功能用例主键,也是自动化桥梁
- `case_key / case_title`: 冗余快照,避免后续名称变化影响历史展示
- `run_order`: 执行顺序
- `status`: 单用例执行状态,见后文状态机
- `pytest_nodeid`: pytest 的 nodeid`tests/test_login.py::test_xxx`
- `result_message`: 简要结果,如断言失败摘要
- `error_message / stack_trace`: 异常信息
- `report_url`: 若支持单用例报告可记录
- `retry_count`: 失败重跑预留
- `ext`: 可存截图、附件、标签等
索引建议:
```sql
CREATE INDEX idx_auto_execution_case_execution_id ON auto_execution_case(execution_id);
CREATE INDEX idx_auto_execution_case_case_id ON auto_execution_case(case_id);
CREATE INDEX idx_auto_execution_case_plan_case_id ON auto_execution_case(plan_case_id);
CREATE INDEX idx_auto_execution_case_status ON auto_execution_case(status);
CREATE UNIQUE INDEX uk_auto_execution_case_unique ON auto_execution_case(execution_id, case_id, plan_case_id);
```
> 如果同一计划允许相同 `case_id` 在不同轮次、多次加入,则唯一索引可调整成 `(execution_id, plan_case_id, case_id)`,其中 `plan_case_id` 允许为空时要注意 PostgreSQL 对 NULL 唯一性的行为。
---
### 6.3 可选事件表 `auto_execution_event`
如果你希望保留每次回调、状态变化、Jenkins 交互轨迹,建议加事件表。
```sql
CREATE TABLE auto_execution_event (
id BIGSERIAL PRIMARY KEY,
execution_id BIGINT NOT NULL,
execution_case_id BIGINT,
event_type VARCHAR(64) NOT NULL,
event_status VARCHAR(32),
payload JSONB DEFAULT '{}'::jsonb,
created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```
事件类型示例:
- `TRIGGER_REQUESTED`
- `JENKINS_QUEUED`
- `JENKINS_STARTED`
- `CASE_STARTED`
- `CASE_FINISHED`
- `EXECUTION_FINISHED`
- `CALLBACK_FAILED`
> 若第一期追求最小化,可先不建此表,把关键日志写到应用日志中。
---
## 7. SQLAlchemy Model 设计(贴合当前项目风格)
建议新增文件:`app/api/model/automationModel.py`
包含:
- `AutoExecution`
- `AutoExecutionCase`
- (可选)`AutoExecutionEvent`
建模风格应与当前 `planModel.py / caseModel.py` 一致:
- 使用 `declarative_base()`
- `Base.to_dict = to_dict`
- 字段命名与数据库下划线保持一致
- `JSONB / TIMESTAMP / SmallInteger / BigInteger`
---
## 8. 状态流转设计
这是后端实现的核心。
### 8.1 主执行状态 `auto_execution.status`
建议枚举:
- `0` 待触发(平台已建单,尚未请求 Jenkins
- `1` 触发中(正在调用 Jenkins
- `2` 排队中Jenkins 已接收,尚未开始构建)
- `3` 执行中Jenkins/pytest 已开始)
- `4` 成功(全部执行完成且失败数=0
- `5` 失败(已结束,但存在失败/异常)
- `6` 已取消
- `7` 触发失败Jenkins API 调用失败)
- `8` 回调异常(任务可能跑完,但平台未完整接收结果)
推荐状态流:
```text
0待触发
-> 1触发中
-> 2排队中
-> 3执行中
-> 4成功
-> 5失败
-> 6已取消
-> 7触发失败
```
补偿状态:
- 任意执行中状态,如果最终结果缺失但监控发现 Jenkins 已完成,可修正到 4/5。
- 回调鉴权失败/数据异常时,可标记 `8 回调异常`
### 8.2 明细状态 `auto_execution_case.status`
建议枚举:
- `0` 待执行
- `1` 执行中
- `2` 通过
- `3` 失败
- `4` 阻塞
- `5` 跳过
- `6` 未找到自动化实现
- `7` 已取消
推荐状态流:
```text
0待执行 -> 1执行中 -> 2通过
-> 3失败
-> 4阻塞
-> 5跳过
-> 6未找到自动化实现
-> 7已取消
```
### 8.3 与现有 `plan_case.status` 的映射关系
当前 `plan_case.status`
- `0` 未开始
- `1` 通过
- `2` 失败
- `3` 阻塞
自动化结果回写时建议映射:
- `auto_execution_case.status=2` -> `plan_case.status=1`
- `auto_execution_case.status=3` -> `plan_case.status=2`
- `auto_execution_case.status=4` -> `plan_case.status=3`
- `auto_execution_case.status in (5,6,7)` -> **不自动覆盖为通过/失败**,建议保留原值或按业务约定:
- 若原来是 `0`,可保持 `0`
- 并把细节记入 `actual_result`
### 8.4 与 `test_plan.status` 的映射关系
当前 `TestPlan.status`
- `0` 草稿
- `1` 进行中
- `2` 已完成
- `3` 已归档
- `4` 已通过
建议复用现有 `_update_plan_status(plan_id)` 逻辑,但将其抽到 `PlanService``AutomationService` 中统一调用。
执行过程中:
- 计划中存在未执行 -> `1 进行中`
- 全部完成,且无失败/阻塞 -> `4 已通过`
- 全部完成,存在失败/阻塞 -> `2 已完成`
- 若原来 `3 已归档`,不自动改动
---
## 9. 接口设计
接口分三类:
1. 前端触发接口
2. 自动化项目拉取执行清单接口
3. Jenkins/pytest 回调接口
以下设计遵循当前项目风格:
- 路由统一注册在 `app/api/views.py`
- `GET` 查列表/详情
- `POST` 创建/更新/触发
- 返回结构统一 `ApiResponse.build_success/build_failure`
---
## 10. 前端触发接口设计
### 10.1 单条用例执行
#### 路由
`POST /api/automation/case/run`
#### 权限
建议新增:`automation:run`
#### 请求体
```json
{
"caseId": 1001,
"envCode": "st",
"runMode": 1,
"jenkinsJobName": "pytest-auto-runner",
"remark": "单条回归验证"
}
```
#### 字段说明
- `caseId`: 必填
- `envCode`: 必填
- `runMode`: 可选,默认 1 串行
- `jenkinsJobName`: 可选,未传则走系统默认配置
- `remark`: 可选
#### 返回
```json
{
"id": 123,
"executionId": 123,
"executionNo": "AE202502120001",
"status": 2
}
```
#### 处理逻辑
1. 校验 `caseId` 是否存在。
2. 校验 `test_case.is_auto == 1`
3. 创建 `auto_execution`
- `trigger_type=1`
- `source_case_id=caseId`
- `project_id``test_case.project_id` 带出
4. 创建一条 `auto_execution_case`
5. 调用 Jenkins。
6. 若调用成功:
- 更新主表 `status=2`
- 记录 `queue_id / job_name`
7. 若调用失败:
- 更新主表 `status=7`
- 返回失败原因
---
### 10.2 计划自动化执行
#### 路由
`POST /api/automation/plan/run`
#### 权限
`automation:run`
#### 请求体
```json
{
"planId": 2001,
"roundNo": 1,
"envCode": "st",
"runMode": 1,
"caseIds": [1001, 1002, 1003],
"jenkinsJobName": "pytest-auto-runner",
"remark": "计划批量回归"
}
```
#### 说明
- `caseIds` 可选:
- 不传或空:执行当前计划下全部自动化用例
- 传值:执行计划中指定子集
- `roundNo` 可选:如果你已有轮次概念,则仅执行该轮次下的 `plan_case`
#### 返回
```json
{
"id": 124,
"executionId": 124,
"executionNo": "AE202502120002",
"totalCount": 15,
"status": 2
}
```
#### 处理逻辑
1. 校验 `planId`
2. 查询 `plan_case + test_case`
- 必须属于该计划
- `test_case.is_auto=1`
- 若传 `caseIds`,则限制子集
- 若传 `roundNo`,则限制轮次
3. 若无可执行自动化用例,直接报错。
4. 创建 `auto_execution`
- `trigger_type=2`
- `plan_id=planId`
- `plan_round_no=roundNo`
5. 批量创建 `auto_execution_case`
6. 调 Jenkins。
7. 更新状态与 Jenkins 信息。
---
### 10.3 自动化执行列表
#### 路由
`GET /api/automation/execution/list`
#### 请求参数
- `projectId`
- `planId`
- `status`
- `triggerType`
- `pageNo`
- `pageSize`
#### 返回字段
- `id`
- `execution_no`
- `trigger_type`
- `plan_id`
- `source_case_id`
- `env_code`
- `status`
- `total_count`
- `passed_count`
- `failed_count`
- `report_url`
- `jenkins_build_number`
- `created_time`
- `start_time`
- `end_time`
---
### 10.4 自动化执行详情
#### 路由
`GET /api/automation/execution/detail`
#### 请求参数
- `executionId`
#### 返回
主表 + 汇总 + Jenkins 信息 + 用例统计。
---
### 10.5 自动化执行明细列表
#### 路由
`GET /api/automation/execution/case/list`
#### 请求参数
- `executionId`
- `status` 可选
- `pageNo`
- `pageSize`
#### 返回
每条 `auto_execution_case` 的详情。
---
## 11. 自动化项目拉取执行清单接口
这是 pytest 项目的核心入口。
### 11.1 获取执行清单
#### 路由
`GET /api/automation/execution/case/pull`
#### 鉴权
建议两层:
1. 正常 token 鉴权(如果自动化项目能拿到平台 token
2. 或使用 `executionId + callbackToken` 鉴权
建议本期采用:
- query: `executionId`
- header: `X-CALLBACK-TOKEN`
#### 请求参数
- `executionId`
#### 请求头
- `X-CALLBACK-TOKEN: xxx`
#### 返回
```json
{
"executionId": 124,
"executionNo": "AE202502120002",
"triggerType": 2,
"projectId": 88,
"planId": 2001,
"envCode": "st",
"runMode": 1,
"items": [
{
"executionCaseId": 501,
"planCaseId": 9001,
"caseId": 1001,
"caseKey": "TC-ZHYY-001",
"caseTitle": "登录成功",
"runOrder": 1
}
]
}
```
#### 处理逻辑
1. 校验 `executionId`
2. 校验回调 token。
3. 查询 `auto_execution``auto_execution_case`
4. 返回待执行清单。
> 如果未来 pytest 项目还需要额外上下文,如环境 URL、测试账号、marker也可以从 `ext` 中一起返回。
---
## 12. 回调接口设计
推荐分 4 个回调点:
1. Jenkins 入队成功
2. Jenkins 开始执行
3. 单条用例结果回写
4. 整体执行完成回写
---
### 12.1 Jenkins 入队回写
#### 路由
`POST /api/automation/execution/queued`
#### 作用
Jenkins 已接受构建,回传 queue/build 基本信息。
#### 请求体
```json
{
"executionId": 124,
"queueId": 5678,
"jobName": "pytest-auto-runner",
"buildNumber": null,
"buildUrl": null
}
```
#### 处理
- 主表状态 `1/2 -> 2`
- 回写 `queue_id / job_name / build_number / build_url`
> 如果平台触发 Jenkins 后拿不到 queueId也可省略这个接口由后续 started 回调补齐。
---
### 12.2 Jenkins/pytest 开始执行回写
#### 路由
`POST /api/automation/execution/start`
#### 请求体
```json
{
"executionId": 124,
"jobName": "pytest-auto-runner",
"buildNumber": 89,
"buildUrl": "http://jenkins/job/pytest-auto-runner/89/",
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console",
"startTime": "2025-02-12 10:00:00"
}
```
#### 处理
- 主表状态更新为 `3 执行中`
- 记录 `buildNumber / buildUrl / consoleUrl / start_time`
---
### 12.3 单条用例结果回写
#### 路由
`POST /api/automation/execution/case/result`
#### 请求体
```json
{
"executionId": 124,
"caseId": 1001,
"executionCaseId": 501,
"status": 2,
"pytestNodeid": "tests/test_login.py::test_login_success",
"resultMessage": "assert success",
"errorMessage": "",
"stackTrace": "",
"durationSeconds": 12,
"reportUrl": "http://allure/case/1001",
"startedTime": "2025-02-12 10:01:00",
"finishedTime": "2025-02-12 10:01:12",
"ext": {
"attachments": []
}
}
```
#### 处理逻辑
1.`executionCaseId` 优先定位明细;若无则退化为 `executionId + caseId`
2. 更新 `auto_execution_case`
3. 若有关联 `plan_case_id`,同步回写 `plan_case`
- `status`
- `actual_result`
- `executed_time`
- `execution_duration`
4. 重新聚合更新 `auto_execution` 汇总字段。
5. 若是计划执行,可调用计划状态更新逻辑。
---
### 12.4 整体执行完成回写
#### 路由
`POST /api/automation/execution/finish`
#### 请求体
```json
{
"executionId": 124,
"status": 4,
"buildNumber": 89,
"buildUrl": "http://jenkins/job/pytest-auto-runner/89/",
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console",
"reportUrl": "http://allure/report/89",
"startTime": "2025-02-12 10:00:00",
"endTime": "2025-02-12 10:08:00",
"durationSeconds": 480,
"summary": {
"total": 15,
"passed": 12,
"failed": 2,
"blocked": 0,
"skipped": 1,
"notFound": 0
},
"message": "执行完成"
}
```
#### 处理逻辑
1. 更新主表 Jenkins 信息与时间。
2. 若请求体有 summary则直接写回否则后台实时聚合明细表得到。
3. 根据汇总决定主状态:
- `failed > 0 or blocked > 0 or notFound > 0` -> `5失败`
- 否则 -> `4成功`
4. 若是计划执行,调用计划状态刷新。
---
### 12.5 失败/取消回调
#### 路由
`POST /api/automation/execution/abort`
#### 请求体
```json
{
"executionId": 124,
"status": 6,
"message": "Jenkins aborted",
"buildNumber": 89,
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console"
}
```
#### 处理
- 主表置 `6 已取消``5 失败`
- 所有仍为 `0/1` 的明细置为 `7 已取消`
- 刷新汇总
---
## 13. Jenkins 参数设计
推荐只使用少量稳定参数。
### 13.1 Jenkins Job 名称
建议固定为一个或少数几个通用参数化任务,例如:
- `pytest-auto-runner`
- 如果分项目,可 `pytest-auto-runner-zhyy`
不建议“一计划一 Job”。
### 13.2 构建参数
```text
EXECUTION_ID
CALLBACK_TOKEN
PLATFORM_BASE_URL
ENV_CODE
RUN_MODE
TRIGGER_TYPE
```
#### 含义
- `EXECUTION_ID`: 平台执行主键
- `CALLBACK_TOKEN`: 回调鉴权 token
- `PLATFORM_BASE_URL`: 平台地址,如 `http://xxx/api`
- `ENV_CODE`: st/dev/pre
- `RUN_MODE`: 1串行 2并行
- `TRIGGER_TYPE`: 1单条 2计划
### 13.3 可选参数
```text
GIT_BRANCH
PYTEST_ARGS
REPORT_BASE_URL
```
存放位置建议:
- 平台配置中心 / 环境变量 / 常量文件
- 不建议写死在 controller
### 13.4 Jenkins Build API 请求方式
推荐后端调用:
```text
POST /job/{jobName}/buildWithParameters
```
鉴权建议:
- Basic Auth + API Token
- 若 Jenkins 开启 crumb 防护,还要先请求 crumb
---
## 14. Jenkins 回调与结果回写逻辑
### 14.1 平台触发 Jenkins 时的处理
`AutomationService.trigger_execution(...)` 需要:
1. 生成 `callback_token`
2. 创建执行单和明细
3. 更新主状态 `0 -> 1`
4. 调 Jenkins API
5. 成功:
- 主状态 `1 -> 2`
- 保存 queue/build 基础信息
6. 失败:
- 主状态 `1 -> 7`
- 保存错误信息到 `trigger_message`
### 14.2 pytest 项目开始时的处理
pytest 启动脚本:
1. 根据 Jenkins 参数拿到 `EXECUTION_ID`
2. 调平台 `/execution/case/pull`
3. 成功拿清单后调用 `/execution/start`
4. 开始逐条执行
### 14.3 pytest 每执行一条用例
执行器需要:
1.`case_id` 定位到 pytest 对应测试项
2. 执行完成后回调 `/execution/case/result`
3. 回调内容至少包括:
- executionCaseId / caseId
- status
- durationSeconds
- nodeid
- errorMessage/resultMessage
### 14.4 pytest 全量执行结束
执行器需要:
1. 汇总总数
2. 计算开始/结束时间
3. 发布 report_url如 Allure
4.`/execution/finish`
### 14.5 Jenkins 失败兜底
即使 pytest 未回调成功Jenkins `post { failure { ... } }` 阶段也应该:
- 至少调用 `/execution/abort``/execution/finish(status=5)`
- 避免平台主单永久卡在 `2/3`
---
## 15. 详细执行流程
### 15.1 单条用例执行流程
```text
前端点击“执行自动化”
-> POST /api/automation/case/run
-> controller 校验参数
-> service 查询 test_case
-> 校验 is_auto=1
-> dao 创建 auto_execution
-> dao 创建 auto_execution_case
-> service 调 Jenkins
-> Jenkins 接受构建
-> 平台主单状态变更为排队中
-> pytest 项目启动
-> 拉取 execution 清单
-> 回调 start
-> 执行 case_id 对应 pytest 用例
-> 回调 case/result
-> 回调 finish
-> 平台更新 execution 汇总
```
### 15.2 计划批量执行流程
```text
前端点击“执行自动化用例”
-> POST /api/automation/plan/run
-> controller 校验参数
-> service 查询 plan_case + test_case
-> 过滤 is_auto=1
-> dao 创建 auto_execution
-> dao 批量创建 auto_execution_case
-> service 调 Jenkins
-> Jenkins 启动 pytest 项目
-> pytest 按 execution_id 拉取清单
-> 回调 execution/start
-> 按顺序或并行执行每个 case_id
-> 每条执行后回调 case/result
-> 平台同步回写 plan_case
-> 全部完成后回调 finish
-> 平台刷新 auto_execution 汇总
-> 平台刷新 test_plan 状态
```
---
## 16. Controller / Service / DAO 详细职责设计
### 16.1 Controller 层
新增 `AutomationController`,建议方法:
- `case_run()`
- `plan_run()`
- `execution_list()`
- `execution_detail()`
- `execution_case_list()`
- `execution_case_pull()`
- `execution_queued()`
- `execution_start()`
- `execution_case_result()`
- `execution_finish()`
- `execution_abort()`
Controller 责任:
- 参数读取 `_get`
- 基础必填校验
- 调 service
- 返回统一结构
不要在 controller 里直接写 SQL 或请求 Jenkins。
---
### 16.2 Service 层
新增 `AutomationService`,建议职责:
#### 触发类
- `create_case_execution(session, req_data, user_id)`
- `create_plan_execution(session, req_data, user_id)`
- `trigger_jenkins(session, execution_id, job_name)`
#### 查询类
- `get_execution_detail(session, execution_id)`
- `list_executions(session, filters, page, size)`
- `list_execution_cases(session, execution_id, filters, page, size)`
- `pull_execution_cases(session, execution_id, callback_token)`
#### 回调类
- `mark_execution_queued(...)`
- `mark_execution_started(...)`
- `save_case_result(...)`
- `finish_execution(...)`
- `abort_execution(...)`
#### 聚合类
- `refresh_execution_summary(session, execution_id)`
- `sync_plan_case_result(session, execution_case)`
- `refresh_plan_status(session, plan_id)`
Service 层应承接主要业务逻辑与状态流转。
---
### 16.3 DAO 层
新增 `AutomationDao`,建议方法:
#### 主表
- `create_execution(session, add_info)`
- `update_execution_by_id(session, execution_id, update_info)`
- `get_execution_by_id(session, execution_id)`
- `list_execution_by_filters(session, filters, page, size)`
#### 明细表
- `batch_create_execution_cases(session, batch_info_list)`
- `get_execution_case_by_id(session, execution_case_id)`
- `get_execution_case_by_unique(session, execution_id, case_id, plan_case_id=None)`
- `update_execution_case_by_id(session, execution_case_id, update_info)`
- `list_execution_case_by_filters(session, filters, page, size)`
#### 聚合
- `count_execution_case_summary(session, execution_id)`
#### 执行清单查询
- `query_plan_auto_cases(session, plan_id, round_no=None, case_ids=None)`
- `query_case_auto_item(session, case_id)`
DAO 保持与当前项目风格一致:
- 独立 CRUD
- `session.done(close=False)`
- 失败返回 `(0, err_msg)` / 查询返回对象
---
## 17. 与现有代码的修改点清单
### 17.1 新增文件
1. `app/api/model/automationModel.py`
2. `app/api/dao/automationDao.py`
3. `app/api/service/automationService.py`
4. `app/api/controller/automationController.py`
5. `common/jenkinsRequest.py`
### 17.2 修改文件
#### `app/api/views.py`
新增路由注册:
- `/automation/case/run`
- `/automation/plan/run`
- `/automation/execution/list`
- `/automation/execution/detail`
- `/automation/execution/case/list`
- `/automation/execution/case/pull`
- `/automation/execution/queued`
- `/automation/execution/start`
- `/automation/execution/case/result`
- `/automation/execution/finish`
- `/automation/execution/abort`
#### `app/api/service/planService.py`
建议增加公共方法:
- `refresh_plan_status(session, plan_id)`
`PlanController._update_plan_status()` 中的逻辑迁移出来,避免 controller 内部藏业务逻辑。
#### `app/api/controller/planController.py`
- `plan_case_execute()` 仍保留手工执行。
- 自动化执行入口不要继续加到这个方法里,而是走新 automation 接口。
#### `const.py`
建议新增 Jenkins 配置:
- `JENKINS_BASE_URL`
- `JENKINS_USER`
- `JENKINS_TOKEN`
- `JENKINS_DEFAULT_JOB`
- `AUTOMATION_CALLBACK_SECRET`
- `PLATFORM_BASE_URL`
> 生产上更建议来自环境变量,但当前项目已有常量直写风格,第一期可先保持一致。
---
## 18. `common/jenkinsRequest.py` 设计建议
参考 `common/getRequest.py` 风格,但 Jenkins 需要更灵活处理:
建议封装:
- `get_crumb()`
- `build_with_parameters(job_name, params)`
- `get_build_info(job_name, build_number)` 可选
### 建议返回内容
触发后尽量返回:
- success
- status_code
- queue_id如果能解析到
- location header
- error_message
如果当前 Jenkins 网关层不方便拿 queueId也可以第一期只判断触发成功与否。
---
## 19. 回写 `plan_case` 的规则细化
每次 `/execution/case/result` 到达后:
如果 `auto_execution_case.plan_case_id` 不为空,则同步:
### 19.1 状态映射
- `2 通过` -> `plan_case.status = 1`
- `3 失败` -> `plan_case.status = 2`
- `4 阻塞` -> `plan_case.status = 3`
- `5 跳过 / 6未找到 / 7取消` -> 默认不改 `plan_case.status`
### 19.2 文本字段
- `plan_case.actual_result`
- 成功:记录 `resultMessage`
- 失败:优先写 `errorMessage`
- `plan_case.executed_time`:写 `finished_time` 或当前时间
- `plan_case.execution_duration`:写 `duration_seconds`
- `defect_links / attachments`:本期不自动写,后续有缺陷系统集成再扩展
### 19.3 幂等性
同一个 `executionCaseId` 多次回调时:
- 允许覆盖 `status / duration / message`
- 但如果主状态已结束4/5/6应限制后续异常回调只做日志记录或按最后一次有效值覆盖避免状态混乱
---
## 20. 聚合汇总逻辑
### 20.1 执行汇总字段计算规则
基于 `auto_execution_case` 统计:
- `total_count = count(*)`
- `pending_count = status=0`
- `running_count = status=1`
- `passed_count = status=2`
- `failed_count = status=3`
- `blocked_count = status=4`
- `skipped_count = status=5`
- `not_found_count = status=6`
### 20.2 主状态自动判定规则
`refresh_execution_summary` 中:
1. 若存在 `running_count > 0` -> 主状态至少为 `3`
2. 若全部结束:
- `failed + blocked + not_found > 0` -> `5`
- 否则 -> `4`
3. 若任务被外部取消 -> `6`
4. 若触发阶段失败 -> `7`
### 20.3 结束时间规则
当全部明细都进入结束态2/3/4/5/6/7
- 自动补 `end_time`
-`start_time` 已有,则计算 `duration_seconds`
---
## 21. 计划执行选取逻辑
计划执行时,从 `plan_case` 选出真正需要跑的明细。
### 21.1 查询条件
- `PlanCase.plan_id = planId`
- 如果传 `roundNo``PlanCase.round_no = roundNo`
- 如果传 `caseIds``PlanCase.case_id in caseIds`
- 关联 `TestCase.is_delete = 0`
- `TestCase.is_auto = 1`
- 如有需要,也可加 `TestCase.status = 1 or 4`(正常/评审通过)
### 21.2 排序规则
默认:
- `plan_case.id asc`
### 21.3 去重规则
如果同一计划下同一轮次允许一个 `case_id` 出现多次:
- 不去重,按 `plan_case.id` 逐条执行
- 这样更符合计划执行场景
如果你明确保证不重复,也可唯一化。
---
## 22. 单条执行选取逻辑
输入 `caseId` 后:
- 直接查询 `test_case`
- 校验 `is_auto=1`
- 创建 1 条 `auto_execution_case`
- `plan_case_id` 为空
这样单条执行与计划执行复用同一套执行主流程。
---
## 23. 幂等与并发控制
### 23.1 重复点击执行按钮
建议在创建任务前校验:
- 是否存在同一 `planId + envCode + status in (0,1,2,3)` 的运行中任务
- 是否存在同一 `caseId + envCode + status in (0,1,2,3)` 的运行中任务
策略建议:
- 默认不允许重复触发,返回“已有执行中任务”
- 如果后续需要“强制重跑”,再加 `force=true`
### 23.2 回调幂等
对于 `/execution/case/result`
- 若同一条记录已是最终状态,再收到相同状态回调,应视为幂等成功
- 若收到不同终态,以最后一次回调覆盖,但写日志
### 23.3 事务一致性
`save_case_result()` 推荐事务内做:
1. 更新 `auto_execution_case`
2. 更新 `plan_case`
3. 提交
4. 再调用聚合更新 `auto_execution`
如果要更强一致性,也可在同一事务完成,但当前项目的 session 风格是轻量提交,可分两步。
---
## 24. 错误场景与兜底策略
### 24.1 用例不存在
- 单条执行:直接返回失败
- 计划执行:忽略已删除用例,仅对有效自动化用例建单;若最终为空则报错
### 24.2 用例未接入自动化
- 单条执行:直接报错“该用例未实现自动化”
- 计划执行:过滤掉 `is_auto=0`;若全部都不是自动化则报错
### 24.3 Jenkins 触发失败
- `auto_execution.status = 7`
- `trigger_message` 存失败原因
### 24.4 pytest 找不到 `case_id` 对应实现
- 回调单条明细 `status = 6`
- 主单最终通常为 `5`
### 24.5 部分用例执行完finish 回调丢失
- Jenkins `post` 阶段兜底调 `/finish``/abort`
- 平台可后续增加定时补偿任务,扫描长时间停留在 `3 执行中` 的记录
### 24.6 平台回调接口鉴权失败
- 返回失败
- 自动化项目应重试
- 平台日志记录 `executionId`
---
## 25. 接口返回码与错误语义建议
当前项目 `const.py` 中错误码较泛,第一期可以沿用现有:
- 40009 新增失败
- 40011 获取失败
- 40012 更新失败
但为了 AI 编码更明确,建议新增语义化错误码:
```python
40020: '自动化执行任务创建失败'
40021: '自动化用例不存在'
40022: '该用例未接入自动化'
40023: '计划下无可执行自动化用例'
40024: 'Jenkins 触发失败'
40025: '执行记录不存在'
40026: '回调鉴权失败'
40027: '执行结果回写失败'
```
> 若你想最小改动,也可先不加新错误码,统一用已有 `40009/40011/40012`。
---
## 26. 权限设计建议
建议新增权限点:
- `automation:run`
- `automation:list`
- `automation:detail`
- `automation:callback`
其中:
- 前端用户接口需要登录 + `automation:run/list/detail`
- Jenkins/pytest 回调接口建议:
- 放到 `should_skip_auth()` 白名单中,绕过用户 token
- 改为内部 token/header 校验
否则自动化项目拿用户 token 不稳定。
---
## 27. 回调鉴权设计
### 27.1 推荐方案
每次创建执行任务时生成 `callback_token`,存 `auto_execution.callback_token`
Jenkins/pytest 回调时传:
```text
X-CALLBACK-TOKEN: {callback_token}
```
平台校验:
- `executionId` 存在
- header token 与库中一致
### 27.2 好处
- 不依赖用户登录 token
- 每次执行隔离
- token 泄露影响范围仅单次任务
---
## 28. pytest 自动化项目对接契约
虽然本期不写自动化项目代码,但后端方案必须给到契约。
### 28.1 自动化项目输入
来自 Jenkins 参数:
- `EXECUTION_ID`
- `CALLBACK_TOKEN`
- `PLATFORM_BASE_URL`
- `ENV_CODE`
- `RUN_MODE`
### 28.2 自动化项目行为
1.`/execution/case/pull`
2.`case_id` 映射到 pytest 测试项
3. 单条执行后调 `/execution/case/result`
4. 全部结束调 `/execution/finish`
### 28.3 `case_id` 到 pytest 的映射方式
你已经确定桥梁是 `case_id`,建议自动化项目采用以下之一:
#### 方式 Apytest marker
```python
@pytest.mark.case_id(1001)
def test_login_success():
...
```
执行器启动时收集 marker建立 `case_id -> nodeid` 映射。
#### 方式 B维护映射文件
`case_map.json`
但从维护性看,**推荐 marker 方式**。
### 28.4 找不到映射时
回调:
- `status=6`
- `resultMessage='case_id 未映射到 pytest 用例'`
---
## 29. MVC 代码落地顺序建议(交给 AI 编码的实施计划)
### Phase 1数据层
1. 新建数据库表:
- `auto_execution`
- `auto_execution_case`
2. 新建 `automationModel.py`
3. 新建 `automationDao.py`
4. 编写:
- 创建主单
- 批量创建明细
- 查询详情/列表
- 聚合汇总
### Phase 2服务层
1. 新建 `automationService.py`
2. 实现:
- 单条执行建单
- 计划执行建单
- Jenkins 触发
- 执行清单拉取
- 单条结果回写
- 整体完成回写
- 汇总刷新
3.`PlanController._update_plan_status()` 逻辑迁移成服务方法
### Phase 3控制器与路由
1. 新建 `automationController.py`
2.`views.py` 注册接口
3. 增加回调接口白名单处理
### Phase 4Jenkins 客户端
1. 新建 `common/jenkinsRequest.py`
2. 封装 Basic Auth + buildWithParameters
3. 在 service 中接入
### Phase 5联调准备
1. 给 pytest 项目提供接口文档
2. 给 Jenkins Pipeline 提供参数清单
3. 约定回调协议
---
## 30. 推荐的 AI 编码拆分任务
为了让 AI 更稳地写代码,建议拆成 5 个子任务:
### 任务 1建模与 DAO
输出:
- `automationModel.py`
- `automationDao.py`
### 任务 2查询与列表接口
输出:
- execution list/detail/case list
### 任务 3单条执行 / 计划执行建单逻辑
输出:
- case run
- plan run
- Jenkins trigger
### 任务 4回调与状态流转
输出:
- pull/start/case_result/finish/abort
- plan_case 回写
- plan 状态更新
### 任务 5Jenkins 客户端与配置收口
输出:
- `common/jenkinsRequest.py`
- `const.py` 新配置
- 回调白名单
---
## 31. 最终推荐实现决策
### 必做
1. 新增 `auto_execution``auto_execution_case`
2. 独立 automation MVC 模块
3. 平台只传 `execution_id` 给 Jenkins
4. pytest 项目按 `execution_id` 拉清单
5. 单条回写 + 整体回写
6. 回写 `plan_case``test_plan` 状态
### 建议做
1. `callback_token` 鉴权
2. 主/明细状态机
3. 幂等处理
4. Jenkins 失败兜底回调
### 可后置
1. 并行执行
2. 失败重跑
3. 定时补偿
4. 执行事件表
5. 自动创建缺陷/附件
---
## 32. 可直接交给 AI 的开发说明摘要
你可以把下面这段直接给 AI 作为实现要求:
1. 基于当前 Flask MVC 结构,新增 automation 域controller/service/dao/model。
2. 新增表 `auto_execution``auto_execution_case`,字段按本方案实现。
3. 提供接口:
- `POST /api/automation/case/run`
- `POST /api/automation/plan/run`
- `GET /api/automation/execution/list`
- `GET /api/automation/execution/detail`
- `GET /api/automation/execution/case/list`
- `GET /api/automation/execution/case/pull`
- `POST /api/automation/execution/queued`
- `POST /api/automation/execution/start`
- `POST /api/automation/execution/case/result`
- `POST /api/automation/execution/finish`
- `POST /api/automation/execution/abort`
4. 单条执行通过 `test_case.id` 建单,计划执行通过 `plan_case + test_case.is_auto=1` 批量建单。
5. 主状态与明细状态按本方案的枚举实现。
6. `/case/result` 回调时同步更新 `plan_case.status / actual_result / executed_time / execution_duration`
7. 每次明细回写后聚合更新 `auto_execution` 的计数和主状态。
8. 计划执行时,执行结果变化后要刷新 `test_plan.status`,把当前 `PlanController._update_plan_status()` 逻辑迁到 service 层复用。
9. 新增 `common/jenkinsRequest.py`,封装 Jenkins `buildWithParameters` 调用。
10. Jenkins 参数至少包含:`EXECUTION_ID``CALLBACK_TOKEN``PLATFORM_BASE_URL``ENV_CODE``RUN_MODE``TRIGGER_TYPE`
11. 回调接口采用 `X-CALLBACK-TOKEN` 鉴权,并加入免登录白名单。
---
## 33. PostgreSQL 建表 SQL可直接落库
> 说明:以下 SQL 仅覆盖本次自动化执行接入新增对象,不包含已有 `test_case / plan_case / test_plan` 表。
> 当前代码已按 PostgreSQL + SQLAlchemy 映射实现,建议先在测试环境执行 SQL再启动服务联调。
### 33.1 自动化执行主表
```sql
CREATE TABLE IF NOT EXISTS auto_execution (
id BIGSERIAL PRIMARY KEY,
execution_no VARCHAR(64) NOT NULL UNIQUE,
trigger_type SMALLINT NOT NULL,
project_id BIGINT NOT NULL,
plan_id BIGINT,
plan_round_no INTEGER,
source_case_id BIGINT,
env_code VARCHAR(32) NOT NULL,
run_mode SMALLINT DEFAULT 1,
status SMALLINT NOT NULL DEFAULT 0,
jenkins_job_name VARCHAR(128),
jenkins_queue_id BIGINT,
jenkins_build_number BIGINT,
jenkins_build_url VARCHAR(512),
console_url VARCHAR(512),
report_url VARCHAR(512),
total_count INTEGER DEFAULT 0,
pending_count INTEGER DEFAULT 0,
running_count INTEGER DEFAULT 0,
passed_count INTEGER DEFAULT 0,
failed_count INTEGER DEFAULT 0,
blocked_count INTEGER DEFAULT 0,
skipped_count INTEGER DEFAULT 0,
not_found_count INTEGER DEFAULT 0,
trigger_by BIGINT,
trigger_source VARCHAR(32) DEFAULT 'platform',
trigger_message TEXT,
start_time TIMESTAMP,
end_time TIMESTAMP,
duration_seconds INTEGER,
callback_token VARCHAR(128),
ext JSONB DEFAULT '{}'::jsonb,
created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```
### 33.2 自动化执行明细表
```sql
CREATE TABLE IF NOT EXISTS auto_execution_case (
id BIGSERIAL PRIMARY KEY,
execution_id BIGINT NOT NULL,
plan_case_id BIGINT,
case_id BIGINT NOT NULL,
case_key VARCHAR(64),
case_title VARCHAR(255),
run_order INTEGER DEFAULT 0,
status SMALLINT NOT NULL DEFAULT 0,
pytest_nodeid VARCHAR(512),
result_message TEXT,
error_message TEXT,
stack_trace TEXT,
report_url VARCHAR(512),
duration_seconds INTEGER,
started_time TIMESTAMP,
finished_time TIMESTAMP,
retry_count INTEGER DEFAULT 0,
ext JSONB DEFAULT '{}'::jsonb,
created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```
### 33.3 索引
```sql
CREATE INDEX IF NOT EXISTS idx_auto_execution_plan_id ON auto_execution(plan_id);
CREATE INDEX IF NOT EXISTS idx_auto_execution_project_id ON auto_execution(project_id);
CREATE INDEX IF NOT EXISTS idx_auto_execution_status ON auto_execution(status);
CREATE INDEX IF NOT EXISTS idx_auto_execution_created_time ON auto_execution(created_time DESC);
CREATE INDEX IF NOT EXISTS idx_auto_execution_trigger_by ON auto_execution(trigger_by);
CREATE INDEX IF NOT EXISTS idx_auto_execution_case_execution_id ON auto_execution_case(execution_id);
CREATE INDEX IF NOT EXISTS idx_auto_execution_case_case_id ON auto_execution_case(case_id);
CREATE INDEX IF NOT EXISTS idx_auto_execution_case_plan_case_id ON auto_execution_case(plan_case_id);
CREATE INDEX IF NOT EXISTS idx_auto_execution_case_status ON auto_execution_case(status);
```
### 33.4 唯一约束建议
当前代码在 `get_execution_case_by_unique()` 中按:
- `execution_id + case_id`
-`plan_case_id` 时再附加 `plan_case_id`
如果你确认同一执行内允许同一个 `case_id` 因不同 `plan_case_id` 出现多次,建议使用下面唯一索引:
```sql
CREATE UNIQUE INDEX IF NOT EXISTS uk_auto_execution_case_exec_plan_case
ON auto_execution_case(execution_id, plan_case_id, case_id);
```
如果你确认同一执行中 `case_id` 不会重复,则可改为更强约束:
```sql
CREATE UNIQUE INDEX IF NOT EXISTS uk_auto_execution_case_exec_case
ON auto_execution_case(execution_id, case_id);
```
> 推荐和当前业务保持一致:计划里可能重复挂同一个 case 时,用第一种。
### 33.5 外键建议
如果你当前库中已有规范外键管理,可以加;如果历史库本身较轻、线上变更风险高,第一期可不加外键,只保留业务层约束。
可选外键:
```sql
ALTER TABLE auto_execution_case
ADD CONSTRAINT fk_auto_execution_case_execution_id
FOREIGN KEY (execution_id) REFERENCES auto_execution(id);
ALTER TABLE auto_execution_case
ADD CONSTRAINT fk_auto_execution_case_plan_case_id
FOREIGN KEY (plan_case_id) REFERENCES plan_case(id);
ALTER TABLE auto_execution_case
ADD CONSTRAINT fk_auto_execution_case_case_id
FOREIGN KEY (case_id) REFERENCES test_case(id);
```
### 33.6 updated_time 自动更新时间建议
当前 SQLAlchemy 已声明 `server_onupdate`,但 PostgreSQL 本身不会仅靠字段定义自动刷新 `updated_time`
如果你希望数据库层自动维护更新时间,建议补 trigger
```sql
CREATE OR REPLACE FUNCTION update_modified_column()
RETURNS TRIGGER AS $
BEGIN
NEW.updated_time = CURRENT_TIMESTAMP;
RETURN NEW;
END;
$ language 'plpgsql';
DROP TRIGGER IF EXISTS trg_auto_execution_updated_time ON auto_execution;
CREATE TRIGGER trg_auto_execution_updated_time
BEFORE UPDATE ON auto_execution
FOR EACH ROW
EXECUTE PROCEDURE update_modified_column();
DROP TRIGGER IF EXISTS trg_auto_execution_case_updated_time ON auto_execution_case;
CREATE TRIGGER trg_auto_execution_case_updated_time
BEFORE UPDATE ON auto_execution_case
FOR EACH ROW
EXECUTE PROCEDURE update_modified_column();
```
### 33.7 权限初始化 SQL
你当前代码新增了以下权限点:
- `automation:run`
- `automation:list`
- `automation:detail`
如果你们 RBAC 表结构支持直接插入权限编码,可参考下面 SQL字段名按你实际库调整
```sql
-- 以下为示例,请按实际 rbac_permission 表结构调整字段
INSERT INTO rbac_permission (name, code, type, parent_id, created_time, updated_time)
VALUES
('自动化执行', 'automation:run', 2, NULL, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP),
('自动化执行列表', 'automation:list', 2, NULL, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP),
('自动化执行详情', 'automation:detail', 2, NULL, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
ON CONFLICT (code) DO NOTHING;
```
如果你们权限还要挂菜单或角色,再补:
- 角色权限关联表
- 菜单权限关联表
---
## 34. pytest + Jenkins 联调请求示例
以下示例是按你当前后端已实现接口整理的,可直接给自动化项目同学或 Jenkins Pipeline 使用。
### 34.1 Jenkins 触发参数约定
平台调用 Jenkins `buildWithParameters` 时传:
```text
EXECUTION_ID=123
CALLBACK_TOKEN=abc123token
PLATFORM_BASE_URL=http://127.0.0.1:5010/it/api
ENV_CODE=st
RUN_MODE=1
TRIGGER_TYPE=2
```
其中:
- `CALLBACK_TOKEN`:用于 `/automation/execution/case/pull`
- `X-CALLBACK-SECRET`:用于回调写入接口,由 Jenkins 环境变量统一注入
---
### 34.2 Jenkins Pipeline 示例
```groovy
pipeline {
agent any
parameters {
string(name: 'EXECUTION_ID', defaultValue: '', description: '平台执行ID')
string(name: 'CALLBACK_TOKEN', defaultValue: '', description: '拉取执行清单token')
string(name: 'PLATFORM_BASE_URL', defaultValue: 'http://127.0.0.1:5010/it/api', description: '平台地址')
string(name: 'ENV_CODE', defaultValue: 'st', description: '环境编码')
string(name: 'RUN_MODE', defaultValue: '1', description: '1串行 2并行')
string(name: 'TRIGGER_TYPE', defaultValue: '2', description: '1单条 2计划')
}
environment {
CALLBACK_SECRET = credentials('AUTOMATION_CALLBACK_SECRET')
}
stages {
stage('Checkout') {
steps {
checkout scm
}
}
stage('Notify Start') {
steps {
bat '''
curl -X POST "%PLATFORM_BASE_URL%/automation/execution/start" ^
-H "Content-Type: application/json" ^
-H "X-CALLBACK-SECRET: %CALLBACK_SECRET%" ^
-d "{\"executionId\": %EXECUTION_ID%, \"jobName\": \"pytest-auto-runner\", \"buildNumber\": %BUILD_NUMBER%, \"buildUrl\": \"%BUILD_URL%\", \"consoleUrl\": \"%BUILD_URL%console\"}"
'''
}
}
stage('Run Pytest') {
steps {
bat '''
python run_execution.py ^
--execution-id %EXECUTION_ID% ^
--callback-token %CALLBACK_TOKEN% ^
--platform-base-url %PLATFORM_BASE_URL% ^
--env-code %ENV_CODE% ^
--run-mode %RUN_MODE% ^
--callback-secret %CALLBACK_SECRET%
'''
}
}
}
post {
failure {
bat '''
curl -X POST "%PLATFORM_BASE_URL%/automation/execution/abort" ^
-H "Content-Type: application/json" ^
-H "X-CALLBACK-SECRET: %CALLBACK_SECRET%" ^
-d "{\"executionId\": %EXECUTION_ID%, \"status\": 5, \"message\": \"Jenkins failed\", \"buildNumber\": %BUILD_NUMBER%, \"consoleUrl\": \"%BUILD_URL%console\"}"
'''
}
aborted {
bat '''
curl -X POST "%PLATFORM_BASE_URL%/automation/execution/abort" ^
-H "Content-Type: application/json" ^
-H "X-CALLBACK-SECRET: %CALLBACK_SECRET%" ^
-d "{\"executionId\": %EXECUTION_ID%, \"status\": 6, \"message\": \"Jenkins aborted\", \"buildNumber\": %BUILD_NUMBER%, \"consoleUrl\": \"%BUILD_URL%console\"}"
'''
}
}
}
```
---
### 34.3 pytest 项目:拉取执行清单示例
```bash
curl -X GET "http://127.0.0.1:5010/it/api/automation/execution/case/pull?executionId=123" \
-H "X-CALLBACK-TOKEN: abc123token"
```
响应示例:
```json
{
"code": 20000,
"data": {
"executionId": 123,
"executionNo": "AE202502120001",
"triggerType": 2,
"projectId": 88,
"planId": 2001,
"envCode": "st",
"runMode": 1,
"items": [
{
"executionCaseId": 501,
"planCaseId": 9001,
"caseId": 1001,
"caseKey": "TC-ZHYY-001",
"caseTitle": "登录成功",
"runOrder": 1
}
]
},
"msg": "success"
}
```
---
### 34.4 pytest 项目:开始执行回调示例
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/start" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"jobName": "pytest-auto-runner",
"buildNumber": 89,
"buildUrl": "http://jenkins/job/pytest-auto-runner/89/",
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console",
"startTime": "2025-02-12 10:00:00"
}'
```
---
### 34.5 pytest 项目:单条用例结果回调示例
#### 通过
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/case/result" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"executionCaseId": 501,
"caseId": 1001,
"status": 2,
"pytestNodeid": "tests/test_login.py::test_login_success",
"resultMessage": "assert success",
"errorMessage": "",
"stackTrace": "",
"durationSeconds": 12,
"reportUrl": "http://allure/case/1001",
"startedTime": "2025-02-12 10:01:00",
"finishedTime": "2025-02-12 10:01:12",
"ext": {
"attachments": []
}
}'
```
#### 失败
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/case/result" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"executionCaseId": 502,
"caseId": 1002,
"status": 3,
"pytestNodeid": "tests/test_login.py::test_login_fail",
"resultMessage": "assert failed",
"errorMessage": "AssertionError: 登录失败提示不匹配",
"stackTrace": "Traceback ...",
"durationSeconds": 8,
"reportUrl": "http://allure/case/1002",
"startedTime": "2025-02-12 10:01:13",
"finishedTime": "2025-02-12 10:01:21"
}'
```
#### 未找到自动化实现
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/case/result" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"executionCaseId": 503,
"caseId": 1003,
"status": 6,
"resultMessage": "case_id 未映射到 pytest 用例",
"errorMessage": "",
"durationSeconds": 0,
"finishedTime": "2025-02-12 10:01:22"
}'
```
---
### 34.6 pytest 项目:执行完成回调示例
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/finish" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"buildNumber": 89,
"buildUrl": "http://jenkins/job/pytest-auto-runner/89/",
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console",
"reportUrl": "http://allure/report/89",
"startTime": "2025-02-12 10:00:00",
"endTime": "2025-02-12 10:08:00",
"durationSeconds": 480,
"summary": {
"total": 15,
"passed": 12,
"failed": 2,
"blocked": 0,
"skipped": 1,
"notFound": 0
},
"message": "执行完成"
}'
```
> 注意:当前后端实现里 `finish_execution()` 最终仍会以明细表聚合结果为准刷新主单状态,所以 pytest 回传 `summary` 更像补充信息,不是唯一可信来源。
---
### 34.7 Jenkins 失败/取消兜底回调示例
#### 失败
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/abort" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"status": 5,
"message": "Jenkins failed",
"buildNumber": 89,
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console"
}'
```
#### 取消
```bash
curl -X POST "http://127.0.0.1:5010/it/api/automation/execution/abort" \
-H "Content-Type: application/json" \
-H "X-CALLBACK-SECRET: your-callback-secret" \
-d '{
"executionId": 123,
"status": 6,
"message": "Jenkins aborted",
"buildNumber": 89,
"consoleUrl": "http://jenkins/job/pytest-auto-runner/89/console"
}'
```
---
### 34.8 pytest 侧最小执行脚本伪代码
```python
import requests
import pytest
def pull_cases(base_url, execution_id, callback_token):
resp = requests.get(
f"{base_url}/automation/execution/case/pull",
params={"executionId": execution_id},
headers={"X-CALLBACK-TOKEN": callback_token},
timeout=30,
)
resp.raise_for_status()
return resp.json()["data"]
def callback_case_result(base_url, callback_secret, payload):
requests.post(
f"{base_url}/automation/execution/case/result",
json=payload,
headers={"X-CALLBACK-SECRET": callback_secret},
timeout=30,
).raise_for_status()
if __name__ == '__main__':
# 1. 拉取 execution 清单
# 2. 建立 case_id -> pytest nodeid 映射
# 3. 逐条执行
# 4. 每条回调结果
# 5. 最后回调 finish
pass
```
---
## 35. 结论
最适合你当前项目的方案不是改造现有 `plan_case_execute`,而是:
- **在现有 MVC 上新增独立 automation 执行域**
- **以 `case_id` 作为功能用例与 pytest 自动化用例的桥梁**
- **平台负责建单、编排、回写和状态聚合**
- **Jenkins 负责调度**
- **pytest 项目负责按 `execution_id` 拉取清单并执行**
这样设计:
- 和你当前项目结构最兼容
- AI 最容易按层写代码
- 后续扩展单条执行、计划批量、并行、重跑都不会推翻重来