# 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`,建议自动化项目采用以下之一: #### 方式 A:pytest 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 4:Jenkins 客户端 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 状态更新 ### 任务 5:Jenkins 客户端与配置收口 输出: - `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 最容易按层写代码 - 后续扩展单条执行、计划批量、并行、重跑都不会推翻重来