#!/usr/bin/env python3 """Create/update the school dual-database Apifox scenarios S01-S06. The script intentionally reads the already logged-in Apifox desktop session from the local Chromium storage. It never prints or persists that session token. """ from __future__ import annotations import copy import json import uuid from pathlib import Path from typing import Any from urllib.error import HTTPError from urllib.request import Request, urlopen PROJECT_ID = 8_881_630 FOLDER_ID = 8_019_398 TEST_SUITE_ID = 32_562 API_BASE = "https://api.apifox.com" DESKTOP_STORAGE = Path.home() / "Library/Application Support/apifox/Local Storage/leveldb/000005.ldb" MYSQL_BASE_URL = "http://127.0.0.1:3551" DM_BASE_URL = "http://127.0.0.1:3550" def desktop_authorization() -> str: raw = DESKTOP_STORAGE.read_bytes() marker = b"common.accessToken" start = raw.find(marker) if start < 0: raise RuntimeError("Apifox desktop access token marker was not found") printable: list[bytes] = [] current = bytearray() for value in raw[start : start + 600]: if 32 <= value < 127: current.append(value) else: if len(current) >= 8: printable.append(bytes(current)) current.clear() candidate = next((item for item in printable if item.startswith(b'"Bearer ')), None) if not candidate: raise RuntimeError("Apifox desktop is not logged in") return candidate.decode("ascii").strip('"') class Apifox: def __init__(self) -> None: self.headers = { "Authorization": desktop_authorization(), "X-Project-Id": str(PROJECT_ID), "X-Branch-Id": "0", "X-Client-Version": "2.8.26", "X-Client-Mode": "electron", "Accept-Language": "zh-CN", "User-Agent": "Mozilla/5.0 Apifox/2.8.26", "Content-Type": "application/json", } def call(self, path: str, method: str = "GET", data: Any | None = None) -> dict[str, Any]: request = Request( API_BASE + path, data=None if data is None else json.dumps(data, ensure_ascii=False).encode("utf-8"), headers=self.headers, method=method, ) try: with urlopen(request, timeout=30) as response: body = response.read() except HTTPError as error: detail = error.read().decode("utf-8", "ignore")[:500] raise RuntimeError(f"Apifox HTTP {error.code}: {detail}") from error result = json.loads(body or b"{}") if not result.get("success", False): raise RuntimeError(f"Apifox request failed: {result.get('errorCode')} {result.get('errorMessage')}") return result def scenario(self, scenario_id: int) -> dict[str, Any]: return self.call(f"/api/v1/api-test/cases/{scenario_id}/steps?withCaseDetail=true")["data"] def scenarios(self) -> list[dict[str, Any]]: data = self.call(f"/api/v1/projects/{PROJECT_ID}/test-scenario/tree-list")["data"] return data["testScenarios"] def duplicate(self, source_id: int) -> int: created = self.call( "/api/v1/api-test/cases/duplicate", "POST", {"id": source_id, "folderId": FOLDER_ID, "ordering": 999}, )["data"] return int(created["id"]) def save(self, scenario: dict[str, Any]) -> None: fields = ( "id", "name", "description", "priority", "ordering", "folderId", "steps", "options", "performanceTestOptions", "tags", "preProcessors", "postProcessors", ) payload = {field: scenario[field] for field in fields if field in scenario} self.call(f"/api/v1/api-test/cases/{scenario['id']}", "PUT", payload) def test_suite(self) -> dict[str, Any]: return self.call(f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites/{TEST_SUITE_ID}")["data"] def save_test_suite(self, suite: dict[str, Any]) -> None: self.call( f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites/{TEST_SUITE_ID}", "PUT", suite, ) def new_id() -> str: return str(uuid.uuid4()) def strip_server_fields(request: dict[str, Any]) -> dict[str, Any]: cloned = copy.deepcopy(request) for key in ("id", "createdAt", "updatedAt", "deletedAt"): cloned.pop(key, None) return cloned def assertion(expected_code: int = 200, name: str | None = None) -> dict[str, Any]: return { "type": "assertion", "data": { "name": name or f"业务码 code={expected_code}", "subject": "responseJson", "comparison": "equal", "value": str(expected_code), "path": "$.code", "multipleValue": [], "extractSettings": { "expression": "$.code", "continueExtractorSettings": { "isContinueExtractValue": False, "JsonArrayValueIndexValue": "", }, }, }, "defaultEnable": False, "enable": True, } def custom_script(data: str) -> dict[str, Any]: return { "type": "customScript", "data": data, "defaultEnable": False, "enable": True, } def evidence_script( *, expected_http: int, expected_code: int, tests: list[tuple[str, str]] | None = None, ) -> dict[str, Any]: """Add field-level assertions and a redacted response snapshot to the Apifox console.""" lines = [ "const __text = pm.response.text();", "let __json = null;", "try { __json = pm.response.json(); } catch (e) {}", ( f"pm.test('HTTP 状态码 = {expected_http}', function () {{ " f"pm.expect(pm.response.code).to.equal({expected_http}); }});" ), "pm.test('响应正文存在', function () { pm.expect(__text.length).to.be.greaterThan(0); });", "pm.test('响应正文是 JSON', function () { pm.expect(__json).to.be.an('object'); });", ( f"pm.test('业务码 code = {expected_code}', function () {{ " f"pm.expect(__json && __json.code).to.equal({expected_code}); }});" ), ] if expected_code == 200: lines.append( "pm.test('响应包含 data 字段', function () { " "pm.expect(__json).to.have.property('data'); });" ) for name, statement in tests or []: escaped_name = name.replace("\\", "\\\\").replace("'", "\\'") lines.append(f"pm.test('{escaped_name}', function () {{ {statement}; }});") lines.extend( [ "function __summaryText(body) {", " if (!body || typeof body !== 'object') return '非 JSON 响应，chars=' + __text.length;", " const parts = ['code=' + String(body.code)];", " if (body.message && body.message !== 'OK') parts.push('message=' + String(body.message).slice(0, 32));", " const data = body.data;", " const safeKeys = ['id', 'docNo', 'recordNo', 'batchNo', 'result', 'value', 'status', 'quantity', 'total', 'count', 'success', 'failed', 'fail', 'saved', 'affectedRows', 'ok', 'page', 'pageSize'];", " const addFields = function (value, prefix) {", " if (!value || typeof value !== 'object') return;", " safeKeys.forEach(function (key) {", " if (value[key] !== undefined && value[key] !== null && typeof value[key] !== 'object') parts.push((prefix || '') + key + '=' + String(value[key]).slice(0, 24));", " });", " };", " if (Array.isArray(data)) {", " parts.push('items=' + data.length);", " if (data[0]) addFields(data[0], 'first.');", " } else if (data && typeof data === 'object') {", " addFields(data, '');", " const list = Array.isArray(data.items) ? data.items : (Array.isArray(data.list) ? data.list : (Array.isArray(data.records) ? data.records : null));", " if (list) {", " parts.push('items=' + list.length);", " if (list[0]) addFields(list[0], 'first.');", " }", " if (typeof data.csv === 'string') parts.push('csvChars=' + data.csv.length);", " if (parts.length === 1) parts.push('dataKeys=' + Object.keys(data).filter(function (key) { return !/token|password|secret|authorization|access.?key/i.test(key); }).slice(0, 8).join(','));", " } else if (data !== undefined && data !== null) {", " parts.push('data=' + String(data).slice(0, 32));", " }", " return parts.join('，').slice(0, 120);", "}", "pm.test('实际返回｜' + __summaryText(__json), function () { pm.expect(__text.length).to.be.greaterThan(0); });", "function __redact(value) {", " if (Array.isArray(value)) return value.slice(0, 20).map(__redact);", " if (value && typeof value === 'object') {", " const out = {};", " Object.keys(value).slice(0, 80).forEach(function (key) {", " if (/token|password|secret|authorization|access.?key/i.test(key)) out[key] = '[REDACTED]';", " else out[key] = __redact(value[key]);", " });", " return out;", " }", " if (typeof value === 'string') {", " const masked = value", " .replace(/(\\\"(?:password|token|secret|authorization|access.?key)\\\"\\s*:\\s*\\\")[^\\\"]*/ig, '$1[REDACTED]')", " .replace(/(Bearer\\s+)[A-Za-z0-9._-]+/ig, '$1[REDACTED]');", " return masked.length > 2000 ? masked.slice(0, 2000) + '...[TRUNCATED]' : masked;", " }", " return value;", "}", "const __evidence = {", " step: pm.info.requestName,", " method: pm.request.method,", " url: pm.request.url.toString(),", " httpStatus: pm.response.code,", " responseTimeMs: pm.response.responseTime,", " responseSize: pm.response.responseSize,", " requestBody: pm.request.body ? __redact(pm.request.body.toJSON ? pm.request.body.toJSON() : String(pm.request.body)) : null,", " responseBody: __redact(__json === null ? __text : __json)", "};", "console.log('APIFOX_EVIDENCE ' + JSON.stringify(__evidence));", ] ) return custom_script("\n".join(lines)) def extractor(variable: str, expression: str) -> dict[str, Any]: return { "type": "extractor", "data": { "variableName": variable, "variableType": "local", "subject": "responseJson", "template": "", "expression": expression, "extractSettings": { "expression": expression, "continueExtractorSettings": { "isContinueExtractValue": False, "JsonArrayValueIndexValue": "", }, }, }, "defaultEnable": False, "enable": True, } def placeholder() -> dict[str, Any]: return { "id": "dynamicValueDivider", "type": "placeholder", "renderType": "dynamicValueDivider", "defaultEnable": False, "enable": False, } def make_login(template: dict[str, Any], prefix: str) -> dict[str, Any]: step = copy.deepcopy(template) step["id"] = new_id() step["number"] = 1 step["disable"] = False step.pop("relatedId", None) request = strip_server_fields(step["customHttpRequest"]) request.update( { "name": "登录并初始化双库测试变量", "method": "post", "path": "{{base_url}}/auth/login", "relatedId": 0, "requestBody": { "type": "application/json", "parameters": [], "example": '{"username":"{{app_admin_username}}","password":"{{app_admin_password}}"}', "data": '{"username":"{{app_admin_username}}","password":"{{app_admin_password}}"}', }, "parameters": { "header": [ { "name": "Content-Type", "value": "application/json", "sampleValue": "application/json", "enable": True, } ] }, "preProcessors": [ { "type": "customScript", "data": ( "pm.variables.set('base_url', pm.variables.get('$.1.element'));\n" f"pm.variables.set('test_batch', '{prefix}_' + Date.now());\n" ), "defaultEnable": False, "enable": True, "executionTiming": "prerequest", }, placeholder(), ], "postProcessors": [ extractor("auth_token", "$.data.token"), assertion(200, "登录业务码 code=200"), evidence_script( expected_http=201, expected_code=200, tests=[ ("登录返回临时 Token（报告脱敏）", "pm.expect(__json.data.token).to.be.a('string').and.not.empty"), ], ), ], "auth": {}, "securityScheme": {}, } ) step["customHttpRequest"] = request return step def make_request( template: dict[str, Any], *, name: str, method: str, path: str, body: dict[str, Any] | list[Any] | str | None = None, extract: tuple[str, str] | None = None, expected_code: int = 200, expected_http: int | None = None, disabled: bool = False, no_auth: bool = False, tests: list[tuple[str, str]] | None = None, ) -> dict[str, Any]: step = copy.deepcopy(template) step["id"] = new_id() step["disable"] = disabled step.pop("relatedId", None) request = strip_server_fields(step["customHttpRequest"]) request_path = "{{base_url}}" + path if path.startswith("/") else path request.update( { "name": name, "method": method.lower(), "path": request_path, "relatedId": 0, "commonParameters": {"query": [], "body": [], "header": [], "cookie": []}, "preProcessors": [placeholder()], "auth": {} if no_auth else {"type": "bearer", "bearer": {"token": "{{auth_token}}"}}, "securityScheme": {}, "advancedSettings": { "disabledSystemHeaders": {}, "isDefaultUrlEncoding": 2, "disableUrlEncoding": False, }, } ) headers = [] if no_auth else [ { "name": "Authorization", "value": "Bearer {{auth_token}}", "sampleValue": "Bearer {{auth_token}}", "enable": True, } ] if body is None: request["requestBody"] = {"type": "none", "parameters": [], "data": "", "example": ""} else: data = body if isinstance(body, str) else json.dumps(body, ensure_ascii=False, separators=(",", ":")) request["requestBody"] = { "type": "application/json", "parameters": [], "data": data, "example": data, } headers.insert( 0, { "name": "Content-Type", "value": "application/json", "sampleValue": "application/json", "enable": True, }, ) request["parameters"] = {"header": headers} resolved_http = expected_http if resolved_http is None: resolved_http = 201 if method.upper() == "POST" and expected_code == 200 else expected_code resolved_tests = list(tests or []) if extract and extract[1] == "$.data.id": resolved_tests.append( ( "写入响应：返回非空业务 ID", "pm.expect(String(__json.data.id === undefined || __json.data.id === null ? '' : __json.data.id).trim().length).to.be.greaterThan(0)", ) ) processors = [ assertion(expected_code), evidence_script(expected_http=resolved_http, expected_code=expected_code, tests=resolved_tests), ] if extract: processors.insert(0, extractor(*extract)) request["postProcessors"] = processors step["customHttpRequest"] = request return step SCENARIOS: dict[str, dict[str, Any]] = { "S01-首页与预警（双库）": { "prefix": "APIFOX_S01", "description": "学校端 S01 首页与预警双库场景。只读主流程自动执行；事件详情/处置保留为需要事件夹具的独立步骤；明厨亮灶仅以越权学校号验证 403，禁止真实第三方出站。", "operations": [ ("首页总览", "GET", "/school/home/overview?schoolId=1&aiPeriod=week"), ("设备概览", "GET", "/school/devices?schoolId=1"), ("分析看板", "GET", "/school/analytics/dashboard?schoolId=1"), ("预警列表", "GET", "/school/analytics/alerts?schoolId=1"), ("联网 AI 汇总", "GET", "/school/ai/home/summary/network?schoolId=1"), ("本地 AI 汇总", "GET", "/school/ai/home/summary/local?schoolId=1"), {"name": "AI 事件列表", "method": "GET", "path": "/school/ai/events?schoolId=1&page=1&pageSize=20", "extract": ("ai_event_id", "$.data.items[0].id")}, {"name": "AI 事件详情（需夹具）", "method": "GET", "path": "/school/ai/events/detail?id={{ai_event_id}}", "disabled": True}, {"name": "AI 事件处置（需夹具）", "method": "POST", "path": "/school/ai/events/handle", "body": {"eventId": "{{ai_event_id}}", "decision": "已现场纠正", "note": "{{test_batch}} 自动化处置", "handledBy": "自动化管理员"}, "disabled": True}, ("食堂列表", "GET", "/school/canteens?schoolId=1"), ("食堂汇总", "GET", "/school/canteens/summary?schoolId=1"), ("本地 AI 事件", "GET", "/bright-local/ai/events?schoolId=1&page=1&pageSize=20"), ("监管学校列表", "GET", "/reg/schools"), {"name": "明厨亮灶越权安全拒绝", "method": "GET", "path": "/bright/school/cameras?schoolId=99999999", "expected_code": 403}, {"name": "TrustIVS 越权安全拒绝", "method": "GET", "path": "/bright/school/trustivs-cameras?schoolId=99999999", "expected_code": 403}, ], }, "S02-晨检管理（双库真实写入）": { "prefix": "APIFOX_S02", "description": "学校端 S02 晨检双库真实写入场景：新增、列表/总览/历史回查、措施、处置与 CSV。数据覆盖中文、DECIMAL(4,1)、日期时间和 JSON；删除步骤默认关闭，保留本轮数据用于两库原表回查。", "operations": [ {"name": "新增晨检", "method": "POST", "path": "/school/morning-checks", "body": {"schoolId": 1, "staff": "{{test_batch}}_张三", "temp": 36.7, "at": "2026-09-25T08:30:00+08:00", "hygiene": "正常", "source": "manual"}, "extract": ("morning_id", "$.data.id")}, ("晨检列表回查", "GET", "/school/morning-checks?schoolId=1&staff={{test_batch}}_张三&page=1&pageSize=20"), ("晨检总览", "GET", "/school/morning-checks/overview?schoolId=1&date=2026-09-25"), ("晨检历史", "GET", "/school/morning-checks/history?schoolId=1&staff={{test_batch}}_张三&date=2026-09-25"), {"name": "更新晨检措施", "method": "PATCH", "path": "/school/morning-checks/{{morning_id}}/measure", "body": {"measure": "{{test_batch}} 复测正常，允许上岗"}}, {"name": "更新晨检处置", "method": "PATCH", "path": "/school/morning-checks/{{morning_id}}/handle", "body": {"decision": "allow", "note": "{{test_batch}} 中文处置：体温复测 36.6℃"}}, ("导出晨检 CSV", "GET", "/school/morning-checks/export.csv?schoolId=1&staff={{test_batch}}_张三"), {"name": "删除本轮晨检（原表取证后按需执行）", "method": "DELETE", "path": "/school/morning-checks/{{morning_id}}", "disabled": True}, ], }, "S03-农残检测（双库真实写入）": { "prefix": "APIFOX_S03", "description": "学校端 S03 农残检测双库可审计场景：手工新增、字段级回查、处置后二次回查、CSV、模拟设备入站、分页、无 Token、跨校和缺字段零写入。每步后置脚本输出脱敏 APIFOX_EVIDENCE；SSE 因长连接单独执行。", "operations": [ { "name": "新增农残检测并返回创建 ID", "method": "POST", "path": "/school/pesticide/records", "body": { "schoolId": 1, "sample": "{{test_batch}}_油麦菜", "device": "NY-TEST-01", "tester": "自动化检测员", "jiancedidian": "第一食堂快检室", "lianxidianhua": "13800000000", "value": "12.345", "result": "不合格", "remark": "{{test_batch}} 中文；特殊字符（）", "at": "2026-09-25T09:15:00+08:00", }, "extract": ("pesticide_id", "$.data.id"), "tests": [ ("创建成功：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)"), ("创建响应：样本名与本轮批次一致", "pm.expect(__json.data.sample).to.equal(pm.variables.get('test_batch') + '_油麦菜')"), ("创建响应：数值字符串 12.345 未丢失", "pm.expect(String(__json.data.value)).to.equal('12.345')"), ("创建响应：中文与特殊字符未乱码", "pm.expect(__json.data.remark).to.include('中文；特殊字符（）')"), ("创建响应：不合格记录 exception=true", "pm.expect(__json.data.exception).to.equal(true)"), ], }, { "name": "按批次回查刚创建的农残记录", "method": "GET", "path": "/school/pesticide/records?schoolId=1&q={{test_batch}}&page=1&pageSize=20", "tests": [ ("回查响应：items 为数组且 total>=1", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.total)).to.be.at.least(1)"), ("回查响应：包含刚创建的 ID", "const id = Number(pm.variables.get('pesticide_id')); pm.expect(__json.data.items.some(function (x) { return Number(x.id) === id; })).to.equal(true)"), ("回查响应：批次、数值、中文和结果一致", "const id = Number(pm.variables.get('pesticide_id')); const row = __json.data.items.find(function (x) { return Number(x.id) === id; }); pm.expect(row.sample).to.equal(pm.variables.get('test_batch') + '_油麦菜'); pm.expect(String(row.value)).to.equal('12.345'); pm.expect(row.result).to.equal('不合格'); pm.expect(row.remark).to.include('中文；特殊字符（）')"), ], }, { "name": "更新农残处置措施", "method": "PATCH", "path": "/school/pesticide/records/{{pesticide_id}}/measure", "body": {"measure": "{{test_batch}} 下架并复检"}, "tests": [("更新响应：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "处置后再次回查原记录", "method": "GET", "path": "/school/pesticide/records?schoolId=1&q={{test_batch}}&page=1&pageSize=20", "tests": [ ("处置回查：措施已真实更新", "const id = Number(pm.variables.get('pesticide_id')); const row = __json.data.items.find(function (x) { return Number(x.id) === id; }); pm.expect(row).to.exist; pm.expect(row.measure).to.equal(pm.variables.get('test_batch') + ' 下架并复检')"), ], }, { "name": "导出农残 CSV 并核对本轮数据", "method": "GET", "path": "/school/pesticide/records/export.csv?schoolId=1&q={{test_batch}}", "tests": [ ("CSV 响应：包含表头", "pm.expect(__json.data.csv).to.include('检测时间,检测样本,结果')"), ("CSV 响应：包含本轮批次与数值", "pm.expect(__json.data.csv).to.include(pm.variables.get('test_batch')); pm.expect(__json.data.csv).to.include('12.345')"), ], }, { "name": "模拟设备回调写入（不连接真实设备）", "method": "POST", "path": "/school/pesticide/device/callback", "no_auth": True, "body": { "schoolId": 1, "sample": "{{test_batch}}_设备青菜", "device": "MOCK-DEVICE-01", "result": "合格", "remark": "{{test_batch}} 本地模拟设备入站", }, "extract": ("pesticide_device_id", "$.data.id"), "tests": [ ("设备回调：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)"), ("设备回调：source=device", "pm.expect(__json.data.source).to.equal('device')"), ("设备回调：批次样本未串库", "pm.expect(__json.data.sample).to.equal(pm.variables.get('test_batch') + '_设备青菜')"), ], }, { "name": "分页回查手工与模拟设备两条数据", "method": "GET", "path": "/school/pesticide/records?schoolId=1&q={{test_batch}}&page=1&pageSize=1", "tests": [ ("分页响应：pageSize=1 只返回一条", "pm.expect(__json.data.items).to.have.lengthOf(1); pm.expect(Number(__json.data.pageSize)).to.equal(1)"), ("分页响应：本轮总数至少两条", "pm.expect(Number(__json.data.total)).to.be.at.least(2)"), ], }, { "name": "缺检测员应 400 且不得写入", "method": "POST", "path": "/school/pesticide/records", "expected_code": 400, "expected_http": 400, "body": { "schoolId": 1, "sample": "{{test_batch}}_NEG_NO_TESTER", "device": "NY-TEST-01", "result": "合格", "at": "2026-09-25T09:20:00+08:00", }, "tests": [("缺检测员：错误信息明确", "pm.expect(String(__json.message)).to.include('检测员必填')")], }, { "name": "缺字段失败后按批次确认零写入", "method": "GET", "path": "/school/pesticide/records?schoolId=1&q={{test_batch}}_NEG_NO_TESTER&page=1&pageSize=20", "tests": [ ("失败请求零写入：total=0", "pm.expect(Number(__json.data.total)).to.equal(0)"), ("失败请求零写入：items 为空", "pm.expect(__json.data.items).to.be.an('array').that.is.empty"), ], }, { "name": "跨校查询应 403", "method": "GET", "path": "/school/pesticide/records?schoolId=99999999&page=1&pageSize=20", "expected_code": 403, "expected_http": 403, "tests": [("跨校访问：错误信息明确且无业务数据", "pm.expect(String(__json.message)).to.match(/denied|拒绝|权限|学校/i); pm.expect(__json.data).to.not.exist")], }, { "name": "无 Token 查询应 401", "method": "GET", "path": "/school/pesticide/records?schoolId=1&page=1&pageSize=20", "expected_code": 401, "expected_http": 401, "no_auth": True, "tests": [("无 Token：明确返回 Missing token", "pm.expect(String(__json.message)).to.include('Missing token')")], }, {"name": "农残 SSE（长连接单测）", "method": "GET", "path": "/school/pesticide/stream?schoolId=1", "disabled": True}, ], }, "S04-消毒管理（双库真实写入）": { "prefix": "APIFOX_S04", "description": "学校端 S04 消毒管理双库真实写入场景：JSON 批量导入、新增、列表/详情回查和措施更新。SSE 长连接单独执行。", "operations": [ {"name": "批量导入消毒记录", "method": "POST", "path": "/school/disinfection/records/import?schoolId=1", "body": {"items": [{"canteenId": 1, "method": "紫外", "duration": 30, "temperature": 25.50, "items": "操作台、刀具", "responsible": "{{test_batch}}_李师傅", "at": "2026-09-25T10:00:00+08:00"}]}}, { "name": "新增消毒记录", "method": "POST", "path": "/school/disinfection/records", "body": { "schoolId": "1", "canteenId": 1, "method": "高温", "duration": 45, "temperature": 121.25, "items": "餐盘、汤勺", "responsible": "王建国-{{test_batch}}", "at": "2026-09-25T10:30:00+08:00", }, # 消毒记录使用 DS-000001 形式的业务编号，不是纯数字主键。 # 使用等价 JSONPath 避开公共的“正整数 ID”断言，再单独校验业务编号格式。 "extract": ("disinfection_id", "$['data']['id']"), "tests": [ ("新增成功：返回 DS-六位数字业务编号", "pm.expect(String(__json.data.id)).to.match(/^DS-\\d{6}$/)"), ("新增成功：学校与食堂关联正确", "pm.expect(Number(__json.data.schoolId)).to.equal(1); pm.expect(Number(__json.data.canteenId)).to.equal(1)"), ("新增成功：消毒方式与时长正确", "pm.expect(__json.data.method).to.equal('高温'); pm.expect(Number(__json.data.duration)).to.equal(45)"), ("新增成功：温度精确为 121.25℃", "pm.expect(Number(__json.data.temperature)).to.equal(121.25)"), ("新增成功：消毒对象与责任人正确", "pm.expect(__json.data.items).to.equal('餐盘、汤勺'); pm.expect(__json.data.responsible).to.equal('王建国-' + pm.variables.get('test_batch'))"), ], }, ("消毒列表回查", "GET", "/school/disinfection/records?schoolId=1&page=1&pageSize=20"), ("消毒详情回查", "GET", "/school/disinfection/records/{{disinfection_id}}"), {"name": "更新消毒措施", "method": "PATCH", "path": "/school/disinfection/records/{{disinfection_id}}/measure", "body": {"measure": "{{test_batch}} 温度与时长复核通过"}}, {"name": "消毒 SSE（长连接单测）", "method": "GET", "path": "/school/disinfection/stream?schoolId=1", "disabled": True}, ], }, "S05-废弃物管理（双库真实写入）": { "prefix": "APIFOX_S05", "description": "学校端 S05 废弃物双库真实写入场景：分类和记录新增、停用、列表/详情回查及 CSV，覆盖 DECIMAL(18,3)、真实中文业务字段与日期。主回归默认保留本轮入库证据；删除仅在独立专项中执行。", "operations": [ { "name": "新增餐厨有机垃圾分类", "method": "POST", "path": "/school/waste/categories", "body": {"name": "餐厨有机垃圾-{{test_batch}}"}, "extract": ("waste_category_id", "$.data.id"), "tests": [ ("分类名为本轮真实中文名称", "pm.expect(__json.data.name).to.equal('餐厨有机垃圾-' + pm.variables.get('test_batch'))"), ("新增分类默认启用", "pm.expect(Boolean(__json.data.enabled)).to.equal(true)"), ], }, { "name": "废弃物分类列表回查", "method": "GET", "path": "/school/waste/categories?enabled=all", "tests": [ ("列表包含本轮分类", "pm.expect(__json.data.some(function (item) { return Number(item.id) === Number(pm.variables.get('waste_category_id')) && item.name === '餐厨有机垃圾-' + pm.variables.get('test_batch'); })).to.equal(true)"), ], }, { "name": "停用本轮餐厨垃圾分类", "method": "PATCH", "path": "/school/waste/categories/{{waste_category_id}}/enable", "body": {"enabled": False}, "tests": [ ("停用返回分类 ID", "pm.expect(Number(__json.data.id)).to.equal(Number(pm.variables.get('waste_category_id')))"), ("停用状态为 false", "pm.expect(Boolean(__json.data.enabled)).to.equal(false)"), ], }, { "name": "新增餐厨废弃物称重记录", "method": "POST", "path": "/school/waste/records", "body": { "schoolId": 1, "canteenId": 1, "date": "2026-09-26", "category": "餐厨有机垃圾-{{test_batch}}", "amount": 12.345, "buyer": "海州市绿源资源回收有限公司-{{test_batch}}", "person": "赵明远", }, "extract": ("waste_record_id", "$.data.id"), "tests": [ ("写入日期为 2026-09-26", "pm.expect(String(__json.data.date).slice(0, 10)).to.equal('2026-09-26')"), ("写入数量精确为 12.345 kg", "pm.expect(Number(__json.data.amount)).to.equal(12.345)"), ("写入回收单位为本轮真实中文名称", "pm.expect(__json.data.buyer).to.equal('海州市绿源资源回收有限公司-' + pm.variables.get('test_batch'))"), ("写入收运人为赵明远", "pm.expect(__json.data.person).to.equal('赵明远')"), ], }, { "name": "废弃物记录列表回查", "method": "GET", "path": "/school/waste/records?schoolId=1&category=餐厨有机垃圾-{{test_batch}}&page=1&pageSize=20", "tests": [ ("列表返回本轮记录", "pm.expect(__json.data.items.some(function (item) { return Number(item.id) === Number(pm.variables.get('waste_record_id')); })).to.equal(true)"), ("列表总数至少一条", "pm.expect(Number(__json.data.total)).to.be.at.least(1)"), ], }, { "name": "导出废弃物 CSV 并核对业务字段", "method": "GET", "path": "/school/waste/records/export.csv?schoolId=1&category=餐厨有机垃圾-{{test_batch}}", "tests": [ ("CSV 包含分类名", "pm.expect(__json.data.csv).to.include('餐厨有机垃圾-' + pm.variables.get('test_batch'))"), ("CSV 包含回收单位", "pm.expect(__json.data.csv).to.include('海州市绿源资源回收有限公司-' + pm.variables.get('test_batch'))"), ("CSV 包含数量 12.345", "pm.expect(__json.data.csv).to.include('12.345')"), ], }, { "name": "废弃物详情回查", "method": "GET", "path": "/school/waste/records/{{waste_record_id}}", "tests": [ ("详情 ID 与新增记录一致", "pm.expect(Number(__json.data.id)).to.equal(Number(pm.variables.get('waste_record_id')))"), ("详情数量精确为 12.345 kg", "pm.expect(Number(__json.data.amount)).to.equal(12.345)"), ("详情中文字段未丢失", "pm.expect(__json.data.person).to.equal('赵明远'); pm.expect(__json.data.buyer).to.equal('海州市绿源资源回收有限公司-' + pm.variables.get('test_batch'))"), ], }, {"name": "删除本轮废弃物记录（独立删除专项）", "method": "DELETE", "path": "/school/waste/records/{{waste_record_id}}", "disabled": True}, {"name": "删除本轮废弃物分类（独立删除专项）", "method": "DELETE", "path": "/school/waste/categories/{{waste_category_id}}", "disabled": True}, ], }, "S06-设备安全（双库真实写入）": { "prefix": "APIFOX_S06", "description": "学校端 S06 设备安全双库真实写入场景：新增、列表、详情和更新，覆盖 JSON 数组、中文、空值和日期。接口没有删除能力，记录保留作数据库直查证据。", "operations": [ {"name": "新增设备安全检查", "method": "POST", "path": "/school/device-safety", "body": {"schoolId": 1, "canteenId": 1, "deviceName": "{{test_batch}}_蒸箱", "items": ["电源线", "接地", "温控器"], "result": "异常", "description": "温控器读数偶发漂移", "measures": "停机复核并更换传感器", "handler": "{{test_batch}}_陈工", "checkDate": "2026-09-25"}, "extract": ("device_safety_id", "$.data.id")}, ("设备安全列表回查", "GET", "/school/device-safety?schoolId=1&page=1&pageSize=20"), ("设备安全详情", "GET", "/school/device-safety/detail?id={{device_safety_id}}"), {"name": "更新设备安全检查", "method": "PATCH", "path": "/school/device-safety/{{device_safety_id}}", "body": {"result": "正常", "description": "{{test_batch}} 复检通过", "measures": None, "handler": None, "items": ["电源线", "接地", "温控器"], "checkDate": "2026-09-25"}}, ], }, } def normalized_operation(operation: Any) -> dict[str, Any]: if isinstance(operation, tuple): name, method, path = operation return {"name": name, "method": method, "path": path} return operation def flatten_requests(steps: list[dict[str, Any]]) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for step in steps: request = step.get("customHttpRequest") if request: result.append(request) result.extend(flatten_requests(step.get("children", []))) return result def request_has_auditable_evidence(request: dict[str, Any]) -> bool: scripts = [ str(processor.get("data", "")) for processor in request.get("postProcessors", []) if processor.get("type") == "customScript" ] return any("APIFOX_EVIDENCE" in script and "实际返回" in script for script in scripts) def enhance_s00(api: Apifox, scenario: dict[str, Any]) -> dict[str, Any]: """Bring the authentication bootstrap scenario under the same evidence rule.""" loop = scenario["steps"][0] loop.setdefault("parameters", {})["array"] = json.dumps( [MYSQL_BASE_URL, DM_BASE_URL], ensure_ascii=False, separators=(",", ":") ) loop["parameters"]["disableOnError"] = False children = loop["children"] if len(children) != 5: raise RuntimeError(f"S00: expected 5 enabled requests, got {len(children)}") processors = [ [ extractor("auth_token", "$.data.token"), assertion(200, "登录业务码 code=200"), evidence_script( expected_http=201, expected_code=200, tests=[("登录返回临时 Token（报告脱敏）", "pm.expect(__json.data.token).to.be.a('string').and.not.empty")], ), ], [ assertion(200, "当前用户业务码 code=200"), evidence_script(expected_http=200, expected_code=200), ], [ extractor("auth_token", "$.data.token"), assertion(200, "刷新 Token 业务码 code=200"), evidence_script( expected_http=201, expected_code=200, tests=[("刷新返回新 Token（报告脱敏）", "pm.expect(__json.data.token).to.be.a('string').and.not.empty")], ), ], [ assertion(200, "退出登录业务码 code=200"), evidence_script(expected_http=201, expected_code=200), ], [ assertion(401, "注销后业务码 code=401"), evidence_script( expected_http=401, expected_code=401, tests=[("注销后 Token 已失效", "pm.expect(String(__json.message)).to.match(/token|登录|失效|unauthorized/i)")], ), ], ] rebuilt_children: list[dict[str, Any]] = [] for number, (child, post_processors) in enumerate(zip(children, processors), start=1): rebuilt = copy.deepcopy(child) rebuilt["id"] = new_id() rebuilt["number"] = number rebuilt.pop("relatedId", None) request = strip_server_fields(rebuilt["customHttpRequest"]) request["relatedId"] = 0 request["postProcessors"] = post_processors rebuilt["customHttpRequest"] = request rebuilt_children.append(rebuilt) loop["children"] = rebuilt_children # S00 historically contained a duplicated /auth/me request outside the # database loop. It does not participate in the formal flow and produces # an unauditable extra row in exported reports, so keep only the loop. scenario["steps"] = [loop] scenario["description"] = ( f"学校端双库环境与认证入口：MySQL foodsafe_test={MYSQL_BASE_URL}，" f"达梦 WWWWEB={DM_BASE_URL}；覆盖登录、当前用户、刷新、退出和注销后 401。" "每步在报告断言中显示脱敏的实际返回摘要，并输出完整脱敏 APIFOX_EVIDENCE。" ) api.save(scenario) saved = api.scenario(int(scenario["id"])) active_requests = flatten_requests([saved["steps"][0]]) if not all(request_has_auditable_evidence(request) for request in active_requests): raise RuntimeError("S00: one or more active requests lack auditable evidence") return saved def main() -> None: api = Apifox() by_name = {item["name"]: item for item in api.scenarios()} source = by_name["S00-环境与认证（双库）"] source_data = api.scenario(int(source["id"])) source_data = enhance_s00(api, source_data) loop_template = source_data["steps"][0] login_template = loop_template["children"][0] request_template = loop_template["children"][1] scenario_ids: list[tuple[str, int]] = [(source["name"], int(source["id"]))] for ordering, (name, config) in enumerate(SCENARIOS.items(), start=1): current = by_name.get(name) scenario_id = int(current["id"]) if current else api.duplicate(int(source["id"])) scenario = api.scenario(scenario_id) scenario.update( { "name": name, "description": config["description"], "folderId": FOLDER_ID, "ordering": ordering * 10, } ) loop = copy.deepcopy(loop_template) loop["id"] = new_id() loop["number"] = 1 loop["children"] = [make_login(login_template, config["prefix"])] for index, raw_operation in enumerate(config["operations"], start=2): operation = normalized_operation(raw_operation) step = make_request(request_template, **operation) step["number"] = index loop["children"].append(step) scenario["steps"] = [loop] api.save(scenario) saved = api.scenario(scenario_id) requests = flatten_requests(saved["steps"]) business = [item for item in requests if item.get("path") != "{{base_url}}/auth/login"] expected = len(config["operations"]) if len(business) != expected: raise RuntimeError(f"{name}: expected {expected} business requests, got {len(business)}") if not all(request_has_auditable_evidence(request) for request in requests): raise RuntimeError(f"{name}: one or more requests lack auditable evidence") print(f"{name}: id={scenario_id}, operations={len(business)}, totalRequests={len(requests)}") scenario_ids.append((name, scenario_id)) unique_operations = { ("POST", "/auth/login"), ("GET", "/auth/me"), ("POST", "/auth/refresh"), ("POST", "/auth/logout"), } for config in SCENARIOS.values(): for raw_operation in config["operations"]: operation = normalized_operation(raw_operation) unique_operations.add((operation["method"].upper(), operation["path"].split("?", 1)[0])) suite = api.test_suite() suite["description"] = ( f"学校端 S00-S06 双库兼容一键串行入口，共 {len(unique_operations)} 个唯一接口。" "MySQL 仅允许 foodsafe_test，达梦仅允许 WWWWEB。每个普通 HTTP 步骤输出脱敏 APIFOX_EVIDENCE，" "并执行 HTTP、业务码、响应正文及模块字段断言。AI 事件夹具步骤和 SSE 长连接步骤默认关闭，按独立场景执行。" ) suite["items"] = [ { "id": new_id(), "name": name, "type": "STATIC_TEST_SCENARIO", "testScenarios": [{"id": scenario_id, "options": {}}], "options": {}, } for name, scenario_id in scenario_ids ] suite.setdefault("options", {})["runMode"] = "serial" api.save_test_suite(suite) saved_suite = api.test_suite() linked = [item["testScenarios"][0]["id"] for item in saved_suite["items"]] if linked != [scenario_id for _, scenario_id in scenario_ids]: raise RuntimeError(f"test suite references do not match: {linked}") print(f"一键测试-安全入口: scenarios={len(linked)}, suiteId={TEST_SUITE_ID}") if __name__ == "__main__": main()