# ============================================================ # 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"], "维保冲突应能解析出涉及的订单号"