297 lines
13 KiB
Python
297 lines
13 KiB
Python
# ============================================================
|
||
# M5-89 紧约束/影子价、IIS 与冲突归因 黄金测试
|
||
# 验收口径：对不可行问题产出最小冲突集（删除任一约束后可恢复可行）；
|
||
# 可行输入返回空 + 明确说明；主控参数结论可由求解日志复核；
|
||
# 归因文案为现场计划员可读的中文。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import pytest
|
||
|
||
from server.aps_domain.attribution import (
|
||
build_attribution_report,
|
||
detect_iis,
|
||
feasibility_after_relax,
|
||
hard_violations,
|
||
relief_constraints,
|
||
)
|
||
from server.aps_domain.constraints import (
|
||
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, today0
|
||
|
||
|
||
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, strategy="COMPREHENSIVE"):
|
||
# 黄金用例固定以当前验收基线 2026-08-02 的次日开排，禁止跨午夜漂移。
|
||
start = "2026-08-03"
|
||
params = EngineParams(
|
||
orderIds=[], engineType="RULE", strategyTemplate=strategy,
|
||
planningHorizonDays=14, startDate=start,
|
||
constraints=engine_constraint_flags(world),
|
||
)
|
||
return get_engine("RULE").solve(world, params, _next_id_factory())
|
||
|
||
|
||
def _resolve_all(world):
|
||
for c in world["conflicts"]:
|
||
c["isResolved"] = True
|
||
|
||
|
||
def _inject_conflict(world, vid, **fields):
|
||
"""注入一条冲突记录（与现有测试同款手法），id 自动续号。"""
|
||
cid = max((c.get("id") or 0 for c in world["conflicts"]), default=0) + 1
|
||
rec = {"id": cid, "versionId": vid, "severity": "MAJOR",
|
||
"isResolved": False, "resolutionAction": ""}
|
||
rec.update(fields)
|
||
world["conflicts"].append(rec)
|
||
return rec
|
||
|
||
|
||
def _first_capacity_on(world):
|
||
"""取本次求解产生的首个产能冲突，不绑定机器当前日期。"""
|
||
return next(
|
||
c for c in world["conflicts"]
|
||
if c["conflictType"] == "CAPACITY"
|
||
)
|
||
|
||
|
||
def _production_order_on_capacity_line(world, capacity_conflict):
|
||
line = next(
|
||
row for row in world["lines"]
|
||
if row["name"] == capacity_conflict["resourceName"]
|
||
)
|
||
return next(
|
||
row["orderNo"] for row in world["productionOrders"]
|
||
if row["lineId"] == line["id"]
|
||
)
|
||
|
||
|
||
def test_coupled_capacity_due_date_produces_minimal_conflict_set():
|
||
"""资源超额 + 交期冲突（联合成因）→ 最小冲突集 {C7, C8}，
|
||
删除其中任一约束后恢复可行（不可约性质）。"""
|
||
world = seed_world()
|
||
result = _run(world)
|
||
vid = result.versionId
|
||
# 交期冲突升级为硬约束（否则 DELAY 只是风险，不构成不可行）
|
||
apply_profile_save(world, {"constraints": {"C8_due_date": {"kind": "hard", "enabled": True}}})
|
||
cap = _first_capacity_on(world)
|
||
order_no = _production_order_on_capacity_line(world, cap)
|
||
cap["reliefConstraintIds"] = ["C7_capacity", "C8_due_date"]
|
||
for c in world["conflicts"]:
|
||
if c["id"] != cap["id"]:
|
||
c["isResolved"] = True
|
||
_inject_conflict(
|
||
world, vid,
|
||
conflictType="DELAY", resourceType="TIME", orderNo=order_no,
|
||
description=f"{order_no} 完成时间晚于交期 6.0 小时",
|
||
suggestedSolution="分流至替代产线或启用加班",
|
||
reliefConstraintIds=["C8_due_date", "C7_capacity"],
|
||
)
|
||
report = build_attribution_report(world, version_id=vid)
|
||
assert report["feasible"] is False
|
||
assert report["iis"], "不可行输入必须产出非空最小冲突集"
|
||
group = next(
|
||
g for g in report["iis"]
|
||
if set(g["constraintIds"]) == {"C7_capacity", "C8_due_date"}
|
||
)
|
||
# 不可约：删除其中任一约束后该冲突组恢复可行
|
||
assert len(group["verification"]) == 2
|
||
for v in group["verification"]:
|
||
assert v["feasibleAfterDelete"] is True
|
||
assert feasibility_after_relax(world, ["C7_capacity"], version_id=vid) is True
|
||
assert feasibility_after_relax(world, ["C8_due_date"], version_id=vid) is True
|
||
# 违反度可从求解日志描述复核
|
||
cap_att = next(a for a in report["attributions"] if a["conflictId"] == cap["id"])
|
||
assert cap_att["degree"]["kind"] == "overload_minutes"
|
||
assert cap_att["degree"]["overloadMinutes"] > 0
|
||
assert cap_att["reliefConstraintIds"] == ["C7_capacity", "C8_due_date"]
|
||
delay_att = next(
|
||
a for a in report["attributions"] if a["conflictType"] == "DELAY")
|
||
assert delay_att["degree"]["kind"] == "hours_late"
|
||
assert delay_att["degree"]["hoursLate"] == 6.0
|
||
|
||
|
||
def test_feasible_input_returns_empty_iis():
|
||
"""可行输入 → 空结果 + 明确中文说明。"""
|
||
world = seed_world()
|
||
result = _run(world)
|
||
_resolve_all(world)
|
||
rep = detect_iis(world, version_id=result.versionId)
|
||
assert rep["feasible"] is True
|
||
assert rep["iis"] == []
|
||
assert "可行" in rep["message"]
|
||
report = build_attribution_report(world, version_id=result.versionId)
|
||
assert report["feasible"] is True
|
||
assert report["iis"] == []
|
||
assert report["attributions"] == []
|
||
assert report["masterControl"] == []
|
||
assert "可行" in report["summary"]
|
||
|
||
|
||
def test_auto_enrichment_delay_root_cause_is_capacity():
|
||
"""求解日志联合成因：订单延误 + 所在产线当日产能冲突 →
|
||
自动把 C7 产能加入 DELAY 的解除约束集合，IIS 指向产能约束。"""
|
||
world = seed_world()
|
||
result = _run(world)
|
||
vid = result.versionId
|
||
apply_profile_save(world, {"constraints": {"C8_due_date": {"kind": "hard", "enabled": True}}})
|
||
cap = _first_capacity_on(world)
|
||
order_no = _production_order_on_capacity_line(world, cap)
|
||
for c in world["conflicts"]:
|
||
if c["id"] != cap["id"]:
|
||
c["isResolved"] = True
|
||
delay = _inject_conflict(
|
||
world, vid,
|
||
conflictType="DELAY", resourceType="TIME", orderNo=order_no,
|
||
description=f"{order_no} 完成时间晚于交期 6.0 小时",
|
||
suggestedSolution="分流至替代产线或启用加班",
|
||
)
|
||
relief = relief_constraints(world, delay, version_id=vid)
|
||
assert "C8_due_date" in relief
|
||
assert "C7_capacity" in relief, "延误且所在产线产能冲突 → 产能应为联合成因"
|
||
d = detect_iis(world, version_id=vid)
|
||
assert d["feasible"] is False
|
||
assert any(set(g["constraintIds"]) == {"C7_capacity"} for g in d["iis"])
|
||
# 删除 C7（该 IIS 的唯一成员）→ 恢复可行（联合成因的 DELAY 一并解除）
|
||
assert feasibility_after_relax(world, ["C7_capacity"], version_id=vid) is True
|
||
# 归因文案点明产能是延误根源
|
||
report = build_attribution_report(world, version_id=vid)
|
||
delay_att = next(a for a in report["attributions"] if a["conflictType"] == "DELAY")
|
||
assert "C7_capacity" in delay_att["reliefConstraintIds"]
|
||
assert "产能" in report["masterControl"][0]["conclusion"] or any(
|
||
"产能" in m["conclusion"] for m in report["masterControl"])
|
||
|
||
|
||
def test_copy_readability_chinese_fields():
|
||
"""文案可读性：中文关键字段（约束名/订单号/资源名/短日期/建议）。"""
|
||
world = seed_world()
|
||
result = _run(world)
|
||
vid = result.versionId
|
||
apply_profile_save(world, {"constraints": {"C8_due_date": {"kind": "hard", "enabled": True}}})
|
||
cap = _first_capacity_on(world)
|
||
order_no = _production_order_on_capacity_line(world, cap)
|
||
cap["reliefConstraintIds"] = ["C7_capacity", "C8_due_date"]
|
||
for c in world["conflicts"]:
|
||
if c["id"] != cap["id"]:
|
||
c["isResolved"] = True
|
||
_inject_conflict(
|
||
world, vid,
|
||
conflictType="DELAY", resourceType="TIME", orderNo=order_no,
|
||
description=f"{order_no} 完成时间晚于交期 6.0 小时",
|
||
suggestedSolution="分流至替代产线或启用加班",
|
||
reliefConstraintIds=["C8_due_date", "C7_capacity"],
|
||
)
|
||
report = build_attribution_report(world, version_id=vid)
|
||
group = next(
|
||
g for g in report["iis"]
|
||
if set(g["constraintIds"]) == {"C7_capacity", "C8_due_date"}
|
||
)
|
||
iis_text = group["copy"]
|
||
assert "约束" in iis_text and "卡死" in iis_text
|
||
assert "订单" in iis_text and "资源" in iis_text and "建议" in iis_text
|
||
assert "装配产线A" in iis_text # 资源名（中文）
|
||
assert order_no in iis_text # 订单号
|
||
assert "8/3" in iis_text # 短日期
|
||
# 单条归因文案
|
||
att_text = report["attributions"][0]["copy"]
|
||
assert "约束" in att_text and "被违反" in att_text
|
||
assert "涉及订单" in att_text or "资源/对象" in att_text
|
||
assert "解开方式" in att_text
|
||
# 主控参数结论文案
|
||
assert report["masterControl"]
|
||
assert any("紧约束" in m["conclusion"] for m in report["masterControl"])
|
||
# 文案面向现场计划员：不出现纯数学术语
|
||
assert "IIS" not in iis_text and "shadow" not in iis_text.lower()
|
||
assert "IIS" not in att_text
|
||
|
||
|
||
def test_attribution_evidence_verifiable_against_solve_log():
|
||
"""主控参数结论可由求解日志复核：evidence 携带冲突 ID/版本/描述。"""
|
||
world = seed_world()
|
||
result = _run(world)
|
||
vid = result.versionId
|
||
report = build_attribution_report(world, version_id=vid)
|
||
assert report["feasible"] is False
|
||
conflict_ids = {c["id"] for c in world["conflicts"] if c["versionId"] == vid}
|
||
for att in report["attributions"]:
|
||
ev = att["evidence"]
|
||
assert ev["conflictId"] in conflict_ids
|
||
assert ev["versionId"] == vid
|
||
assert ev["description"]
|
||
assert report["masterControl"]
|
||
for m in report["masterControl"]:
|
||
assert m["evidenceRefs"], "主控参数结论必须携带可复核的日志证据"
|
||
for ref in m["evidenceRefs"]:
|
||
cid = int(ref.split("#")[1])
|
||
assert cid in conflict_ids
|
||
for key, entry in report["evidence"].items():
|
||
assert key == f"conflict#{entry['conflictId']}"
|
||
assert entry["description"]
|
||
# 与 world["conflicts"] 逐条可对照
|
||
for att in report["attributions"]:
|
||
raw = next(c for c in world["conflicts"] if c["id"] == att["conflictId"])
|
||
assert att["description"] == raw["description"]
|
||
|
||
|
||
def test_delete_constraint_restores_publish_gate_for_singleton_iis():
|
||
"""默认演示运行产生两组单约束 IIS（C4 维保 / C7 产能）：
|
||
每组内任删一条即恢复该组可行；硬约束关闭请求被门禁拒绝（矩阵 113），
|
||
发布门禁仍同时拦截产能与维保；只有松弛（simulated relax）才能恢复可行。"""
|
||
from server.aps_domain.constraints import ConstraintProfileDenied
|
||
|
||
world = seed_world()
|
||
result = _run(world)
|
||
vid = result.versionId
|
||
d = detect_iis(world, version_id=vid)
|
||
assert d["feasible"] is False
|
||
groups = {g["constraintIds"][0]: g for g in d["iis"]}
|
||
assert "C7_capacity" in groups and "C4_maintenance" in groups
|
||
for g in groups.values():
|
||
assert len(g["constraintIds"]) == 1
|
||
for v in g["verification"]:
|
||
assert v["feasibleAfterDelete"] is True
|
||
# 物理关闭 C7 被拒（硬约束不可关闭）→ 发布门禁仍拦截 CAPACITY 与 EQUIPMENT
|
||
with pytest.raises(ConstraintProfileDenied):
|
||
apply_profile_save(world, {"constraints": {"C7_capacity": {"enabled": False}}})
|
||
blockers = hard_blocking_conflicts(world, version_id=vid)
|
||
assert any(b["constraintId"] == "C7_capacity" for b in blockers)
|
||
assert any(b["constraintId"] == "C4_maintenance" for b in blockers)
|
||
# 从每个 IIS 各松一条 → 整体可行
|
||
assert feasibility_after_relax(
|
||
world, ["C7_capacity", "C4_maintenance"], version_id=vid) is True
|
||
|
||
|
||
def test_hard_violations_expose_solve_log_fields():
|
||
"""硬违反记录可直接展示并对照求解日志：约束/实体/违反度/证据。"""
|
||
world = seed_world()
|
||
result = _run(world)
|
||
vid = result.versionId
|
||
rows = hard_violations(world, version_id=vid)
|
||
assert rows
|
||
for r in rows:
|
||
assert r["constraintId"] in {"C7_capacity", "C4_maintenance"}
|
||
assert r["constraintName"]
|
||
assert r["evidence"]["conflictId"] == r["conflictId"]
|
||
assert r["evidence"]["versionId"] == vid
|
||
assert r["degree"]["value"] > 0
|
||
capacity = next(r for r in rows if r["constraintId"] == "C7_capacity")
|
||
assert capacity["degree"]["kind"] == "overload_minutes"
|
||
assert capacity["degree"]["overloadMinutes"] > 0
|
||
equipment = next(r for r in rows if r["constraintId"] == "C4_maintenance")
|
||
assert equipment["degree"]["kind"] == "overlap_minutes"
|
||
assert equipment["orderNos"], "维保冲突应能解析出涉及的订单号" |