#!/usr/bin/env python3 """Create/update the public data-screen dual-database Apifox baseline. This scenario intentionally uses no Authorization header. It proves that the read-only dashboard endpoints work against MySQL foodsafe_test and DM WWWWEB, while the protected /school routes remain covered by the school-side suite. """ from __future__ import annotations import copy import importlib.util import json from datetime import datetime from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent SHARED_PATH = HERE / "sync_school_s00_s06_scenarios.py" SPEC = importlib.util.spec_from_file_location("school_sync", SHARED_PATH) if SPEC is None or SPEC.loader is None: raise RuntimeError(f"cannot load shared Apifox helper: {SHARED_PATH}") SHARED = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(SHARED) Apifox = SHARED.Apifox MYSQL_BASE_URL = SHARED.MYSQL_BASE_URL DM_BASE_URL = SHARED.DM_BASE_URL PROJECT_ID = SHARED.PROJECT_ID assertion = SHARED.assertion evidence_script = SHARED.evidence_script make_request = SHARED.make_request new_id = SHARED.new_id SCENARIO_NAME = "D00-数据大屏公共读取（双库真实数据）" SUITE_NAME = "数据大屏-双库一键回归" FOLDER_NAME = "大屏端-双库兼容" BACKUP_DIR = HERE.parent / "backups" def screen_folder_id(api: Any) -> int: tree = api.call(f"/api/v1/projects/{PROJECT_ID}/test-scenario/tree-list")["data"] folder = next( ( item for item in tree.get("testScenarioFolders", []) if item.get("name") == FOLDER_NAME ), None, ) if folder is None: raise RuntimeError(f"Apifox 缺少场景目录：{FOLDER_NAME}") return int(folder["id"]) def test(name: str, statement: str) -> tuple[str, str]: return name, statement OPERATIONS: list[dict[str, Any]] = [ { "name": "网关健康检查", "method": "GET", "path": "/health", "tests": [ test("服务健康状态 ok=true", "pm.expect(__json.data.ok).to.equal(true)"), test("服务标识为 gateway-service", "pm.expect(__json.data.service).to.equal('gateway-service')"), ], }, { "name": "今日菜品留样-2026年9月26日", "method": "GET", "path": "/screen/analytics/sampling/records?schoolId=1&canteenId=1&start=2026-09-26&end=2026-09-27&page=1&pageSize=1000", "tests": [ test("留样分页结构完整", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.total)).to.be.at.least(3)"), test("包含三条真实中文午餐留样", "const names = __json.data.items.map(function (x) { return x.sample; }); ['红烧鸡块（午餐留样）','蒜蓉油麦菜（午餐留样）','番茄炒鸡蛋（午餐留样）'].forEach(function (x) { pm.expect(names).to.include(x); })"), test("留样展示字段完整且不泄露手机号", "__json.data.items.forEach(function (x) { pm.expect(x).to.have.all.keys('id','sample','mealTypeName','exception','at','by'); pm.expect(x).to.not.have.property('phone'); })"), ], }, { "name": "菜品留样-2026年9月28日", "method": "GET", "path": "/screen/analytics/sampling/records?schoolId=1&canteenId=1&start=2026-09-28&end=2026-09-29&page=1&pageSize=1000", "tests": [ test("9月28日留样恰有三条真实记录", "pm.expect(__json.data.items).to.be.an('array').with.lengthOf(3); pm.expect(Number(__json.data.total)).to.equal(3)"), test("包含土豆烧牛肉、小白菜和紫菜蛋花汤", "const names = __json.data.items.map(function (x) { return x.sample; }); ['土豆烧牛肉（午餐留样）','清炒小白菜（午餐留样）','紫菜蛋花汤（午餐留样）'].forEach(function (x) { pm.expect(names).to.include(x); })"), test("留样人员为真实中文姓名", "const people = __json.data.items.map(function (x) { return x.by; }); pm.expect(people).to.include('李芳'); pm.expect(people).to.include('王海燕')"), ], }, { "name": "农残检测-2026年9月26日", "method": "GET", "path": "/screen/analytics/pesticide/records?schoolId=1&canteenId=1&start=2026-09-26&end=2026-09-27&page=1&pageSize=1000", "tests": [ test("农残分页结构完整", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.total)).to.be.at.least(3)"), test("菠菜、西红柿合格且韭菜不合格", "const rows = __json.data.items; [['菠菜','合格'],['西红柿','合格'],['韭菜','不合格']].forEach(function (pair) { pm.expect(rows.some(function (x) { return x.sample === pair[0] && x.result === pair[1]; })).to.equal(true); })"), test("农残展示字段最小化", "__json.data.items.forEach(function (x) { pm.expect(x).to.have.all.keys('id','sample','result','at'); pm.expect(x).to.not.have.property('phone'); })"), ], }, { "name": "农残检测-2026年9月28日", "method": "GET", "path": "/screen/analytics/pesticide/records?schoolId=1&canteenId=1&start=2026-09-28&end=2026-09-29&page=1&pageSize=1000", "tests": [ test("9月28日农残恰有三条真实记录", "pm.expect(__json.data.items).to.be.an('array').with.lengthOf(3); pm.expect(Number(__json.data.total)).to.equal(3)"), test("生菜、黄瓜合格且豇豆不合格", "const rows = __json.data.items; [['生菜','合格'],['黄瓜','合格'],['豇豆','不合格']].forEach(function (pair) { pm.expect(rows.some(function (x) { return x.sample === pair[0] && x.result === pair[1]; })).to.equal(true); })"), ], }, { "name": "今日晨检记录", "method": "GET", "path": "/screen/analytics/morning-checks?schoolId=1&canteenId=1&start=2026-09-26&end=2026-09-27&page=1&pageSize=1000", "tests": [ test("晨检分页结构完整", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.total)).to.be.at.least(0); pm.expect(Number(__json.data.page)).to.equal(1)"), test("晨检字段可展示且不泄露手机号", "__json.data.items.forEach(function (x) { pm.expect(x).to.have.property('id'); pm.expect(x).to.have.property('staff'); pm.expect(x).to.have.property('result'); pm.expect(x).to.have.property('at'); pm.expect(x).to.not.have.property('phone'); })"), ], }, { "name": "从业人员公示列表", "method": "GET", "path": "/screen/analytics/personnel?schoolId=1&canteenId=1&page=1&pageSize=1000", "tests": [ test("人员分页结构完整且列表非空", "pm.expect(__json.data.items).to.be.an('array').that.is.not.empty; pm.expect(Number(__json.data.total)).to.be.at.least(1)"), test("人员展示字段完整", "__json.data.items.forEach(function (x) { pm.expect(x).to.have.all.keys('id','name','status','facePhotoUrl','healthCertFrontUrl'); })"), test("人员接口不泄露手机号和证件号", "__json.data.items.forEach(function (x) { pm.expect(x).to.not.have.property('phone'); pm.expect(x).to.not.have.property('idCard'); pm.expect(x).to.not.have.property('password'); })"), ], }, { "name": "AI 预警明细-2026年9月26日", "method": "GET", "path": "/screen/analytics/ai/events?schoolId=1&canteenId=1&start=2026-09-26&end=2026-09-27&page=1&pageSize=1000", "tests": [ test("AI 明细分页结构完整", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.total)).to.be.at.least(3)"), test("包含口罩、帽子、抽烟三类真实预警", "const types = __json.data.items.map(function (x) { return x.type; }); ['口罩识别','帽子识别','抽烟识别'].forEach(function (x) { pm.expect(types).to.include(x); })"), test("包含当天三条未处置真实预警", "const wanted = {'口罩识别':'2026-09-26 10:15:00','帽子识别':'2026-09-26 10:42:00','抽烟识别':'2026-09-26 14:05:00'}; Object.keys(wanted).forEach(function (type) { pm.expect(__json.data.items.some(function (x) { return x.type === type && x.detectedAt === wanted[type] && x.status === 'OPEN'; })).to.equal(true); })"), ], }, { "name": "AI 预警明细-2026年9月28日", "method": "GET", "path": "/screen/analytics/ai/events?schoolId=1&canteenId=1&start=2026-09-28&end=2026-09-29&page=1&pageSize=1000", "tests": [ test("AI 明细分页结构完整", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.total)).to.be.at.least(3)"), test("包含厨师服、打手机、老鼠三类真实预警", "const types = __json.data.items.map(function (x) { return x.type; }); ['厨师服识别','打手机识别','老鼠识别'].forEach(function (x) { pm.expect(types).to.include(x); })"), test("包含9月28日三条未处置真实预警", "const wanted = {'厨师服识别':'2026-09-28 09:20:00','打手机识别':'2026-09-28 11:12:00','老鼠识别':'2026-09-28 02:35:00'}; Object.keys(wanted).forEach(function (type) { pm.expect(__json.data.items.some(function (x) { return x.type === type && x.detectedAt === wanted[type] && x.status === 'OPEN'; })).to.equal(true); })"), ], }, { "name": "AI 预警类型统计-9月26日至28日", "method": "GET", "path": "/screen/analytics/ai/pie?schoolId=1&canteenId=1&start=2026-09-26&end=2026-09-29", "tests": [ test("AI 统计总数与分类数组存在", "pm.expect(Number(__json.data.total)).to.be.at.least(6); pm.expect(__json.data.items).to.be.an('array')"), test("真实预警分类均进入统计", "const names = __json.data.items.filter(function (x) { return Number(x.count) > 0; }).map(function (x) { return x.name; }); ['未戴口罩','未戴帽','未穿工作服','吸烟','打电话','鼠患'].forEach(function (x) { pm.expect(names).to.include(x); })"), ], }, { "name": "人员健康汇总", "method": "GET", "path": "/screen/analytics/personnel-health?schoolId=1&canteenId=1", "tests": [ test("人员健康汇总归属学校1", "pm.expect(Number(__json.data.schoolId)).to.equal(1)"), test("人员健康统计字段均为非负数", "['expectedTotal','checkedUnique','abnormalTotal'].forEach(function (key) { pm.expect(Number(__json.data[key])).to.be.at.least(0); })"), test("健康统计时间窗口完整", "pm.expect(__json.data.window.start).to.be.a('string').and.not.empty; pm.expect(__json.data.window.end).to.be.a('string').and.not.empty"), ], }, { "name": "设备列表", "method": "GET", "path": "/screen/analytics/devices?schoolId=1", "tests": [test("设备列表为数组", "pm.expect(__json.data).to.be.an('array')")], }, { "name": "设备类型字典", "method": "GET", "path": "/screen/analytics/devices/types", "tests": [ test("设备类型为数组", "pm.expect(__json.data).to.be.an('array')"), test("包含摄像头、传感器等核心类型", "['CAMERA','SENSOR','SMOKE'].forEach(function (x) { pm.expect(__json.data).to.include(x); })"), ], }, { "name": "设备状态字典", "method": "GET", "path": "/screen/analytics/devices/statuses", "tests": [ test("设备状态为数组", "pm.expect(__json.data).to.be.an('array')"), test("包含在线、离线和故障", "['ONLINE','OFFLINE','FAULT'].forEach(function (x) { pm.expect(__json.data).to.include(x); })"), ], }, { "name": "明厨亮灶摄像头列表", "method": "GET", "path": "/screen/analytics/cameras?schoolId=1&canteenId=1", "tests": [ test("摄像头列表为非空数组", "pm.expect(__json.data).to.be.an('array').that.is.not.empty"), test("摄像头归属第一食堂且字段完整", "pm.expect(__json.data.some(function (x) { return Number(x.canteenId) === 1 && String(x.name || '').length > 0 && String(x.channelId || '').length > 0; })).to.equal(true)"), ], }, ] def backup(api: Any) -> Path: BACKUP_DIR.mkdir(parents=True, exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") target = BACKUP_DIR / f"screen-d00-before-sync_{stamp}.json" payload = { "generatedAt": datetime.now().astimezone().isoformat(), "projectId": PROJECT_ID, "scenarios": api.scenarios(), "testSuites": api.call(f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites")["data"], } target.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return target def find_or_create_suite(api: Any, scenario_id: int) -> dict[str, Any]: suites = api.call(f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites")["data"] current = next((item for item in suites if item.get("name") == SUITE_NAME), None) item = { "id": new_id(), "name": SCENARIO_NAME, "type": "STATIC_TEST_SCENARIO", "testScenarios": [{"id": scenario_id, "options": {}}], "options": {}, } description = ( "数据大屏无登录公共读取双库一键回归。当前仅纳入已在 MySQL foodsafe_test 与达梦 WWWWEB " "实际返回 200 的接口；每步包含字段级断言、可见实际返回摘要和脱敏 APIFOX_EVIDENCE。" ) if current: suite = api.call( f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites/{int(current['id'])}" )["data"] suite.update({"name": SUITE_NAME, "description": description, "items": [item]}) suite.setdefault("options", {})["runMode"] = "serial" api.call( f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites/{int(suite['id'])}", "PUT", suite, ) return api.call( f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites/{int(suite['id'])}" )["data"] payload = { "name": SUITE_NAME, "description": description, "priority": 2, "ordering": 20, "folderId": 0, "items": [item], "options": {"runMode": "serial", "runnerId": 0}, "tags": [], } created = api.call( f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites", "POST", payload, )["data"] suite_id = int(created["id"] if isinstance(created, dict) else created) return api.call(f"/api/v1/projects/{PROJECT_ID}/api-test/test-suites/{suite_id}")["data"] def main() -> None: api = Apifox() backup_path = backup(api) folder_id = screen_folder_id(api) by_name = {item["name"]: item for item in api.scenarios()} source = by_name["S00-环境与认证（双库）"] source_data = api.scenario(int(source["id"])) loop_template = source_data["steps"][0] request_template = loop_template["children"][1] current = by_name.get(SCENARIO_NAME) scenario_id = int(current["id"]) if current else api.duplicate(int(source["id"])) scenario = api.scenario(scenario_id) scenario.update( { "name": SCENARIO_NAME, "description": ( f"数据大屏无登录公共读取基线：MySQL foodsafe_test={MYSQL_BASE_URL}，" f"达梦 WWWWEB={DM_BASE_URL}。覆盖健康检查、AI 真实预警明细/统计、人员健康、" "设备字典和明厨亮灶摄像头；每步显示关键返回值并输出脱敏完整响应证据。" ), "folderId": folder_id, "ordering": 200, } ) loop = copy.deepcopy(loop_template) loop["id"] = new_id() loop["number"] = 1 loop.setdefault("parameters", {})["array"] = json.dumps( [MYSQL_BASE_URL, DM_BASE_URL], ensure_ascii=False, separators=(",", ":") ) loop["parameters"]["disableOnError"] = False loop["children"] = [] for number, operation in enumerate(OPERATIONS, start=1): step = make_request(request_template, no_auth=True, **operation) step["number"] = number step["customHttpRequest"]["preProcessors"] = [ { "type": "customScript", "data": "pm.variables.set('base_url', pm.variables.get('$.1.element'));", "defaultEnable": False, "enable": True, "executionTiming": "prerequest", }, SHARED.placeholder(), ] loop["children"].append(step) scenario["steps"] = [loop] api.save(scenario) saved = api.scenario(scenario_id) requests = SHARED.flatten_requests(saved["steps"]) if len(requests) != len(OPERATIONS): raise RuntimeError(f"expected {len(OPERATIONS)} requests, got {len(requests)}") if not all(SHARED.request_has_auditable_evidence(request) for request in requests): raise RuntimeError("one or more requests lack auditable response evidence") if any(request.get("auth") for request in requests): raise RuntimeError("public screen scenario unexpectedly contains request auth") suite = find_or_create_suite(api, scenario_id) linked = [ int(item["testScenarios"][0]["id"]) for item in suite.get("items", []) if item.get("testScenarios") ] if linked != [scenario_id]: raise RuntimeError(f"screen suite references do not match: {linked}") print(f"backup={backup_path}") print(f"scenario={SCENARIO_NAME}, id={scenario_id}, operations={len(requests)}, dualExecutions={len(requests) * 2}") print(f"suite={SUITE_NAME}, id={suite['id']}, scenarios={len(linked)}, runMode={suite.get('options', {}).get('runMode')}") if __name__ == "__main__": main()