190 lines
7.9 KiB
Python
190 lines
7.9 KiB
Python
# ============================================================
|
||
# matrix-113（Q 方向）：RULE/CP/HYBRID/GA 四引擎硬约束语义一致
|
||
# 黄金测试（可重生 ✅）
|
||
# 同一世界 + 同一 engine flags：硬约束开启 → 对应冲突出现且发布被拦；
|
||
# 关闭 → 冲突消失；软权重无法关闭硬约束（profile 门禁拒绝）。
|
||
# 审计结论（2026-08-02）：四引擎均经 RuleEngine.materialize_schedule 统一物化，
|
||
# equipment(C4)/capacity(C7)/materialKit(C6) 开关行为一致，无需引擎修复；
|
||
# personnel/tooling/freeze 开关只作用于 FLEX 引擎（柔性世界 flex*），
|
||
# 不在本 RULE/CP/HYBRID/GA 四引擎集合的可观察面内（差异见模块注释）。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
|
||
import pytest
|
||
|
||
from server.aps_domain.constraints import (
|
||
ConstraintProfileDenied,
|
||
apply_profile_save,
|
||
engine_constraint_flags,
|
||
hard_blocking_conflicts,
|
||
)
|
||
from server.engines import get_engine
|
||
from server.engines.base import EngineParams
|
||
from server.state.seed import seed_world
|
||
from server.timeutil import add_minutes, fmt_date, fmt_dt, parse_dt, today0
|
||
|
||
ENGINES = ["RULE", "CP", "HYBRID", "GA"]
|
||
|
||
|
||
def _next_id_factory():
|
||
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, engine, flags):
|
||
"""在传入世界跑一次指定引擎（solve 会就地写版本/冲突）。"""
|
||
start = fmt_date(add_minutes(today0(), 24 * 60))
|
||
params = EngineParams(
|
||
orderIds=[world["salesOrders"][0]["id"]], engineType=engine,
|
||
strategyTemplate="COMPREHENSIVE", planningHorizonDays=14,
|
||
startDate=start, constraints=flags, timeLimitSeconds=2.0,
|
||
)
|
||
return get_engine(engine).solve(world, params, _next_id_factory())
|
||
|
||
|
||
def _single_order_world():
|
||
"""只保留一个订单，放大数量并压缩班次 → 容量冲突对所有引擎都必然出现。"""
|
||
world = seed_world()
|
||
so = world["salesOrders"][0]
|
||
so["items"] = [so["items"][0]]
|
||
so["items"][0]["quantity"] = 3000
|
||
keep_pid = so["items"][0]["productId"]
|
||
world["lineProducts"] = [lp for lp in world["lineProducts"] if lp["productId"] == keep_pid]
|
||
for s in world["shifts"]:
|
||
start_hour = int(str(s["startTime"]).split(":", 1)[0])
|
||
s["endTime"] = f"{start_hour + 2:02d}:00"
|
||
s["breakPeriods"] = []
|
||
return world
|
||
|
||
|
||
def _full_horizon_maintenance_world():
|
||
"""全时域维保覆盖全部设备 → 任何工单必然与维保重叠（EQUIPMENT 必现）。"""
|
||
world = _single_order_world()
|
||
world["maintenance"] = []
|
||
start = parse_dt(fmt_date(add_minutes(today0(), 24 * 60)) + " 08:00")
|
||
end = add_minutes(start, 15 * 24 * 60)
|
||
for i, eq in enumerate(world["equipment"]):
|
||
world["maintenance"].append({
|
||
"id": 100 + i, "equipmentId": eq["id"], "status": "PLANNED",
|
||
"plannedStart": fmt_dt(start), "plannedEnd": fmt_dt(end),
|
||
})
|
||
return world
|
||
|
||
|
||
def _zero_stock_world():
|
||
"""清空全部非成品库存 → 物料齐套冲突（C6）必现。"""
|
||
world = seed_world()
|
||
so = world["salesOrders"][0]
|
||
so["items"] = [so["items"][0]]
|
||
for m in world["materials"]:
|
||
if m.get("type") != "FINISHED_PRODUCT":
|
||
m["stock"] = 0
|
||
m["inTransit"] = 0
|
||
return world
|
||
|
||
|
||
def _conflicts_of(world, version_id, ctype):
|
||
return [c for c in world["conflicts"]
|
||
if c.get("conflictType") == ctype and c.get("versionId") == version_id]
|
||
|
||
|
||
def _blockers_of(world, version_id, ctype):
|
||
return [c for c in hard_blocking_conflicts(world, version_id=version_id)
|
||
if c["conflictType"] == ctype]
|
||
|
||
|
||
def _semantics(engine, world_factory, ctype, flags_off_key, patch=None):
|
||
"""返回 (on_conflict, on_blocker, off_conflict, off_blocker) 四元组。"""
|
||
base = world_factory()
|
||
flags = engine_constraint_flags(base)
|
||
if patch:
|
||
apply_profile_save(base, {"constraints": patch})
|
||
flags = engine_constraint_flags(base)
|
||
w_on = copy.deepcopy(base)
|
||
res = _run(w_on, engine, flags)
|
||
on = (bool(_conflicts_of(w_on, res.versionId, ctype)),
|
||
bool(_blockers_of(w_on, res.versionId, ctype)))
|
||
w_off = copy.deepcopy(base)
|
||
res = _run(w_off, engine, {**flags, flags_off_key: False})
|
||
off = (bool(_conflicts_of(w_off, res.versionId, ctype)),
|
||
bool(_blockers_of(w_off, res.versionId, ctype)))
|
||
return (*on, *off)
|
||
|
||
|
||
@pytest.mark.parametrize("engine", ENGINES)
|
||
def test_equipment_hard_constraint_on_off_consistent_per_engine(engine):
|
||
on_c, on_b, off_c, off_b = _semantics(engine, _full_horizon_maintenance_world,
|
||
"EQUIPMENT", "equipment")
|
||
assert on_c and on_b, f"{engine}: 开启 EQUIPMENT 硬约束应出现冲突并拦发布"
|
||
assert not off_c and not off_b, f"{engine}: 关闭 EQUIPMENT 后冲突应消失"
|
||
|
||
|
||
@pytest.mark.parametrize("engine", ENGINES)
|
||
def test_capacity_hard_constraint_on_off_consistent_per_engine(engine):
|
||
on_c, on_b, off_c, off_b = _semantics(engine, _single_order_world,
|
||
"CAPACITY", "capacity")
|
||
assert on_c and on_b, f"{engine}: 开启 CAPACITY 硬约束应出现冲突并拦发布"
|
||
assert not off_c and not off_b, f"{engine}: 关闭 CAPACITY 后冲突应消失"
|
||
|
||
|
||
@pytest.mark.parametrize("engine", ENGINES)
|
||
def test_material_kit_hard_soft_off_consistent_per_engine(engine):
|
||
# 硬：MAJOR 冲突且拦发布
|
||
on_c, on_b, off_c, off_b = _semantics(
|
||
engine, _zero_stock_world, "MATERIAL_SHORTAGE", "materialKit",
|
||
patch={"C6_material_kit": {"kind": "hard", "enabled": True}})
|
||
assert on_c and on_b, f"{engine}: C6 硬约束开启应出现 MAJOR 冲突并拦发布"
|
||
assert not off_c and not off_b, f"{engine}: C6 关闭后缺料冲突应消失"
|
||
|
||
|
||
@pytest.mark.parametrize("engine", ENGINES)
|
||
def test_material_kit_soft_reports_risk_without_blocking(engine):
|
||
base = _zero_stock_world()
|
||
flags = engine_constraint_flags(base) # C6 默认软
|
||
w = copy.deepcopy(base)
|
||
res = _run(w, engine, flags)
|
||
conflicts = _conflicts_of(w, res.versionId, "MATERIAL_SHORTAGE")
|
||
blockers = _blockers_of(w, res.versionId, "MATERIAL_SHORTAGE")
|
||
assert conflicts and all(c["severity"] == "MINOR" for c in conflicts)
|
||
assert not blockers
|
||
|
||
|
||
def test_all_engines_agree_on_hard_constraint_semantics():
|
||
"""跨引擎一致性：同一世界 + 同一 flags，四引擎开/关语义完全一致。"""
|
||
for ctype, factory, flag_key, patch in [
|
||
("EQUIPMENT", _full_horizon_maintenance_world, "equipment", None),
|
||
("CAPACITY", _single_order_world, "capacity", None),
|
||
("MATERIAL_SHORTAGE", _zero_stock_world, "materialKit",
|
||
{"C6_material_kit": {"kind": "hard", "enabled": True}}),
|
||
]:
|
||
outcomes = {}
|
||
for engine in ENGINES:
|
||
outcomes[engine] = _semantics(engine, factory, ctype, flag_key, patch)
|
||
first = outcomes["RULE"]
|
||
assert first == (True, True, False, False), f"{ctype} RULE 基线异常: {first}"
|
||
for engine in ENGINES[1:]:
|
||
assert outcomes[engine] == first, (
|
||
f"{ctype}: {engine} 与 RULE 语义不一致 {outcomes[engine]} != {first}")
|
||
|
||
|
||
def test_soft_weight_cannot_close_hard_constraint_end_to_end():
|
||
"""软权重归零被门禁拒绝 → flags 仍开启 → 四引擎仍拦发布。"""
|
||
world = seed_world()
|
||
with pytest.raises(ConstraintProfileDenied):
|
||
apply_profile_save(world, {"constraints": {"C7_capacity": {"weight": 0}}})
|
||
assert engine_constraint_flags(world)["capacity"] is True
|
||
base = _single_order_world()
|
||
results = {}
|
||
for engine in ENGINES:
|
||
w = copy.deepcopy(base)
|
||
res = _run(w, engine, engine_constraint_flags(world))
|
||
results[engine] = bool(_blockers_of(w, res.versionId, "CAPACITY"))
|
||
assert results == {e: True for e in ENGINES}
|