#!/usr/bin/env python3 """Create/update the S13 local IoT and bright-kitchen Apifox scenario. Only local database-backed configuration and cached read APIs are used. The scenario never calls a real camera/AI box, never refreshes upstream devices, never uploads media, and never sends MQTT/OSS traffic. """ from __future__ import annotations import copy import importlib.util from pathlib import Path BASE_SCRIPT = Path(__file__).resolve().parent / "sync_school_s00_s06_scenarios.py" SPEC = importlib.util.spec_from_file_location("school_s00_s06_sync", BASE_SCRIPT) if SPEC is None or SPEC.loader is None: raise RuntimeError(f"Cannot load shared Apifox helpers from {BASE_SCRIPT}") shared = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(shared) NAME = "S13-物联网配置与明厨亮灶（双库真实写入）" PREFIX = "APIFOX_S13" DESCRIPTION = ( "学校端 S13 本地物联网配置与明厨亮灶场景：AI 分析盒、本地通道的新增/编辑/列表回查，" "以及本地回放缓存、AI 事件、陌生人列表与 CSV 读取。正向数据使用真实中文业务语义，" "主回归保留两库入库证据；删除步骤默认关闭。不调用真实摄像头/AI 盒，不刷新上游设备，" "不上传图片视频，不发送 MQTT/OSS 数据。" ) OPERATIONS = [ { "name": "读取本地 AI 分析盒配置基线", "method": "GET", "path": "/bright-local/aibox-boxes/full?schoolId=1", "tests": [("分析盒基线返回数组", "pm.expect(__json.data).to.be.an('array')")], }, { "name": "新增康源实验学校食安分析盒", "method": "POST", "path": "/bright-local/aibox-boxes", "body": { "schoolId": 1, "canteenId": 1, "aiboxName": "康源实验学校第一食堂食安分析盒-{{business_batch}}", "aiboxBaseurl": "http://127.0.0.1:18080", "aiboxAccount": "kangyuan_foodsafe_{{batch_tail}}", "aiboxPassword": "{{aibox_password}}", }, "extract": ("aibox_box_id", "$.data.id"), "tests": [("分析盒新增返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)")], }, { "name": "回查新增分析盒与口令脱敏", "method": "GET", "path": "/bright-local/aibox-boxes/full?schoolId=1", "tests": [ ("分析盒列表包含本轮记录", "const id=Number(pm.variables.get('aibox_box_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.aiboxName).to.equal('康源实验学校第一食堂食安分析盒-' + pm.variables.get('business_batch')); pm.expect(Number(row.canteenId)).to.equal(1)"), ("分析盒口令仅返回脱敏值", "const id=Number(pm.variables.get('aibox_box_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row.passwordMasked).to.equal('******'); pm.expect(row).to.not.have.property('aiboxPassword')"), ], }, { "name": "编辑分析盒名称和运维账号", "method": "PATCH", "path": "/bright-local/aibox-boxes?id={{aibox_box_id}}", "body": { "aiboxName": "康源实验学校第一食堂食安分析盒（已核验）-{{business_batch}}", "aiboxAccount": "kangyuan_foodsafe_ops_{{batch_tail}}", }, "tests": [("分析盒编辑返回 ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "回查分析盒编辑结果", "method": "GET", "path": "/bright-local/aibox-boxes/full?schoolId=1", "tests": [ ("分析盒最终中文名称正确", "const id=Number(pm.variables.get('aibox_box_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.aiboxName).to.equal('康源实验学校第一食堂食安分析盒（已核验）-' + pm.variables.get('business_batch'))"), ("分析盒最终账号与本轮批次一致", "const id=Number(pm.variables.get('aibox_box_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row.aiboxAccount).to.equal('kangyuan_foodsafe_ops_' + pm.variables.get('batch_tail'))"), ], }, { "name": "回查分析盒下拉选项", "method": "GET", "path": "/bright-local/aibox-boxes?schoolId=1", "tests": [("下拉选项包含本轮分析盒", "const id=Number(pm.variables.get('aibox_box_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.name).to.include('已核验')")], }, { "name": "新增第一食堂后厨本地通道", "method": "POST", "path": "/bright-local/channels", "body": { "schoolId": 1, "canteenId": 1, "unitName": "康源实验学校", "aiboxId": "{{aibox_box_id}}", "channelId": "KY-CANTEEN-01-{{business_batch}}", "channelName": "第一食堂后厨操作间", "faislenationCode": "KY-AISLE-01-{{business_batch}}", "enabled": False, }, "extract": ("aibox_channel_row_id", "$.data.id"), "tests": [("本地通道新增返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)")], }, { "name": "按通道编号回查新增配置", "method": "GET", "path": "/bright-local/channels?schoolId=1&channelId=KY-CANTEEN-01-{{business_batch}}&page=1&pageSize=20", "tests": [ ("通道列表仅命中本轮记录", "pm.expect(__json.data.items).to.have.lengthOf(1); pm.expect(Number(__json.data.items[0].id)).to.equal(Number(pm.variables.get('aibox_channel_row_id')))"), ("通道与分析盒、食堂关系正确", "const row=__json.data.items[0]; pm.expect(Number(row.aiboxId)).to.equal(Number(pm.variables.get('aibox_box_id'))); pm.expect(Number(row.canteenId)).to.equal(1); pm.expect(row.enabled).to.equal(false)"), ], }, { "name": "编辑本地通道业务名称", "method": "PATCH", "path": "/bright-local/channels?id={{aibox_channel_row_id}}", "body": { "channelName": "第一食堂后厨操作间（设备联调待启用）", "enabled": False, }, "tests": [("通道编辑返回 ok=true", "pm.expect(__json.data.ok).to.equal(true); pm.expect(Number(__json.data.schoolId)).to.equal(1)")], }, { "name": "回查本地通道最终状态", "method": "GET", "path": "/bright-local/channels?schoolId=1&channelId=KY-CANTEEN-01-{{business_batch}}&page=1&pageSize=20", "tests": [ ("通道最终中文名称正确", "pm.expect(__json.data.items[0].channelName).to.equal('第一食堂后厨操作间（设备联调待启用）')"), ("通道保持停用防止真实设备出站", "pm.expect(__json.data.items[0].enabled).to.equal(false)"), ], }, { "name": "读取本轮通道的本地回放缓存", "method": "GET", "path": "/bright-local/playback?schoolId=1&channelId=KY-CANTEEN-01-{{business_batch}}&date=20260926&page=1&pageSize=20&canteenId=1", "tests": [ ("回放缓存仅读：返回分页结构", "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)"), ], }, { "name": "模拟 AI 盒子上报未戴口罩事件（无媒体）", "method": "POST", "path": "/openApi/checkRecord/reportCheckRecords", "body": [ { "fid": "{{local_event_fid}}", "analysisTypeNumber": "AI-PPE-MASK", "fcreatetime": "{{event_time_ms}}", "faisleinfoid": "{{aibox_channel_row_id}}", "faisleinfoname": "第一食堂后厨操作间（本地模拟入站）", "faislename": "第一食堂后厨操作间（本地模拟入站）", "faislenationCode": "KY-AISLE-01-{{business_batch}}", "fanlysistypeid": "口罩识别", "fcamerasn": "KY-CANTEEN-01-{{business_batch}}", "fcompany": "康源实验学校", "fnetgateno": "{{aibox_box_id}}", "fscore": 97, "fismanualjudge": 0, } ], "expected_code": 1, "expected_http": 201, "_special": "check-record-ingest", }, { "name": "回查本轮本地 AI 事件分页", "method": "GET", "path": "/bright-local/ai/events?schoolId=1&type=口罩识别&page=1&pageSize=20", "tests": [ ("AI 事件分页包含本轮 FID", "pm.expect(__json.data.items.some(function(x){return x.id===pm.variables.get('local_event_fid');})).to.equal(true)"), ("本轮事件映射为未戴口罩且状态为 OPEN", "const row=__json.data.items.find(function(x){return x.id===pm.variables.get('local_event_fid');}); pm.expect(row.typeLabel).to.equal('未戴口罩'); pm.expect(row.status).to.equal('OPEN')"), ], }, { "name": "读取本轮本地 AI 事件详情", "method": "GET", "path": "/bright-local/ai/events/detail?id={{local_event_fid}}", "tests": [ ("事件详情 FID 与本轮一致", "pm.expect(__json.data.fid).to.equal(pm.variables.get('local_event_fid'))"), ("处置前事件无历史日志", "pm.expect(__json.data.logs).to.be.an('array').that.is.empty"), ], }, { "name": "处置本轮未戴口罩事件", "method": "POST", "path": "/bright-local/ai/events/handle", "body": { "eventId": "{{local_event_fid}}", "decision": "已现场纠正", "note": "已提醒后厨人员规范佩戴口罩，并由值班主管现场复核。", "evidence": [], "handledBy": "学校食品安全管理员", }, "tests": [ ("事件处置返回 ok=true", "pm.expect(__json.data.ok).to.equal(true)"), ("事件处置学校范围为 1", "pm.expect(Number(__json.data.schoolId)).to.equal(1)"), ], }, { "name": "回查本轮事件处置结果", "method": "GET", "path": "/bright-local/ai/events/detail?id={{local_event_fid}}", "tests": [ ("事件处置后状态为 CLOSED", "pm.expect(__json.data.status).to.equal('CLOSED'); pm.expect(__json.data.measure).to.equal('已现场纠正')"), ("事件处置日志完整", "pm.expect(__json.data.logs).to.have.lengthOf(1); pm.expect(__json.data.logs[0].decision).to.equal('已现场纠正'); pm.expect(__json.data.logs[0].note).to.equal('已提醒后厨人员规范佩戴口罩，并由值班主管现场复核。'); pm.expect(__json.data.logs[0].by).to.equal('学校食品安全管理员')"), ], }, { "name": "导出本地 AI 事件 CSV", "method": "GET", "path": "/bright-local/ai/events/export.csv?schoolId=1&type=口罩识别&status=CLOSED", "tests": [ ("AI 事件 CSV 包含中文表头", "pm.expect(__json.data.csv).to.include('通道名称'); pm.expect(__json.data.csv).to.include('违规类型')"), ("AI 事件 CSV 包含本轮 FID 与处置状态", "pm.expect(__json.data.csv).to.include(pm.variables.get('local_event_fid')); pm.expect(__json.data.csv).to.include('已处理')"), ], }, { "name": "读取本地陌生人记录分页", "method": "GET", "path": "/bright-local/strangers?schoolId=1&page=1&pageSize=5", "tests": [("陌生人记录仅读：返回分页结构", "pm.expect(__json.data.items).to.be.an('array'); pm.expect(Number(__json.data.page)).to.equal(1); pm.expect(Number(__json.data.pageSize)).to.equal(5)")], }, { "name": "导出本地陌生人 CSV", "method": "GET", "path": "/bright-local/strangers/export.csv?schoolId=1", "tests": [("陌生人 CSV 包含中文表头", "pm.expect(__json.data.csv).to.include('AIBox名称'); pm.expect(__json.data.csv).to.include('发生时间')")], }, {"name": "删除本轮通道（独立删除专项）", "method": "DELETE", "path": "/bright-local/channels?id={{aibox_channel_row_id}}", "disabled": True}, {"name": "删除本轮分析盒（独立删除专项）", "method": "DELETE", "path": "/bright-local/aibox-boxes?id={{aibox_box_id}}", "disabled": True}, ] def main() -> None: api = shared.Apifox() 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] login_template = loop_template["children"][0] request_template = loop_template["children"][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": DESCRIPTION, "folderId": shared.FOLDER_ID, "ordering": 140}) loop = copy.deepcopy(loop_template) loop["id"] = shared.new_id() loop["number"] = 1 login = shared.make_login(login_template, PREFIX) login_pre = login["customHttpRequest"]["preProcessors"][0]["data"] login["customHttpRequest"]["preProcessors"][0]["data"] = ( login_pre + "const __stamp = '20260926' + String(Date.now()).slice(-6);\n" + "const __tail = String(Date.now()).slice(-8);\n" + "pm.variables.set('business_batch', __stamp);\n" + "pm.variables.set('batch_tail', __tail);\n" + "pm.variables.set('aibox_password', 'ShianBox-' + __tail + '-Aa1!');\n" + "pm.variables.set('local_event_fid', 'S13-' + __stamp);\n" + "pm.variables.set('event_time_ms', String(Date.now()));\n" ) loop["children"] = [login] for number, operation in enumerate(OPERATIONS, start=2): request_operation = dict(operation) special = request_operation.pop("_special", None) step = shared.make_request(request_template, **request_operation) if special == "check-record-ingest": request = step["customHttpRequest"] request["postProcessors"] = [ shared.assertion(1, "设备入站业务码 code=1"), shared.custom_script( "const __text=pm.response.text(); const __json=pm.response.json();\n" "pm.test('HTTP 状态码 = 201', function(){pm.expect(pm.response.code).to.equal(201);});\n" "pm.test('设备入站业务码 code = 1', function(){pm.expect(String(__json.code)).to.equal('1');});\n" "pm.test('设备入站 saved = 1', function(){pm.expect(Number(__json.saved)).to.equal(1);});\n" "pm.test('请求不含图片视频或外部地址', function(){const b=pm.request.body.raw||pm.request.body.toString(); pm.expect(b).to.not.include('imgBase64'); pm.expect(b).to.not.include('fvideopath'); pm.expect(b).to.not.match(/https?:\\/\\/(?!127\\.0\\.0\\.1)/);});\n" "pm.test('实际返回｜code=' + String(__json.code) + '，saved=' + String(__json.saved), function(){pm.expect(Number(__json.saved)).to.equal(1);});\n" "console.log('APIFOX_EVIDENCE ' + JSON.stringify({step:pm.info.requestName,method:pm.request.method,url:pm.request.url.toString(),httpStatus:pm.response.code,responseTimeMs:pm.response.responseTime,requestBody:JSON.parse(pm.request.body.raw),responseBody:__json}));" ), ] step["number"] = number loop["children"].append(step) scenario["steps"] = [loop] api.save(scenario) saved = api.scenario(scenario_id) requests = shared.flatten_requests(saved["steps"]) business = requests[1:] if len(business) != len(OPERATIONS): raise RuntimeError(f"{NAME}: expected {len(OPERATIONS)} business requests, got {len(business)}") if not all(shared.request_has_auditable_evidence(request) for request in requests): raise RuntimeError(f"{NAME}: one or more requests lack auditable evidence") delete_steps = [ child for child in saved["steps"][0]["children"] if child.get("customHttpRequest", {}).get("method", "").upper() == "DELETE" ] if len(delete_steps) != 2 or not all(step.get("disable") is True for step in delete_steps): raise RuntimeError("S13 main regression must keep both DELETE steps disabled") suite = api.test_suite() existing = [item for item in suite.get("items", []) if item.get("name") != NAME] existing.append({ "id": shared.new_id(), "name": NAME, "type": "STATIC_TEST_SCENARIO", "testScenarios": [{"id": scenario_id, "options": {}}], "options": {}, }) suite["items"] = existing suite["description"] = ( "学校端 S00-S13 已解除门禁部分的双库兼容一键串行入口。MySQL 仅允许 foodsafe_test，" "达梦仅允许 WWWWEB。S12 公告和食堂配置已完成 DBCOMPAT-0021 修复及双库重启读回复测；" "S13 仅执行本地数据库配置与缓存读取，禁止真实设备、视频、MQTT、OSS 和第三方出站。" ) 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 scenario_id not in linked: raise RuntimeError("S13 was not linked into the one-click test suite") print(f"{NAME}: id={scenario_id}, operations={len(business)}, totalRequests={len(requests)}") print(f"一键测试-安全入口: scenarios={len(linked)}, suiteId={shared.TEST_SUITE_ID}") if __name__ == "__main__": main()