qiaoxinjiu/effekt-interface

Fork 0

Files

qiaoxinjiu 2fea5adb44 提交所有代码到 qiaoxinjiu 分支

2026-05-11 14:29:16 +08:00

55 KiB

Raw Permalink Blame History

pytest + Jenkins 自动化执行接入后端详细方案（MVC / 可直接交付 AI 编码）

1. 目标与边界

1.1 目标

基于当前项目 D:\zhyy\effekt-interface 的现有 MVC 结构，为“计划中的自动化用例执行”设计一套可直接落地的后端方案，支持：

单条功能用例触发自动化执行。
测试计划维度批量触发自动化执行。
自动化项目基于 pytest 执行指定 case_id 对应的用例。
Jenkins 作为统一调度入口。
平台记录执行任务、执行明细、状态流转、Jenkins 构建信息、回调结果。
方案需适配当前项目的 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 执行，因此本期不强制新增自动化用例定位字段。

但建议增加以下可选字段，便于后续扩展：

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`

用途：记录一次自动化执行任务，支持单条执行和计划执行。

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 等

索引建议：

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 的状态和结果。

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: 可存截图、附件、标签等

索引建议：

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 交互轨迹，建议加事件表。

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 回调异常（任务可能跑完，但平台未完整接收结果）

推荐状态流：

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 已取消

推荐状态流：

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. 接口设计

接口分三类：

前端触发接口
自动化项目拉取执行清单接口
Jenkins/pytest 回调接口

以下设计遵循当前项目风格：

路由统一注册在 app/api/views.py
GET 查列表/详情
POST 创建/更新/触发
返回结构统一 ApiResponse.build_success/build_failure

10. 前端触发接口设计

10.1 单条用例执行

路由

POST /api/automation/case/run

权限

建议新增：automation:run

请求体

{
  "caseId": 1001,
  "envCode": "st",
  "runMode": 1,
  "jenkinsJobName": "pytest-auto-runner",
  "remark": "单条回归验证"
}

字段说明

caseId: 必填
envCode: 必填
runMode: 可选，默认 1 串行
jenkinsJobName: 可选，未传则走系统默认配置
remark: 可选

{
  "id": 123,
  "executionId": 123,
  "executionNo": "AE202502120001",
  "status": 2
}

处理逻辑

校验 caseId 是否存在。
校验 test_case.is_auto == 1。
创建 auto_execution：
- trigger_type=1
- source_case_id=caseId
- project_id 从 test_case.project_id 带出
创建一条 auto_execution_case。
调用 Jenkins。
若调用成功：
- 更新主表 status=2
- 记录 queue_id / job_name
若调用失败：
- 更新主表 status=7
- 返回失败原因

10.2 计划自动化执行

路由

POST /api/automation/plan/run

权限

automation:run

请求体

{
  "planId": 2001,
  "roundNo": 1,
  "envCode": "st",
  "runMode": 1,
  "caseIds": [1001, 1002, 1003],
  "jenkinsJobName": "pytest-auto-runner",
  "remark": "计划批量回归"
}

说明

caseIds 可选：
- 不传或空：执行当前计划下全部自动化用例
- 传值：执行计划中指定子集
roundNo 可选：如果你已有轮次概念，则仅执行该轮次下的 plan_case

{
  "id": 124,
  "executionId": 124,
  "executionNo": "AE202502120002",
  "totalCount": 15,
  "status": 2
}

处理逻辑

校验 planId。
查询 plan_case + test_case：
- 必须属于该计划
- test_case.is_auto=1
- 若传 caseIds，则限制子集
- 若传 roundNo，则限制轮次
若无可执行自动化用例，直接报错。
创建 auto_execution：
- trigger_type=2
- plan_id=planId
- plan_round_no=roundNo
批量创建 auto_execution_case。
调 Jenkins。
更新状态与 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

鉴权

建议两层：

正常 token 鉴权（如果自动化项目能拿到平台 token）
或使用 executionId + callbackToken 鉴权

建议本期采用：

query: executionId
header: X-CALLBACK-TOKEN

请求参数

executionId

请求头

X-CALLBACK-TOKEN: xxx

{
  "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
    }
  ]
}

处理逻辑

校验 executionId。
校验回调 token。
查询 auto_execution 和 auto_execution_case。
返回待执行清单。

如果未来 pytest 项目还需要额外上下文，如环境 URL、测试账号、marker，也可以从 ext 中一起返回。

12. 回调接口设计

推荐分 4 个回调点：

Jenkins 入队成功
Jenkins 开始执行
单条用例结果回写
整体执行完成回写

12.1 Jenkins 入队回写

路由

POST /api/automation/execution/queued

作用

Jenkins 已接受构建，回传 queue/build 基本信息。

请求体

{
  "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

请求体

{
  "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

请求体

{
  "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": []
  }
}

处理逻辑

按 executionCaseId 优先定位明细；若无则退化为 executionId + caseId。
更新 auto_execution_case。
若有关联 plan_case_id，同步回写 plan_case：
- status
- actual_result
- executed_time
- execution_duration
重新聚合更新 auto_execution 汇总字段。
若是计划执行，可调用计划状态更新逻辑。

12.4 整体执行完成回写

路由

POST /api/automation/execution/finish

请求体

{
  "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": "执行完成"
}

处理逻辑

更新主表 Jenkins 信息与时间。
若请求体有 summary，则直接写回；否则后台实时聚合明细表得到。
根据汇总决定主状态：
- failed > 0 or blocked > 0 or notFound > 0 -> 5失败
- 否则 -> 4成功
若是计划执行，调用计划状态刷新。

12.5 失败/取消回调

路由

POST /api/automation/execution/abort

请求体

{
  "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 构建参数

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 可选参数

GIT_BRANCH
PYTEST_ARGS
REPORT_BASE_URL

存放位置建议：

平台配置中心 / 环境变量 / 常量文件
不建议写死在 controller

13.4 Jenkins Build API 请求方式

推荐后端调用：

POST /job/{jobName}/buildWithParameters

鉴权建议：

Basic Auth + API Token
若 Jenkins 开启 crumb 防护，还要先请求 crumb

14. Jenkins 回调与结果回写逻辑

14.1 平台触发 Jenkins 时的处理

AutomationService.trigger_execution(...) 需要：

生成 callback_token
创建执行单和明细
更新主状态 0 -> 1
调 Jenkins API
成功：
- 主状态 1 -> 2
- 保存 queue/build 基础信息
失败：
- 主状态 1 -> 7
- 保存错误信息到 trigger_message

14.2 pytest 项目开始时的处理

pytest 启动脚本：

根据 Jenkins 参数拿到 EXECUTION_ID
调平台 /execution/case/pull
成功拿清单后调用 /execution/start
开始逐条执行

14.3 pytest 每执行一条用例

执行器需要：

将 case_id 定位到 pytest 对应测试项
执行完成后回调 /execution/case/result
回调内容至少包括：
- executionCaseId / caseId
- status
- durationSeconds
- nodeid
- errorMessage/resultMessage

14.4 pytest 全量执行结束

执行器需要：

汇总总数
计算开始/结束时间
发布 report_url（如 Allure）
调 /execution/finish

14.5 Jenkins 失败兜底

即使 pytest 未回调成功，Jenkins post { failure { ... } } 阶段也应该：

至少调用 /execution/abort 或 /execution/finish(status=5)
避免平台主单永久卡在 2/3

15. 详细执行流程

15.1 单条用例执行流程

前端点击“执行自动化”
-> 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 计划批量执行流程

前端点击“执行自动化用例”
-> 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 新增文件

app/api/model/automationModel.py
app/api/dao/automationDao.py
app/api/service/automationService.py
app/api/controller/automationController.py
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 中：

若存在 running_count > 0 -> 主状态至少为 3
若全部结束：
- failed + blocked + not_found > 0 -> 5
- 否则 -> 4
若任务被外部取消 -> 6
若触发阶段失败 -> 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() 推荐事务内做：

更新 auto_execution_case
更新 plan_case
提交
再调用聚合更新 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 编码更明确，建议新增语义化错误码：

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 回调时传：

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 自动化项目行为

调 /execution/case/pull
将 case_id 映射到 pytest 测试项
单条执行后调 /execution/case/result
全部结束调 /execution/finish

28.3 `case_id` 到 pytest 的映射方式

你已经确定桥梁是 case_id，建议自动化项目采用以下之一：

方式 A：pytest marker

@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：数据层

新建数据库表：
- auto_execution
- auto_execution_case
新建 automationModel.py
新建 automationDao.py
编写：
- 创建主单
- 批量创建明细
- 查询详情/列表
- 聚合汇总

Phase 2：服务层

新建 automationService.py
实现：
- 单条执行建单
- 计划执行建单
- Jenkins 触发
- 执行清单拉取
- 单条结果回写
- 整体完成回写
- 汇总刷新
把 PlanController._update_plan_status() 逻辑迁移成服务方法

Phase 3：控制器与路由

新建 automationController.py
在 views.py 注册接口
增加回调接口白名单处理

Phase 4：Jenkins 客户端

新建 common/jenkinsRequest.py
封装 Basic Auth + buildWithParameters
在 service 中接入

Phase 5：联调准备

给 pytest 项目提供接口文档
给 Jenkins Pipeline 提供参数清单
约定回调协议

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 状态更新

任务 5：Jenkins 客户端与配置收口

输出：

common/jenkinsRequest.py
const.py 新配置
回调白名单

31. 最终推荐实现决策

必做

新增 auto_execution、auto_execution_case
独立 automation MVC 模块
平台只传 execution_id 给 Jenkins
pytest 项目按 execution_id 拉清单
单条回写 + 整体回写
回写 plan_case 和 test_plan 状态

建议做

callback_token 鉴权
主/明细状态机
幂等处理
Jenkins 失败兜底回调

可后置

并行执行
失败重跑
定时补偿
执行事件表
自动创建缺陷/附件

32. 可直接交给 AI 的开发说明摘要

你可以把下面这段直接给 AI 作为实现要求：

基于当前 Flask MVC 结构，新增 automation 域：controller/service/dao/model。
新增表 auto_execution、auto_execution_case，字段按本方案实现。
提供接口：
- 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
单条执行通过 test_case.id 建单，计划执行通过 plan_case + test_case.is_auto=1 批量建单。
主状态与明细状态按本方案的枚举实现。
/case/result 回调时同步更新 plan_case.status / actual_result / executed_time / execution_duration。
每次明细回写后聚合更新 auto_execution 的计数和主状态。
计划执行时，执行结果变化后要刷新 test_plan.status，把当前 PlanController._update_plan_status() 逻辑迁到 service 层复用。
新增 common/jenkinsRequest.py，封装 Jenkins buildWithParameters 调用。
Jenkins 参数至少包含：EXECUTION_ID、CALLBACK_TOKEN、PLATFORM_BASE_URL、ENV_CODE、RUN_MODE、TRIGGER_TYPE。
回调接口采用 X-CALLBACK-TOKEN 鉴权，并加入免登录白名单。

33. PostgreSQL 建表 SQL（可直接落库）

说明：以下 SQL 仅覆盖本次自动化执行接入新增对象，不包含已有 test_case / plan_case / test_plan 表。当前代码已按 PostgreSQL + SQLAlchemy 映射实现，建议先在测试环境执行 SQL，再启动服务联调。

33.1 自动化执行主表

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 自动化执行明细表

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 索引

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 出现多次，建议使用下面唯一索引：

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 不会重复，则可改为更强约束：

CREATE UNIQUE INDEX IF NOT EXISTS uk_auto_execution_case_exec_case
ON auto_execution_case(execution_id, case_id);

推荐和当前业务保持一致：计划里可能重复挂同一个 case 时，用第一种。

33.5 外键建议

如果你当前库中已有规范外键管理，可以加；如果历史库本身较轻、线上变更风险高，第一期可不加外键，只保留业务层约束。

可选外键：

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：

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，字段名按你实际库调整：

-- 以下为示例，请按实际 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 时传：

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 示例

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 项目：拉取执行清单示例

curl -X GET "http://127.0.0.1:5010/it/api/automation/execution/case/pull?executionId=123" \
  -H "X-CALLBACK-TOKEN: abc123token"

响应示例：

{
  "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 项目：开始执行回调示例

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 项目：单条用例结果回调示例

通过

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": []
    }
  }'

失败

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"
  }'

未找到自动化实现

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 项目：执行完成回调示例

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 失败/取消兜底回调示例

失败

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"
  }'

取消

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 侧最小执行脚本伪代码

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 最容易按层写代码
后续扩展单条执行、计划批量、并行、重跑都不会推翻重来

55 KiB Raw Permalink Blame History Unescape Escape

pytest + Jenkins 自动化执行接入后端详细方案（MVC / 可直接交付 AI 编码）

1. 目标与边界

1.1 目标

1.2 已知现状

1.3 本期边界

2. 总体架构设计

2.1 责任划分

平台（本项目）

Jenkins

pytest 自动化项目

2.2 为什么不直接把 case_id 列表塞进 Jenkins 参数

3. 与当前 MVC 框架的集成方式

3.1 新增模块建议

3.2 为什么独立成 automation 域

3.3 与现有域的关系

与 Case 域的关系

与 Plan 域的关系

4. 数据库表结构设计

5. 现有表改造建议

5.1 test_case 表改造

6. 新增核心表结构

6.1 自动化执行主表 auto_execution

6.2 自动化执行明细表 auto_execution_case

6.3 可选事件表 auto_execution_event

7. SQLAlchemy Model 设计（贴合当前项目风格）

8. 状态流转设计

8.1 主执行状态 auto_execution.status

8.2 明细状态 auto_execution_case.status

8.3 与现有 plan_case.status 的映射关系

8.4 与 test_plan.status 的映射关系

9. 接口设计

10. 前端触发接口设计

10.1 单条用例执行

路由

权限

请求体

字段说明

返回

处理逻辑

10.2 计划自动化执行

路由

权限

请求体

说明

返回

处理逻辑

10.3 自动化执行列表

路由

请求参数

返回字段

10.4 自动化执行详情

路由

请求参数

返回

10.5 自动化执行明细列表

路由

请求参数

返回

11. 自动化项目拉取执行清单接口

11.1 获取执行清单

路由

鉴权

请求参数

请求头

返回

处理逻辑

12. 回调接口设计

12.1 Jenkins 入队回写

路由

作用

请求体

处理

12.2 Jenkins/pytest 开始执行回写

路由

请求体

处理

12.3 单条用例结果回写

路由

请求体

55 KiB

Raw Permalink Blame History

5.1 `test_case` 表改造

6.1 自动化执行主表 `auto_execution`

6.2 自动化执行明细表 `auto_execution_case`

6.3 可选事件表 `auto_execution_event`

8.1 主执行状态 `auto_execution.status`

8.2 明细状态 `auto_execution_case.status`

8.3 与现有 `plan_case.status` 的映射关系

8.4 与 `test_plan.status` 的映射关系

`app/api/views.py`

`app/api/service/planService.py`

`app/api/controller/planController.py`

`const.py`

18. `common/jenkinsRequest.py` 设计建议

19. 回写 `plan_case` 的规则细化

24.4 pytest 找不到 `case_id` 对应实现