#!/usr/bin/env python3 """Create/update the S09 inventory and supply-chain Apifox scenario.""" from __future__ import annotations import copy import importlib.util from pathlib import Path BASE_SCRIPT = Path( "/Users/guodongwei/Documents/zhct/zhctprompt/" "work_food_safety/2026-09-24-apifox-full-api-automation/" "apifox/scripts/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 = "S09-库存与供应链（双库真实写入）" PREFIX = "APIFOX_S09" DESCRIPTION = ( "学校端 S09 库存与供应链双库真实写入场景：商品、供应商、入库单更新及附件、" "FIFO 出库、库存批次、盘点和 CSV 导出。模拟资料仅保存 example.invalid URL 文本，" "不发起文件、OSS、设备或第三方请求。测试记录保留用于 MySQL foodsafe_test 与达梦 WWWWEB 原表取证。" ) INBOUND_BODY = { "schoolId": 1, "canteenId": 2, "supplierId": "{{supplier_id}}", "date": "2026-09-26", "operator": "库管员李师傅", "items": [ { "productId": "{{product_id}}", "qty": 5.125, "unitPrice": 12.34, "prodDate": "2026-09-25", "shelfLifeDays": 30, } ], "tickets": [ {"type": "ticket_quarantine", "imageUrl": "https://example.invalid/apifox-s09-quarantine.jpg"}, {"type": "ticket_invoice", "imageUrl": "https://example.invalid/apifox-s09-invoice.jpg"}, ], "images": ["https://example.invalid/apifox-s09-inbound.jpg"], } UPDATED_INBOUND_BODY = copy.deepcopy(INBOUND_BODY) UPDATED_INBOUND_BODY["items"][0]["qty"] = 6.125 UPDATED_INBOUND_BODY["items"][0]["unitPrice"] = 12.35 OPERATIONS = [ { "name": "读取库存商品基线", "method": "GET", "path": "/school/inventory/products?schoolId=1", "tests": [("商品基线：返回数组", "pm.expect(__json.data).to.be.an('array')")], }, { "name": "新增鲜香菇商品", "method": "POST", "path": "/school/inventory/products", "body": { "schoolId": 1, "name": "鲜香菇一级品（批次{{test_batch}}）", "unit": "kg", "category": "蔬菜", "spec": "散装一级品", "lastPrice": 12.34, }, "extract": ("product_id", "$.data.id"), "tests": [("商品新增：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)")], }, { "name": "更新鲜香菇名称规格与价格", "method": "PATCH", "path": "/school/inventory/products?id={{product_id}}", "body": { "schoolId": 1, "name": "鲜香菇一级品（批次{{test_batch}}）", "spec": None, "lastPrice": 12.36, }, "tests": [("商品更新：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "回查更新后商品", "method": "GET", "path": "/school/inventory/products?schoolId=1", "tests": [ ("商品回查：包含创建 ID", "const id=Number(pm.variables.get('product_id')); pm.expect(__json.data.some(function(x){return Number(x.id)===id;})).to.equal(true)"), ("商品回查：中文名与两位价格正确", "const id=Number(pm.variables.get('product_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row.name).to.equal('鲜香菇一级品（批次' + pm.variables.get('test_batch') + '）'); pm.expect(Number(row.lastPrice || row.last_price)).to.equal(12.36)"), ], }, { "name": "模板导入低筋面粉商品", "method": "POST", "path": "/school/inventory/products/import/template", "body": { "schoolId": 1, "items": [ { "name": "低筋面粉5kg装（批次{{test_batch}}）", "unit": "袋", "category": "粮油", "spec": "5kg/袋", "lastPrice": 23.45, } ], }, "tests": [("模板导入：count=1", "pm.expect(Number(__json.data.count)).to.equal(1)")], }, { "name": "回查模板导入商品", "method": "GET", "path": "/school/inventory/products?schoolId=1", "tests": [("导入回查：低筋面粉存在且价格 23.45", "const expected='低筋面粉5kg装（批次' + pm.variables.get('test_batch') + '）'; const row=__json.data.find(function(x){return x.name===expected;}); pm.expect(row).to.exist; pm.expect(Number(row.lastPrice || row.last_price)).to.equal(23.45)")], }, { "name": "新增惠民农副产品配送中心", "method": "POST", "path": "/school/inventory/suppliers", "body": {"schoolId": 1, "name": "惠民农副产品配送中心（档案{{test_batch}}）", "contact": "王师傅"}, "extract": ("supplier_id", "$.data.id"), "tests": [("供应商新增：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)")], }, { "name": "创建入库单并保存三个附件引用", "method": "POST", "path": "/school/inventory/inbound/doc", "body": INBOUND_BODY, "extract": ("inbound_doc_no", "$.data.docNo"), "tests": [("入库创建：返回 IN- 单号", "pm.expect(String(__json.data.docNo)).to.match(/^IN-/)")], }, { "name": "更新入库数量和单价", "method": "PATCH", "path": "/school/inventory/inbound/doc?docNo={{inbound_doc_no}}", "body": UPDATED_INBOUND_BODY, "tests": [("入库更新：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "入库单详情回查", "method": "GET", "path": "/school/inventory/inbound/doc?docNo={{inbound_doc_no}}&schoolId=1", "tests": [ ("入库详情：唯一明细数量 6.125 价格 12.35", "pm.expect(__json.data.items).to.have.lengthOf(1); pm.expect(Number(__json.data.items[0].qty)).to.equal(6.125); pm.expect(Number(__json.data.items[0].unitPrice || __json.data.items[0].unit_price)).to.equal(12.35)"), ("入库详情：附件合计 3 条", "pm.expect(__json.data.attachments).to.have.lengthOf(3)"), ], }, { "name": "入库单列表回查", "method": "GET", "path": "/school/inventory/inbound/docs?schoolId=1", "tests": [("入库列表：包含本轮单号", "const no=String(pm.variables.get('inbound_doc_no')); pm.expect(__json.data.some(function(x){return String(x.docNo || x.doc_no)===no;})).to.equal(true)")], }, { "name": "导出入库 CSV", "method": "GET", "path": "/school/inventory/inbound/export.csv?schoolId=1", "tests": [("入库 CSV：包含本轮单号和操作人", "pm.expect(__json.data.csv).to.include(String(pm.variables.get('inbound_doc_no'))); pm.expect(__json.data.csv).to.include('库管员李师傅')")], }, { "name": "查询 FIFO 可出库批次", "method": "GET", "path": "/school/inventory/outbound/batches?schoolId=1&productId={{product_id}}&canteenId=2", "tests": [("FIFO 批次：包含本轮入库明细并提取批次 ID", "const no=String(pm.variables.get('inbound_doc_no')); const row=__json.data.find(function(x){return String(x.docNo || x.doc_no)===no;}); pm.expect(row).to.exist; pm.expect(Number(row.inboundId || row.inbound_id)).to.be.greaterThan(0); pm.variables.set('inbound_id', String(row.inboundId || row.inbound_id))")], }, { "name": "创建午餐备餐出库", "method": "POST", "path": "/school/inventory/outbound", "body": {"schoolId": 1, "canteenId": 2, "productId": "{{product_id}}", "qty": 1.125, "purpose": "午餐备餐领用", "by": "出库员周师傅", "receiver": "领用员赵师傅", "date": "2026-09-26"}, "extract": ("outbound_id", "$.data.id"), "tests": [("出库创建：返回正整数 ID", "pm.expect(Number(__json.data.id)).to.be.greaterThan(0)")], }, { "name": "出库列表回查", "method": "GET", "path": "/school/inventory/outbound?schoolId=1", "tests": [("出库回查：包含创建 ID 与中文用途", "const id=Number(pm.variables.get('outbound_id')); const row=__json.data.find(function(x){return Number(x.id)===id;}); pm.expect(row).to.exist; pm.expect(row.purpose).to.equal('午餐备餐领用'); pm.expect(Number(row.qty)).to.equal(1.125)")], }, { "name": "导出出库 CSV", "method": "GET", "path": "/school/inventory/outbound/export.csv?schoolId=1", "tests": [("出库 CSV：包含中文用途", "pm.expect(__json.data.csv).to.include('午餐备餐领用')")], }, { "name": "出库后库存回查", "method": "GET", "path": "/school/inventory/stock?schoolId=1", "tests": [("库存回查：本轮商品数量 5.000", "const id=Number(pm.variables.get('product_id')); const row=__json.data.find(function(x){return Number(x.productId || x.product_id)===id;}); pm.expect(row).to.exist; pm.expect(Number(row.qty)).to.equal(5)")], }, { "name": "库存批次回查", "method": "GET", "path": "/school/inventory/stock/batches?schoolId=1", "tests": [("库存批次：本轮入库批次数量 5.000", "const id=Number(pm.variables.get('inbound_id')); const row=__json.data.find(function(x){return Number(x.inboundId || x.inbound_id)===id;}); pm.expect(row).to.exist; pm.expect(Number(row.qty)).to.equal(5)")], }, { "name": "导出库存 CSV", "method": "GET", "path": "/school/inventory/stock/export.csv?schoolId=1", "tests": [("库存 CSV：包含鲜香菇一级品", "pm.expect(__json.data.csv).to.include('鲜香菇一级品（批次' + pm.variables.get('test_batch') + '）')")], }, { "name": "执行批次盘点", "method": "POST", "path": "/school/inventory/stock/stocktake-batch", "body": {"schoolId": 1, "inboundId": "{{inbound_id}}", "actualQty": 4.125, "operator": "盘点员孙师傅"}, "tests": [("盘点写入：ok=true", "pm.expect(__json.data.ok).to.equal(true)")], }, { "name": "盘点后库存回查", "method": "GET", "path": "/school/inventory/stock?schoolId=1", "tests": [("盘点回查：本轮商品数量 4.125", "const id=Number(pm.variables.get('product_id')); const row=__json.data.find(function(x){return Number(x.productId || x.product_id)===id;}); pm.expect(row).to.exist; pm.expect(Number(row.qty)).to.equal(4.125)")], }, ] 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": 100}) 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 + "pm.variables.set('test_batch', '20260926' + String(Date.now()).slice(-6));\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-S09 双库兼容一键串行入口，当前共 92 个唯一接口。" "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("S09 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()