#!/usr/bin/env python3 """Create or update the auditable S08 foods/operators Apifox scenario.""" from __future__ import annotations import copy import importlib.util from pathlib import Path BASE_SCRIPT = Path(__file__).with_name("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 = "S08-菜品与留样人员（双库真实写入）" PREFIX = "APIFOX_S08" DESCRIPTION = ( "学校端 S08 菜品与留样人员双库可审计场景：菜品新增、存在性、列表、更新、批量导入、CSV、" "64/65 字符边界、人员新增、留样人员新增/详情/更新和权限反例。使用学校 1、无设备序列号的" "食堂 2，禁止真实 MQTT/第三方出站。删除仅针对本场景 marker，默认关闭，需运行前即时确认。" ) CODE_65 = "X" * 65 OPERATIONS = [ { "name": "读取菜品列表基线", "method": "GET", "path": "/school/foods?schoolId=1&canteenId=2&status=all&page=1&pageSize=20", "tests": [ ("菜品基线：items 为数组", "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(20)"), ], }, { "name": "新增中文菜品并返回 ID", "method": "POST", "path": "/school/foods", "body": { "schoolId": 1, "canteenId": 2, "code": "{{test_batch}}_F01", "name": "{{test_batch}}_宫保鸡丁", "imageUrl": None, "remark": "中文备注：少盐少油", "enabled": True, "by": "自动化测试员", }, "extract": ("food_id", "$.data.id"), "tests": [ ("菜品创建：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)"), ], }, { "name": "检查刚创建菜品名称已存在", "method": "GET", "path": "/school/foods/exists?schoolId=1&canteenId=2&name={{test_batch}}_宫保鸡丁", "tests": [("存在性：exists=true", "pm.expect(__json.data.exists).to.equal(true)")], }, { "name": "按 marker 回查新增菜品", "method": "GET", "path": "/school/foods?schoolId=1&canteenId=2&q={{test_batch}}&status=all&page=1&pageSize=20", "tests": [ ("菜品回查：包含创建 ID", "const id=Number(pm.variables.get('food_id')); pm.expect(__json.data.items.some(function(x){return Number(x.id)===id;})).to.equal(true)"), ("菜品回查：中文、启用、NULL 图片正确", "const id=Number(pm.variables.get('food_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(row.name).to.equal(pm.variables.get('test_batch') + '_宫保鸡丁'); pm.expect(Boolean(row.enabled)).to.equal(true); pm.expect(row.imageUrl === null || row.image_url === null || row.imageUrl === '').to.equal(true)"), ], }, { "name": "更新菜品中文、备注、NULL 图片和停用状态", "method": "PATCH", "path": "/school/foods/{{food_id}}", "body": { "schoolId": 1, "canteenId": 2, "name": "{{test_batch}}_鱼香肉丝", "imageUrl": None, "remark": "中文备注：复核配方后停用", "enabled": False, "by": "自动化复核员", }, "tests": [("菜品更新：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "更新后回查菜品已持久化", "method": "GET", "path": "/school/foods?schoolId=1&canteenId=2&q={{test_batch}}&status=disabled&page=1&pageSize=20", "tests": [ ("更新回查：名称和备注正确", "const id=Number(pm.variables.get('food_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.name).to.equal(pm.variables.get('test_batch') + '_鱼香肉丝'); pm.expect(row.remark).to.equal('中文备注：复核配方后停用')"), ("更新回查：enabled=false 且图片仍为空", "const id=Number(pm.variables.get('food_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(Boolean(row.enabled)).to.equal(false); pm.expect(row.imageUrl === null || row.image_url === null || row.imageUrl === '').to.equal(true)"), ], }, { "name": "批量导入一条合法菜品和一条超长编号", "method": "POST", "path": "/school/foods/import", "body": { "schoolId": 1, "by": "自动化导入员", "items": [ { "code": "{{test_batch}}_I01", "name": "{{test_batch}}_红烧茄子", "canteenId": 2, "imageUrl": None, "remark": "批量导入中文备注", }, { "code": CODE_65, "name": "{{test_batch}}_NEG_CODE_65", "canteenId": 2, "remark": "该项必须失败且不得写入", }, ], }, "tests": [ ("批量导入：success=1", "pm.expect(Number(__json.data.success)).to.equal(1)"), ("批量导入：fail=1 且 total=2", "pm.expect(Number(__json.data.fail)).to.equal(1); pm.expect(Number(__json.data.total)).to.equal(2)"), ], }, { "name": "批量导入后核对合法与失败项", "method": "GET", "path": "/school/foods?schoolId=1&canteenId=2&q={{test_batch}}&status=all&page=1&pageSize=50", "tests": [ ("导入回查：合法红烧茄子存在", "pm.expect(__json.data.items.some(function(x){return x.name===pm.variables.get('test_batch') + '_红烧茄子';})).to.equal(true)"), ("导入回查：超长编号项零写入", "pm.expect(__json.data.items.some(function(x){return x.name===pm.variables.get('test_batch') + '_NEG_CODE_65';})).to.equal(false)"), ], }, { "name": "导出菜品 CSV 并核对本轮数据", "method": "GET", "path": "/school/foods/export.csv?schoolId=1&q={{test_batch}}&status=all", "tests": [ ("菜品 CSV：包含中文表头", "pm.expect(__json.data.csv).to.include('菜品编号,菜品名称,食堂')"), ("菜品 CSV：包含本轮 marker 和中文菜名", "pm.expect(__json.data.csv).to.include(pm.variables.get('test_batch')); pm.expect(__json.data.csv).to.include('鱼香肉丝'); pm.expect(__json.data.csv).to.include('红烧茄子')"), ], }, { "name": "直接新增 65 字符菜品编号应 400", "method": "POST", "path": "/school/foods", "expected_code": 400, "expected_http": 400, "body": { "schoolId": 1, "canteenId": 2, "code": CODE_65, "name": "{{test_batch}}_NEG_DIRECT_CODE_65", "enabled": True, "by": "自动化边界测试员", }, "tests": [("65 字编号：返回 64 字符上限错误", "pm.expect(String(__json.message)).to.match(/菜品编号|64/)")], }, { "name": "65 字符编号失败后确认零写入", "method": "GET", "path": "/school/foods?schoolId=1&canteenId=2&q={{test_batch}}_NEG_DIRECT_CODE_65&status=all&page=1&pageSize=20", "tests": [ ("65 字编号零写入：total=0", "pm.expect(Number(__json.data.total)).to.equal(0)"), ("65 字编号零写入：items 为空", "pm.expect(__json.data.items).to.be.an('array').that.is.empty"), ], }, { "name": "读取学校人员基线", "method": "GET", "path": "/school/personnel?schoolId=1&canteenId=2&page=1&pageSize=20", "tests": [("人员基线：items 为数组", "pm.expect(__json.data.items).to.be.an('array')")], }, { "name": "新增本轮专用留样人员", "method": "POST", "path": "/school/personnel", "body": { "schoolId": 1, "canteenId": 2, "name": "{{test_batch}}_李留样", "gender": "女", "phone": "13900001111", "jobTitle": "留样管理员", "enabled": False, }, "extract": ("personnel_id", "$.data.id"), "tests": [("人员创建：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)")], }, { "name": "回查本轮专用人员", "method": "GET", "path": "/school/personnel/detail?id={{personnel_id}}", "tests": [ ("人员回查：中文姓名正确", "pm.expect(__json.data.name).to.equal(pm.variables.get('test_batch') + '_李留样')"), ("人员回查：学校、食堂、停用状态正确", "pm.expect(Number(__json.data.schoolId || __json.data.school_id)).to.equal(1); pm.expect(Number(__json.data.canteenId || __json.data.canteen_id)).to.equal(2); pm.expect(Boolean(__json.data.enabled)).to.equal(false)"), ], }, { "name": "新增留样人员授权", "method": "POST", "path": "/school/sampling/operators", "body": { "schoolId": 1, "canteenId": 2, "personnelIds": ["{{personnel_id}}"], "password": "{{operator_password}}", "faceOssUrl": None, "faceFilePath": None, }, "tests": [("留样人员新增：count=1", "pm.expect(Number(__json.data.count)).to.equal(1)")], }, { "name": "留样人员列表回查且不得泄露密码", "method": "GET", "path": "/school/sampling/operators?schoolId=1&canteenId=2&page=1&pageSize=100", "tests": [ ("留样人员回查：包含本轮人员 ID", "const id=String(pm.variables.get('personnel_id')); pm.expect(__json.data.items.some(function(x){return String(x.id)===id;})).to.equal(true)"), ("留样人员列表：不包含 password 字段", "pm.expect(__json.data.items.every(function(x){return !Object.prototype.hasOwnProperty.call(x,'password');})).to.equal(true)"), ], }, { "name": "留样人员编辑详情可回显密码但证据脱敏", "method": "GET", "path": "/school/sampling/operators/detail?schoolId=1&id={{personnel_id}}", "tests": [ ("留样人员详情：ID 正确", "pm.expect(String(__json.data.id)).to.equal(String(pm.variables.get('personnel_id')))"), ("留样人员详情：密码专列存在", "pm.expect(String(__json.data.password)).to.equal(String(pm.variables.get('operator_password')))"), ], }, { "name": "更新留样人员中文、时间、布尔值和 NULL 图片", "method": "POST", "path": "/school/sampling/operators/upsert", "body": { "schoolId": 1, "canteenId": 2, "id": "{{personnel_id}}", "userName": "{{test_batch}}_李留样复核", "userNo": "{{test_batch}}_OP01", "jobTitle": "留样复核员", "enabled": False, "faceOssUrl": None, "faceFilePath": None, "updTime": "2026-09-26 09:10:11", "pushTime": "2026-09-26 09:10:12", }, "tests": [("留样人员更新：count=1", "pm.expect(Number(__json.data.count)).to.equal(1)")], }, { "name": "更新后回查留样人员字段", "method": "GET", "path": "/school/sampling/operators?schoolId=1&canteenId=2&page=1&pageSize=100", "tests": [ ("人员更新回查：中文姓名、时间正确", "const id=String(pm.variables.get('personnel_id')); const row=__json.data.items.find(function(x){return String(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.userName).to.equal(pm.variables.get('test_batch') + '_李留样复核'); pm.expect(row.updTime).to.equal('2026-09-26 09:10:11'); pm.expect(row.pushTime).to.equal('2026-09-26 09:10:12')"), ("人员更新回查：仍不泄露密码", "const id=String(pm.variables.get('personnel_id')); const row=__json.data.items.find(function(x){return String(x.id)===id;}); pm.expect(Object.prototype.hasOwnProperty.call(row,'password')).to.equal(false)"), ], }, { "name": "跨校菜品列表应 403", "method": "GET", "path": "/school/foods?schoolId=99999999&page=1&pageSize=20", "expected_code": 403, "expected_http": 403, "tests": [("跨校菜品：无业务数据", "pm.expect(__json.data).to.not.exist")], }, { "name": "无 Token 留样人员列表应 401", "method": "GET", "path": "/school/sampling/operators?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": "删除本轮菜品（需即时确认）", "method": "DELETE", "path": "/school/foods/{{food_id}}?schoolId=1&code={{test_batch}}_F01", "disabled": True, "tests": [("菜品删除：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "删除本轮留样人员授权（需即时确认）", "method": "POST", "path": "/school/sampling/operators/remove", "disabled": True, "body": {"schoolId": 1, "ids": ["{{personnel_id}}"]}, "tests": [("留样人员删除：count=1", "pm.expect(Number(__json.data.count)).to.equal(1)")], }, { "name": "删除本轮人员（需即时确认）", "method": "POST", "path": "/school/personnel/delete", "disabled": True, "body": {"id": "{{personnel_id}}"}, "tests": [("人员删除：ok=true", "pm.expect(__json.data.ok).to.equal(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": 90, } ) loop = copy.deepcopy(loop_template) loop["id"] = shared.new_id() loop["number"] = 1 login = shared.make_login(login_template, PREFIX) login["customHttpRequest"]["preProcessors"][0]["data"] += ( "pm.variables.set('operator_password', 'S08-' + Date.now() + '-Aa1!');\n" ) loop["children"] = [login] for number, operation in enumerate(OPERATIONS, start=2): step = shared.make_request(request_template, **operation) 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 = [request for request in requests if request.get("path") != "{{base_url}}/auth/login"] 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") 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-S08 双库兼容一键串行入口，当前共 74 个唯一接口。" "MySQL 仅允许 foodsafe_test，达梦仅允许 WWWWEB。写接口必须接口回查并与数据库原表证据合并；" "S08 删除步骤、设备回调和 SSE 默认关闭，解除门禁后独立运行。" ) 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("S08 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()