# ============================================================ # OR-04 紧急插单专用流黄金测试 # 固化：快评不碰主干；采用后生成 DRAFT 版本且可回滚至采用前 # ============================================================ from __future__ import annotations import copy from server.aps_domain.rush import apply_rush, evaluate_rush from server.state.seed import seed_world from server.timeutil import add_minutes, fmt_date, today0 def _next_id_factory(world): tables = { "audit": "auditEvents", "conflict": "conflicts", "log": "logs", "productionOrder": "productionOrders", "salesOrder": "salesOrders", "scheduleVersion": "scheduleVersions", "workOrder": "workOrders", } counters: dict[str, int] = {} def next_id(kind: str) -> int: if kind not in counters: rows = world.get(tables.get(kind, kind + "s"), []) counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0) counters[kind] += 1 return counters[kind] return next_id def test_rush_evaluate_does_not_mutate_trunk(): """插单快评全程沙盒：主干订单数/急单标记不变。""" world = seed_world() before = copy.deepcopy(world) n_orders = len(world["salesOrders"]) impact = evaluate_rush(world, { "customerName": "快评客户", "customerLevel": "VIP", "productId": 1, "quantity": 200, "deliveryDate": fmt_date(add_minutes(today0(), 4 * 24 * 60)), "priority": 1, "strategy": "DELIVERY_FIRST", }) assert impact["affectedOrderCount"] >= 0 assert "delayDelta" in impact and "conflictDelta" in impact assert impact["after"]["orderCount"] == impact["baseline"]["orderCount"] + 1 assert len(world["salesOrders"]) == n_orders assert world["salesOrders"] == before["salesOrders"] assert world["productionOrders"] == before["productionOrders"] assert world["scheduleVersions"] == before["scheduleVersions"] def test_rush_apply_creates_draft_and_keeps_baseline_rollback_point(): """采用插单：主干新增急单 + DRAFT 版本；采用前快照可还原。""" world = seed_world() baseline_versions = len(world["scheduleVersions"]) snapshot = copy.deepcopy(world) impact = evaluate_rush(world, { "customerName": "采用客户", "customerLevel": "VIP", "productCode": "CTRL-A", "quantity": 150, "deliveryDate": fmt_date(add_minutes(today0(), 5 * 24 * 60)), "strategy": "DELIVERY_FIRST", }) # 快评后主干仍干净 assert len(world["salesOrders"]) == len(snapshot["salesOrders"]) applied = apply_rush(world, _next_id_factory(world), { "payload": impact["payload"], "strategy": impact["strategy"], "evalId": impact["evalId"], }) order = applied["order"] result = applied["result"] assert order["isRush"] is True assert order["status"] == "APPROVED" assert any(so["id"] == order["id"] for so in world["salesOrders"]) assert len(world["scheduleVersions"]) == baseline_versions + 1 assert world["scheduleVersions"][-1]["status"] == "DRAFT" assert result.versionNo == world["scheduleVersions"][-1]["versionNo"] assert result.orderCount == 8 # 种子 7 + 急单 1 # 回滚到采用前快照（模拟检查点恢复） world.clear() world.update(copy.deepcopy(snapshot)) assert len(world["salesOrders"]) == 7 assert len(world["scheduleVersions"]) == baseline_versions assert all(not so.get("orderNo", "").endswith(str(order["id"])) or so["id"] != order["id"] for so in world["salesOrders"]) def test_rush_evaluate_existing_order(): """对已有订单加急：受影响指标来自沙盒差分。""" world = seed_world() target = world["salesOrders"][3] # 非急单 assert not target.get("isRush") impact = evaluate_rush(world, {"orderNo": target["orderNo"], "strategy": "COMPREHENSIVE"}) assert impact["mode"] == "existing" assert impact["rushOrder"]["orderNo"] == target["orderNo"] assert not target.get("isRush") # 主干未改 assert impact["baseline"]["orderCount"] == 7 assert impact["after"]["orderCount"] == 7 def _trial_flex_world() -> dict: """试排世界夹具：主干种子数据 + planningContext.trialOnly + 最小柔性主数据。 模拟「文件夹快试导入」后的真实形态：排产只认 flex* 主数据， 主干 RULE 引擎在此类世界上只能产出 0 工单的空壳版本。 """ world = seed_world() world["planningContext"] = {"trialOnly": True, "dataDate": "2026-08-04"} world["flexOperations"] = [{"id": 1, "code": "CUT", "name": "下料", "changeoverMin": 10}] world["flexTeams"] = [{ "id": 1, "code": "TEAM-CUT", "name": "下料班", "memberCount": 4, "supportOps": ["CUT"], }] world["flexRoutings"] = [{ "id": 1, "productCode": "CTRL-A", "productName": "控制柜 A", "seq": 1, "operationCode": "CUT", "operationName": "下料", "stdTimePerUnit": 5.0, }] world["flexMaterials"] = [{ "id": 1, "code": "CTRL-A", "name": "控制柜 A", "type": "FINISHED_PRODUCT", "unit": "件", "stock": 0, "inTransit": 0, "safetyStock": 0, "procurementLeadTime": 0, "sourcingType": "MAKE", }] world["flexEquipment"] = [{ "id": 1, "code": "EQ-1", "name": "切割机", "status": "RUNNING", "capabilities": ["CUT"], "availabilityRate": 1.0, }] return world def test_rush_apply_in_trial_world_routes_to_flex_engine(): """试排世界采用急单必须走柔性引擎（回归：急单入池却排不进/看不见）。 锁定三点：急单桥接进 flexOrders（source=SALES/RELEASED）；生成 FV 柔性 草稿版本且急单真实排出虚拟产线；主轨 scheduleVersions 不再新增空壳版本。 """ world = _trial_flex_world() n_trunk_versions = len(world["scheduleVersions"]) n_flex_versions = len(world.get("flexScheduleVersions") or []) applied = apply_rush(world, _next_id_factory(world), { "payload": { "customerName": "试排急单客户", "customerLevel": "VIP", "productCode": "CTRL-A", "quantity": 3, "deliveryDate": fmt_date(add_minutes(today0(), 5 * 24 * 60)), "priority": 1, }, "strategy": "DELIVERY_FIRST", }) order = applied["order"] result = applied["result"] assert order["isRush"] is True assert order["status"] == "APPROVED" # 急单已桥接进柔性订单池 flex_row = next( (o for o in world["flexOrders"] if o.get("orderNo") == order["orderNo"]), None, ) assert flex_row is not None, "急单未桥接进 flexOrders，柔性排产必然看不到它" assert flex_row["source"] == "SALES" assert flex_row["status"] == "RELEASED" assert flex_row["isRush"] is True # 生成的是柔性 FV 草稿版本，而不是主轨 RULE 空壳版本 assert str(result.versionNo).startswith("FV") assert len(world["scheduleVersions"]) == n_trunk_versions, "主轨不应新增空壳版本" assert len(world["flexScheduleVersions"]) == n_flex_versions + 1 assert world["flexScheduleVersions"][-1]["versionNo"] == result.versionNo assert result.orderCount == len(world["flexOrders"]) # 急单真实排出虚拟产线（修复前在此类世界为 0 工单） rush_vls = [ v for v in world["flexVirtualLines"] if v.get("orderNo") == order["orderNo"] and v.get("versionId") == world["flexScheduleVersions"][-1]["id"] ] assert rush_vls, "急单在柔性版本中应有至少一条虚拟产线"