# ============================================================ # RULE 引擎黄金测试（moduleId: golden-rule-engine, 发版门禁 §14.3） # 断言"结构不变式"而非绝对日期（种子日期相对今天，任意运行日稳定）： # G1 版本/PO/WO 计数关系 G2 工位无双占（硬约束 §3.5） # G3 工艺先后序不颠倒 G4 工单落在班次日历内（或兜底可解释） # G5 策略排序生效 G6 冲突结构合法 G7 KPI 值域合法 # ============================================================ from __future__ import annotations # 前向类型引用 import pytest # 测试框架 from server.engines import get_engine # 引擎工厂 from server.engines.base import EngineParams # 引擎入参 from server.engines.queries import get_line_shifts # 班次查询（G4 用） from server.state.seed import seed_world # 种子数据（每用例全新世界） from server.timeutil import add_minutes, fmt_date, parse_dt, today0 # 日期工具 def _fresh_world(): """构造一个全新种子世界（用例间隔离，确定性）。""" return seed_world() # 直接返回种子 def _next_id_factory(): """独立发号器（不依赖 WorldStore，保证用例纯内存）。""" counters: dict[str, int] = {} # 类型 → 当前号 def next_id(kind: str) -> int: # 闭包发号函数 counters[kind] = counters.get(kind, 0) + 1 # 自增 return counters[kind] # 返回新号 return next_id # 返回闭包 def _run(world, strategy="COMPREHENSIVE", engine="RULE"): """执行一次排产（默认从明天开始、展望 14 天）。""" start = fmt_date(add_minutes(today0(), 24 * 60)) # 明天 params = EngineParams(orderIds=[], engineType=engine, strategyTemplate=strategy, planningHorizonDays=14, startDate=start) # 组装参数 return get_engine(engine).solve(world, params, _next_id_factory()) # 求解 # ---------------- G1：计数关系 ---------------- def test_counts_match_seed_structure(): """版本计数必须与种子结构一致：7 单皆可排 → 7 个 PO；工单数 = 各产品工艺步骤数之和。""" world = _fresh_world() # 全新世界 result = _run(world) # 执行排产 assert result.orderCount == 7 # 7 张订单全部参与（无 CANCELLED/COMPLETED） assert result.poCount == 7 # 每单一项 → 7 个生产订单 # 工单数 = Σ 该订单产品的工艺步骤数（A=5 步，B=4 步，C=4 步；种子订单产品为 1,2,1,3,2,3,2） expected_wo = 5 + 4 + 5 + 4 + 4 + 4 + 4 # 逐单累加 = 30 assert result.woCount == expected_wo # 工单总数固定 assert len([v for v in world["scheduleVersions"]]) == 1 # 恰生成一个版本 # ---------------- G2：工位无双占（硬约束） ---------------- def test_no_workstation_overlap(): """同一工位的任意两工单时间区间不得重叠（§3.5 资源硬约束）。""" world = _fresh_world() # 全新世界 _run(world) # 执行排产 by_ws: dict[int, list] = {} # 工位 → [(start,end)] for wo in world["workOrders"]: # 收集全部工单区间 by_ws.setdefault(wo["workstationId"], []).append( (parse_dt(wo["plannedStartTime"]), parse_dt(wo["plannedEndTime"]))) for ws_id, ivs in by_ws.items(): # 逐工位校验 ivs.sort() # 按开始时间排序 for (s1, e1), (s2, e2) in zip(ivs, ivs[1:]): # 相邻两两比较 assert e1 <= s2, f"工位 {ws_id} 存在双占：{e1} > {s2}" # 前段结束不得晚于后段开始 # ---------------- G3：工艺先后序 ---------------- def test_operation_sequence_preserved(): """同一生产订单内，工序顺序号大的工单开始时间不得早于顺序号小的结束时间。""" world = _fresh_world() # 全新世界 _run(world) # 执行排产 by_po: dict[int, list] = {} # PO → 工单列表 for wo in world["workOrders"]: # 归组 by_po.setdefault(wo["productionOrderId"], []).append(wo) for po_id, wos in by_po.items(): # 逐 PO 校验 wos.sort(key=lambda w: w["sequenceNo"]) # 按工艺顺序排列 for prev, nxt in zip(wos, wos[1:]): # 相邻工序比较 assert parse_dt(nxt["plannedStartTime"]) >= parse_dt(prev["plannedEndTime"]), \ f"PO {po_id} 工序颠倒：{nxt['orderNo']} 早于 {prev['orderNo']} 结束" # 后序不得早于前序结束 # ---------------- G4：班次日历约束 ---------------- def test_work_orders_within_shift_calendar(): """工单开始日应为该产线的工作日（兜底顺延的极端情形除外——种子规模下不应触发）。""" world = _fresh_world() # 全新世界 _run(world) # 执行排产 for wo in world["workOrders"]: # 逐工单校验 date_str = wo["plannedStartTime"][:10] # 开始日期 shifts = get_line_shifts(world, wo["lineId"], date_str) # 当日班次 # 种子数据 14 天展望内产能充足，不应触发"30 天无槽"兜底 → 必须落在工作日 assert shifts, f"工单 {wo['orderNo']} 落在非工作日 {date_str}" # ---------------- G5：策略排序生效 ---------------- def test_delivery_first_orders_by_due_date(): """交期优先策略下，最早交期订单（比亚迪 +4 天）的 PO 应最先开始。""" world = _fresh_world() # 全新世界 _run(world, strategy="DELIVERY_FIRST") # 交期优先排产 pos = world["productionOrders"] # 全部 PO earliest = min(pos, key=lambda p: parse_dt(p["plannedStartDate"])) # 最早开始的 PO so = next(s for s in world["salesOrders"] if s["id"] == earliest["salesOrderId"]) # 对应订单 min_delivery = min(s["deliveryDate"] for s in world["salesOrders"]) # 全局最早交期 assert so["deliveryDate"] == min_delivery # 最早开始的必须是最早交期的订单（EDD） # ---------------- G6：冲突结构合法 ---------------- def test_conflicts_are_wellformed(): """所有冲突必须归属本版本、未解决、类型/级别在枚举内（§3.4 证据可溯的前提）。""" world = _fresh_world() # 全新世界 result = _run(world) # 执行排产 valid_types = {"NO_LINE", "NO_WORKSTATION", "MATERIAL_SHORTAGE", "DELAY", "CAPACITY", "EQUIPMENT"} # 类型枚举 valid_sev = {"CRITICAL", "MAJOR", "MINOR"} # 级别枚举 assert len(world["conflicts"]) == result.conflictCount # 计数一致 for c in world["conflicts"]: # 逐条校验 assert c["versionId"] == result.versionId # 归属本版本 assert c["isResolved"] is False # 初始未解决 assert c["conflictType"] in valid_types # 类型合法 assert c["severity"] in valid_sev # 级别合法 # ---------------- G7：KPI 值域 ---------------- def test_kpi_ranges(): """KPI 必须在合法值域：利用率 0~1.5（容差），延迟/成本非负。""" world = _fresh_world() # 全新世界 result = _run(world) # 执行排产 assert 0 <= result.avgUtilization <= 1.5 # 利用率合理（>1 表示超载但不应爆炸） assert result.totalTardiness >= 0 # 延迟非负 assert result.totalCost >= 0 # 成本非负 # ---------------- G8：引擎类型标注诚实 ---------------- @pytest.mark.parametrize("requested", ["RULE", "GA"]) def test_engine_type_recorded_for_rule_proxy(requested): """GA 仍由 RULE 代跑，但结果必须如实记录请求类型。""" world = _fresh_world() result = _run(world, engine=requested) assert result.engineType == requested assert world["scheduleVersions"][0]["engineType"] == requested def test_cp_engine_is_real_not_silent_rule(): """SC-03：请求 CP 必须走 CpSatEngine，不得静默 RULE 代跑却不留 solverMeta。""" from server.engines.cp_engine import CpSatEngine assert isinstance(get_engine("CP"), CpSatEngine) world = _fresh_world() result = _run(world, engine="CP") assert result.engineType == "CP" assert world["scheduleVersions"][0].get("solverMeta", {}).get("backend") == "OR-Tools CP-SAT" def test_hybrid_engine_is_real_not_silent_rule(): """SC-03：HYBRID 必须走 HybridEngine（RULE 热启动 + CP）。""" from server.engines.cp_engine import HybridEngine assert isinstance(get_engine("HYBRID"), HybridEngine) world = _fresh_world() result = _run(world, engine="HYBRID") assert result.engineType == "HYBRID" meta = world["scheduleVersions"][0].get("solverMeta") or {} assert meta.get("warmStart") == "RULE"