#!/usr/bin/env python3 """Create or update the auditable S07 school sampling 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 = "S07-留样管理（双库真实写入）" PREFIX = "APIFOX_S07" DESCRIPTION = ( "学校端 S07 留样管理双库可审计场景：真实新增、列表回查、中文处置、取样状态与时间、CSV、" "非法重量与超长烹饪时间零写入、无 Token 和跨校拒绝。设备回调需可验证绑定，SSE 是长连接，两者默认关闭并独立执行。" ) OPERATIONS = [ { "name": "新增中文留样记录并返回 ID", "method": "POST", "path": "/school/sampling/records", "body": { "schoolId": 1, "canteenId": 1, "sample": "{{test_batch}}_香菇青菜炒肉", "weight": 125.50, "imageUrl": None, "duration": 48, "by": "张老师", "cabinet": "A-01留样柜", "cookTime": "11:30:00", "at": "2026-09-26 12:00:00", "endTime": "2026-09-28 12:00:00", "status": "ACTIVE", }, "extract": ("sampling_id", "$.data.id"), "tests": [ ("创建响应：样品名包含本轮批次", "pm.expect(__json.data.sample).to.equal(pm.variables.get('test_batch') + '_香菇青菜炒肉')"), ("创建响应：重量 125.5 未截断", "pm.expect(Number(__json.data.weight)).to.equal(125.5)"), ("创建响应：留样人和柜位中文正常", "pm.expect(__json.data.by).to.equal('张老师'); pm.expect(__json.data.cabinet).to.equal('A-01留样柜')"), ("创建响应：NULL 图片未被伪造", "pm.expect(__json.data.imageUrl === null || __json.data.image_url === null).to.equal(true)"), ], }, { "name": "按批次回查刚创建的留样记录", "method": "GET", "path": "/school/sampling/records?schoolId=1&canteenId=1&sample={{test_batch}}&page=1&pageSize=20", "tests": [ ("列表回查：包含刚创建的 ID", "const id=Number(pm.variables.get('sampling_id')); pm.expect(__json.data.items.some(function(x){return Number(x.id)===id;})).to.equal(true)"), ("列表回查：样品、重量、留样人一致", "const id=Number(pm.variables.get('sampling_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(row.sample).to.include(pm.variables.get('test_batch')); pm.expect(Number(row.weight)).to.equal(125.5); pm.expect(row.by).to.equal('张老师')"), ], }, { "name": "更新留样处置措施", "method": "PATCH", "path": "/school/sampling/records/{{sampling_id}}/measure", "body": {"measure": "{{test_batch}} 中文处置：复核留样柜温度并完成登记"}, "tests": [("处置更新：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "处置后回查措施已持久化", "method": "GET", "path": "/school/sampling/records?schoolId=1&sample={{test_batch}}&page=1&pageSize=20", "tests": [ ("处置回查：中文措施与本轮批次一致", "const id=Number(pm.variables.get('sampling_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(row.measure).to.equal(pm.variables.get('test_batch') + ' 中文处置：复核留样柜温度并完成登记')"), ], }, { "name": "更新取样状态、取样人和北京时间", "method": "PATCH", "path": "/school/sampling/records/{{sampling_id}}/status", "body": { "status": "CLEARED", "quyangUserName": "王取样员", "takeTime": "2026-09-28 12:00:01", }, "tests": [ ("状态更新：ok=true", "pm.expect(__json.data.ok).to.equal(true)"), ("状态更新：业务响应不伪造完整记录", "pm.expect(__json.data.status).to.not.exist; pm.expect(__json.data.quyangUserName).to.not.exist"), ], }, { "name": "状态更新后再次回查原记录", "method": "GET", "path": "/school/sampling/records?schoolId=1&sample={{test_batch}}&status=CLEARED&page=1&pageSize=20", "tests": [ ("状态回查：原 ID 为 CLEARED", "const id=Number(pm.variables.get('sampling_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.status).to.equal('CLEARED')"), ("状态回查：取样人和时间已保存", "const id=Number(pm.variables.get('sampling_id')); const row=__json.data.items.find(function(x){return Number(x.id)===id;}); pm.expect(row.quyangUserName || row.quyang_user_name).to.equal('王取样员'); pm.expect(String(row.takeTime || row.take_time)).to.include('2026-09-28 12:00:01')"), ], }, { "name": "导出留样 CSV 并核对本轮数据", "method": "GET", "path": "/school/sampling/records/export.csv?schoolId=1&sample={{test_batch}}&status=CLEARED", "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('125.5')"), ("CSV 响应：包含中文处置措施", "pm.expect(__json.data.csv).to.include('复核留样柜温度')"), ], }, { "name": "非法重量 0 应 400 且不得写入", "method": "POST", "path": "/school/sampling/records", "expected_code": 400, "expected_http": 400, "body": { "schoolId": 1, "canteenId": 1, "sample": "{{test_batch}}_NEG_ZERO_WEIGHT", "weight": 0, "duration": 48, "by": "非法重量测试员", "at": "2026-09-26 12:10:00", }, "tests": [("非法重量：返回明确错误", "pm.expect(String(__json.message)).to.match(/重量|weight|positive/i)")], }, { "name": "非法重量失败后按批次确认零写入", "method": "GET", "path": "/school/sampling/records?schoolId=1&sample={{test_batch}}_NEG_ZERO_WEIGHT&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": "19 字完整日期时间不得写入 cookTime", "method": "POST", "path": "/school/sampling/records", "expected_code": 400, "expected_http": 400, "body": { "schoolId": 1, "canteenId": 1, "sample": "{{test_batch}}_NEG_LONG_COOK_TIME", "weight": 125.50, "duration": 48, "by": "超长烹饪时间测试员", "cookTime": "2026-09-26 11:30:00", "at": "2026-09-26 12:10:00", }, "tests": [ ("超长 cookTime：两库统一返回字段长度错误", "pm.expect(String(__json.message)).to.match(/烹饪时间段|cook.?time|16/i)"), ], }, { "name": "超长 cookTime 失败后确认零写入", "method": "GET", "path": "/school/sampling/records?schoolId=1&sample={{test_batch}}_NEG_LONG_COOK_TIME&page=1&pageSize=20", "tests": [ ("超长 cookTime 零写入：total=0", "pm.expect(Number(__json.data.total)).to.equal(0)"), ("超长 cookTime 零写入：items 为空", "pm.expect(__json.data.items).to.be.an('array').that.is.empty"), ], }, { "name": "跨校留样列表应 403", "method": "GET", "path": "/school/sampling/records?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/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": "设备回调（需可验证本地绑定）", "method": "POST", "path": "/school/sampling/device/callback", "no_auth": True, "disabled": True, "body": { "instrument": "{{sampling_instrument}}", "schoolId": 99999999, "sample": "{{test_batch}}_设备留样", "weight": 126.75, "duration": 48, "by": "本地模拟设备", "cabinet": "MOCK-01", }, }, { "name": "留样 SSE（长连接独立测试）", "method": "GET", "path": "/school/sampling/stream?schoolId=1", "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": 80, } ) loop = copy.deepcopy(loop_template) loop["id"] = shared.new_id() loop["number"] = 1 loop["children"] = [shared.make_login(login_template, PREFIX)] 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-S07 双库兼容一键串行入口，当前共 59 个唯一接口。" "MySQL 仅允许 foodsafe_test，达梦仅允许 WWWWEB。写接口必须接口回查并与数据库原表证据合并；" "设备回调和 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("S07 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()