507 lines
18 KiB
Python
507 lines
18 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import date, timedelta
|
||
|
|
from itertools import combinations
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
import server.engines.cp_engine as cp_engine
|
||
|
|
from server.engines.base import EngineParams
|
||
|
|
from server.engines.cp_engine import CpSatEngine, optimize_line_assignment
|
||
|
|
from server.engines.queries import find_routing_steps, find_workstation_for_operation
|
||
|
|
from server.engines.solver_process import SolverProcessError
|
||
|
|
from server.aps_domain.constraints import hard_blocking_conflicts
|
||
|
|
from server.state.seed import build_demo_world
|
||
|
|
from server.timeutil import add_minutes, fmt_dt, parse_dt
|
||
|
|
|
||
|
|
|
||
|
|
_C3_SCHEMA = "cp-calendar-segmented.v1"
|
||
|
|
_CALENDAR_MODE = "calendar-boundary-only"
|
||
|
|
_CONTINUOUS_MODE = "continuous"
|
||
|
|
_SLOT_CONTRACT = {
|
||
|
|
"segments",
|
||
|
|
"processingMinutes",
|
||
|
|
"elapsedSpanMinutes",
|
||
|
|
"pauseMinutes",
|
||
|
|
"segmentCount",
|
||
|
|
"calendarCompliant",
|
||
|
|
"calendarMode",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _calendar_date(world: dict, line_id: int, *, working: bool) -> str:
|
||
|
|
return min(
|
||
|
|
row["date"]
|
||
|
|
for row in world["shiftCalendar"]
|
||
|
|
if row["lineId"] == line_id and bool(row["isWorking"]) is working
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _single_line_world(
|
||
|
|
*,
|
||
|
|
quantity: int,
|
||
|
|
orders: int = 1,
|
||
|
|
start_date: str | None = None,
|
||
|
|
day_shift_only: bool = False,
|
||
|
|
) -> tuple[dict, str]:
|
||
|
|
world = build_demo_world()
|
||
|
|
resolved_start = start_date or _calendar_date(world, 1, working=True)
|
||
|
|
selected = [
|
||
|
|
row for row in world["salesOrders"]
|
||
|
|
if row["items"][0]["productId"] == 1
|
||
|
|
][:orders]
|
||
|
|
assert len(selected) == orders
|
||
|
|
for order in selected:
|
||
|
|
order["deliveryDate"] = resolved_start
|
||
|
|
order["items"][0]["quantity"] = quantity
|
||
|
|
world["salesOrders"] = selected
|
||
|
|
world["lineProducts"] = [
|
||
|
|
row for row in world["lineProducts"]
|
||
|
|
if row["productId"] == 1 and row["lineId"] == 1
|
||
|
|
]
|
||
|
|
routing_id = next(
|
||
|
|
row["id"] for row in world["routings"]
|
||
|
|
if row["productId"] == 1 and row["isDefault"]
|
||
|
|
)
|
||
|
|
world["routingSteps"] = [
|
||
|
|
row for row in world["routingSteps"]
|
||
|
|
if row["routingId"] == routing_id and row["sequenceNo"] == 1
|
||
|
|
]
|
||
|
|
if day_shift_only:
|
||
|
|
for row in world["shiftCalendar"]:
|
||
|
|
if row["lineId"] == 1 and row["shiftId"] != 1:
|
||
|
|
row["isWorking"] = False
|
||
|
|
world["scheduleParams"]["freezeWindowHours"] = 0
|
||
|
|
world["scheduleParams"]["deliveryBufferRatio"] = 1.0
|
||
|
|
return world, resolved_start
|
||
|
|
|
||
|
|
|
||
|
|
def _params(start_date: str, *, cumulative: bool = False) -> EngineParams:
|
||
|
|
return EngineParams(
|
||
|
|
orderIds=[],
|
||
|
|
engineType="CP",
|
||
|
|
strategyTemplate="COMPREHENSIVE",
|
||
|
|
planningHorizonDays=3,
|
||
|
|
startDate=start_date,
|
||
|
|
timeLimitSeconds=3,
|
||
|
|
freezeWindowHours=0,
|
||
|
|
deliveryBufferRatio=1.0,
|
||
|
|
constraints={
|
||
|
|
"capacity": False,
|
||
|
|
"changeover": False,
|
||
|
|
"personnel": cumulative,
|
||
|
|
"tooling": cumulative,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def _diagnose(
|
||
|
|
world: dict,
|
||
|
|
start_date: str,
|
||
|
|
*,
|
||
|
|
relaxed_constraint_id: str | None = None,
|
||
|
|
cumulative: bool = False,
|
||
|
|
) -> dict:
|
||
|
|
params = _params(start_date, cumulative=cumulative)
|
||
|
|
entries, _, _ = CpSatEngine().collect_and_order(world, params)
|
||
|
|
_, meta = optimize_line_assignment(
|
||
|
|
world,
|
||
|
|
entries,
|
||
|
|
params,
|
||
|
|
pipeline_label=(
|
||
|
|
"C3-CALENDAR-BASELINE"
|
||
|
|
if relaxed_constraint_id is None
|
||
|
|
else "C3-CALENDAR-REMOVAL"
|
||
|
|
),
|
||
|
|
relaxed_constraint_ids=(
|
||
|
|
[] if relaxed_constraint_id is None else [relaxed_constraint_id]
|
||
|
|
),
|
||
|
|
diagnostic_mode=True,
|
||
|
|
)
|
||
|
|
assert meta["status"] in {"OPTIMAL", "FEASIBLE"}
|
||
|
|
return meta
|
||
|
|
|
||
|
|
|
||
|
|
def _active_slots(meta: dict) -> list[dict]:
|
||
|
|
return [
|
||
|
|
slot for slot in meta["operationSlots"]
|
||
|
|
if not slot.get("isFrozen") and slot.get("lineId") is not None
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def _assert_slot_contract(slot: dict) -> list[dict]:
|
||
|
|
assert _SLOT_CONTRACT <= set(slot)
|
||
|
|
segments = slot["segments"]
|
||
|
|
assert isinstance(segments, list) and segments
|
||
|
|
assert slot["segmentCount"] == len(segments)
|
||
|
|
assert all(
|
||
|
|
isinstance(segment.get("startMin"), int)
|
||
|
|
and isinstance(segment.get("endMin"), int)
|
||
|
|
and segment["startMin"] < segment["endMin"]
|
||
|
|
for segment in segments
|
||
|
|
)
|
||
|
|
assert segments == sorted(segments, key=lambda segment: segment["startMin"])
|
||
|
|
processing = sum(segment["endMin"] - segment["startMin"] for segment in segments)
|
||
|
|
elapsed = segments[-1]["endMin"] - segments[0]["startMin"]
|
||
|
|
assert slot["processingMinutes"] == processing
|
||
|
|
assert slot["elapsedSpanMinutes"] == elapsed
|
||
|
|
assert slot["pauseMinutes"] == elapsed - processing
|
||
|
|
assert slot["startMin"] == segments[0]["startMin"]
|
||
|
|
assert slot["endMin"] == segments[-1]["endMin"]
|
||
|
|
return segments
|
||
|
|
|
||
|
|
|
||
|
|
def _assert_no_overlap(intervals: list[tuple[int, int]]) -> None:
|
||
|
|
for left, right in combinations(intervals, 2):
|
||
|
|
assert left[1] <= right[0] or right[1] <= left[0], (left, right)
|
||
|
|
|
||
|
|
|
||
|
|
def _counter():
|
||
|
|
values: dict[str, int] = {}
|
||
|
|
|
||
|
|
def next_id(kind: str) -> int:
|
||
|
|
values[kind] = values.get(kind, 0) + 1
|
||
|
|
return values[kind]
|
||
|
|
|
||
|
|
return next_id
|
||
|
|
|
||
|
|
|
||
|
|
def test_short_operation_uses_one_compliant_calendar_segment() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=60)
|
||
|
|
|
||
|
|
meta = _diagnose(world, start_date)
|
||
|
|
|
||
|
|
c3 = meta["c3Calendar"]
|
||
|
|
assert c3["schemaVersion"] == _C3_SCHEMA
|
||
|
|
assert meta["constraintInstanceCounts"]["C3_calendar"] > 0
|
||
|
|
assert "C3_calendar" in meta["activeAssumptionConstraints"]
|
||
|
|
slot = _active_slots(meta)[0]
|
||
|
|
_assert_slot_contract(slot)
|
||
|
|
assert slot["processingMinutes"] == 60
|
||
|
|
assert slot["elapsedSpanMinutes"] == 60
|
||
|
|
assert slot["pauseMinutes"] == 0
|
||
|
|
assert slot["segmentCount"] == 1
|
||
|
|
assert slot["calendarCompliant"] is True
|
||
|
|
assert slot["calendarMode"] == _CALENDAR_MODE
|
||
|
|
|
||
|
|
|
||
|
|
def test_operation_pauses_only_at_calendar_break_boundary() -> None:
|
||
|
|
world, start_date = _single_line_world(
|
||
|
|
quantity=500,
|
||
|
|
day_shift_only=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
meta = _diagnose(world, start_date)
|
||
|
|
|
||
|
|
slot = _active_slots(meta)[0]
|
||
|
|
segments = _assert_slot_contract(slot)
|
||
|
|
assert slot["processingMinutes"] == 280
|
||
|
|
assert slot["segmentCount"] == 2
|
||
|
|
assert slot["pauseMinutes"] == 60
|
||
|
|
assert segments[0]["endMin"] <= 240
|
||
|
|
assert segments[1]["startMin"] >= 300
|
||
|
|
assert slot["calendarCompliant"] is True
|
||
|
|
assert slot["calendarMode"] == _CALENDAR_MODE
|
||
|
|
|
||
|
|
|
||
|
|
def test_nonworking_start_day_defers_all_processing_to_next_working_window() -> None:
|
||
|
|
probe = build_demo_world()
|
||
|
|
start_date = _calendar_date(probe, 1, working=False)
|
||
|
|
next_working_date = min(
|
||
|
|
row["date"]
|
||
|
|
for row in probe["shiftCalendar"]
|
||
|
|
if row["lineId"] == 1
|
||
|
|
and row["isWorking"]
|
||
|
|
and row["date"] > start_date
|
||
|
|
)
|
||
|
|
world, _ = _single_line_world(quantity=60, start_date=start_date)
|
||
|
|
|
||
|
|
meta = _diagnose(world, start_date)
|
||
|
|
|
||
|
|
slot = _active_slots(meta)[0]
|
||
|
|
segments = _assert_slot_contract(slot)
|
||
|
|
next_working_offset = (
|
||
|
|
date.fromisoformat(next_working_date) - date.fromisoformat(start_date)
|
||
|
|
).days * 24 * 60
|
||
|
|
assert all(segment["startMin"] >= next_working_offset for segment in segments)
|
||
|
|
assert slot["calendarCompliant"] is True
|
||
|
|
assert slot["calendarMode"] == _CALENDAR_MODE
|
||
|
|
|
||
|
|
|
||
|
|
def test_c3_removal_restores_one_continuous_interval() -> None:
|
||
|
|
world, start_date = _single_line_world(
|
||
|
|
quantity=500,
|
||
|
|
day_shift_only=True,
|
||
|
|
)
|
||
|
|
baseline = _diagnose(world, start_date)
|
||
|
|
|
||
|
|
relaxed = _diagnose(
|
||
|
|
world,
|
||
|
|
start_date,
|
||
|
|
relaxed_constraint_id="C3_calendar",
|
||
|
|
)
|
||
|
|
|
||
|
|
baseline_slot = _active_slots(baseline)[0]
|
||
|
|
relaxed_slot = _active_slots(relaxed)[0]
|
||
|
|
_assert_slot_contract(baseline_slot)
|
||
|
|
_assert_slot_contract(relaxed_slot)
|
||
|
|
assert baseline_slot["segmentCount"] == 2
|
||
|
|
assert baseline_slot["pauseMinutes"] == 60
|
||
|
|
assert relaxed["relaxedConstraintIds"] == ["C3_calendar"]
|
||
|
|
assert baseline["c3Calendar"]["selectedMode"] == _CALENDAR_MODE
|
||
|
|
assert relaxed["c3Calendar"]["selectedMode"] == _CONTINUOUS_MODE
|
||
|
|
assert relaxed_slot["calendarMode"] == _CONTINUOUS_MODE
|
||
|
|
assert relaxed_slot["segmentCount"] == 1
|
||
|
|
assert relaxed_slot["pauseMinutes"] == 0
|
||
|
|
assert relaxed_slot["elapsedSpanMinutes"] == relaxed_slot["processingMinutes"] == 280
|
||
|
|
assert relaxed_slot["calendarCompliant"] is False
|
||
|
|
|
||
|
|
|
||
|
|
def _shared_resource_world() -> tuple[dict, str]:
|
||
|
|
world = build_demo_world()
|
||
|
|
start_date = _calendar_date(world, 1, working=True)
|
||
|
|
selected = [
|
||
|
|
row for row in world["salesOrders"]
|
||
|
|
if row["items"][0]["productId"] == 1
|
||
|
|
][:2]
|
||
|
|
selected += [
|
||
|
|
row for row in world["salesOrders"]
|
||
|
|
if row["items"][0]["productId"] == 3
|
||
|
|
][:1]
|
||
|
|
assert len(selected) == 3
|
||
|
|
for order in selected:
|
||
|
|
order["deliveryDate"] = start_date
|
||
|
|
order["items"][0]["quantity"] = 500
|
||
|
|
world["salesOrders"] = selected
|
||
|
|
world["lineProducts"] = [
|
||
|
|
row for row in world["lineProducts"]
|
||
|
|
if (row["productId"], row["lineId"]) in {(1, 1), (3, 3)}
|
||
|
|
]
|
||
|
|
routing_ids = {
|
||
|
|
row["id"] for row in world["routings"]
|
||
|
|
if row["productId"] in {1, 3} and row["isDefault"]
|
||
|
|
}
|
||
|
|
world["routingSteps"] = [
|
||
|
|
row for row in world["routingSteps"]
|
||
|
|
if row["routingId"] in routing_ids and row["sequenceNo"] == 1
|
||
|
|
]
|
||
|
|
for row in world["shiftCalendar"]:
|
||
|
|
if row["lineId"] in {1, 3} and row["shiftId"] != 1:
|
||
|
|
row["isWorking"] = False
|
||
|
|
for workstation in world["workstations"]:
|
||
|
|
if workstation["code"] in {"WS001", "WS010"}:
|
||
|
|
workstation["teamId"] = 1
|
||
|
|
workstation["toolingId"] = 1
|
||
|
|
team = next(row for row in world["teams"] if row["id"] == 1)
|
||
|
|
team["memberCount"] = 1
|
||
|
|
team["availableCount"] = 1
|
||
|
|
world["toolings"] = [{
|
||
|
|
"id": 1,
|
||
|
|
"code": "TOOL-1",
|
||
|
|
"name": "Shared tool",
|
||
|
|
"availableCount": 1,
|
||
|
|
}]
|
||
|
|
world["scheduleParams"]["freezeWindowHours"] = 0
|
||
|
|
world["scheduleParams"]["deliveryBufferRatio"] = 1.0
|
||
|
|
return world, start_date
|
||
|
|
|
||
|
|
|
||
|
|
def test_c2_and_c12_apply_to_processing_segments_not_elapsed_spans() -> None:
|
||
|
|
world, start_date = _shared_resource_world()
|
||
|
|
|
||
|
|
meta = _diagnose(world, start_date, cumulative=True)
|
||
|
|
|
||
|
|
slots = _active_slots(meta)
|
||
|
|
assert len(slots) == 3
|
||
|
|
by_workstation: dict[int, list[tuple[int, int]]] = {}
|
||
|
|
by_team: dict[int, list[tuple[int, int]]] = {}
|
||
|
|
by_tooling: dict[int, list[tuple[int, int]]] = {}
|
||
|
|
workstation_by_id = {row["id"]: row for row in world["workstations"]}
|
||
|
|
for slot in slots:
|
||
|
|
segments = _assert_slot_contract(slot)
|
||
|
|
intervals = [(segment["startMin"], segment["endMin"]) for segment in segments]
|
||
|
|
workstation_id = int(slot["workstationId"])
|
||
|
|
by_workstation.setdefault(workstation_id, []).extend(intervals)
|
||
|
|
workstation = workstation_by_id[workstation_id]
|
||
|
|
if workstation.get("teamId") is not None:
|
||
|
|
by_team.setdefault(int(workstation["teamId"]), []).extend(intervals)
|
||
|
|
if workstation.get("toolingId") is not None:
|
||
|
|
by_tooling.setdefault(int(workstation["toolingId"]), []).extend(intervals)
|
||
|
|
for intervals in by_workstation.values():
|
||
|
|
_assert_no_overlap(intervals)
|
||
|
|
_assert_no_overlap(by_team[1])
|
||
|
|
_assert_no_overlap(by_tooling[1])
|
||
|
|
cumulative = meta["cumulative"]
|
||
|
|
team_resource = next(
|
||
|
|
row for row in cumulative["resources"]
|
||
|
|
if row["kind"] == "team" and row["id"] == 1
|
||
|
|
)
|
||
|
|
tooling_resource = next(
|
||
|
|
row for row in cumulative["resources"]
|
||
|
|
if row["kind"] == "tooling" and row["id"] == 1
|
||
|
|
)
|
||
|
|
assert team_resource["peakConcurrent"] == 1
|
||
|
|
assert tooling_resource["peakConcurrent"] == 1
|
||
|
|
|
||
|
|
|
||
|
|
def test_c3_metadata_reports_schema_and_model_limit() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=60)
|
||
|
|
|
||
|
|
meta = _diagnose(world, start_date)
|
||
|
|
|
||
|
|
c3 = meta["c3Calendar"]
|
||
|
|
assert c3["schemaVersion"] == _C3_SCHEMA
|
||
|
|
assert c3["calendarMode"] == _CALENDAR_MODE
|
||
|
|
assert c3["modelSegmentLimit"] == 50_000
|
||
|
|
assert 0 < c3["modelSegmentCount"] <= c3["modelSegmentLimit"]
|
||
|
|
assert c3["calendarCompliant"] is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_single_logical_operation_does_not_activate_c2() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=60)
|
||
|
|
|
||
|
|
meta = _diagnose(world, start_date)
|
||
|
|
|
||
|
|
assert meta["constraintInstanceCounts"]["C2_no_overlap"] == 0
|
||
|
|
assert "C2_no_overlap" not in meta["activeAssumptionConstraints"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_zero_objective_hint_does_not_fix_across_frozen_obstacle() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=60)
|
||
|
|
world["salesOrders"][0]["deliveryDate"] = (
|
||
|
|
date.fromisoformat(start_date) + timedelta(days=10)
|
||
|
|
).isoformat()
|
||
|
|
step = find_routing_steps(world, 1)[0]
|
||
|
|
workstation = find_workstation_for_operation(world, 1, step["operationId"])
|
||
|
|
assert workstation is not None
|
||
|
|
anchor = parse_dt(start_date + " 08:00")
|
||
|
|
world["workOrders"].append({
|
||
|
|
"id": 99001,
|
||
|
|
"orderNo": "WO-FROZEN-CROSS-FREEZE",
|
||
|
|
"lineId": 1,
|
||
|
|
"workstationId": workstation["id"],
|
||
|
|
"plannedStartTime": fmt_dt(add_minutes(anchor, 50 * 60)),
|
||
|
|
"plannedEndTime": fmt_dt(add_minutes(anchor, 70 * 60)),
|
||
|
|
"isFrozen": True,
|
||
|
|
})
|
||
|
|
params = _params(start_date).model_copy(update={"freezeWindowHours": 60})
|
||
|
|
entries, _, _ = CpSatEngine().collect_and_order(world, params)
|
||
|
|
|
||
|
|
_, meta = optimize_line_assignment(
|
||
|
|
world, entries, params, pipeline_label="C3-FROZEN-HINT", diagnostic_mode=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
assert meta["status"] in {"OPTIMAL", "FEASIBLE"}
|
||
|
|
assert meta["zeroObjectiveHintCandidate"] is True
|
||
|
|
assert meta["zeroObjectiveHintFixed"] is False
|
||
|
|
slot = _active_slots(meta)[0]
|
||
|
|
assert slot["startMin"] >= 70 * 60
|
||
|
|
assert meta["constraintInstanceCounts"]["C2_no_overlap"] > 0
|
||
|
|
|
||
|
|
|
||
|
|
def test_c3_model_rejects_more_than_50k_segment_decisions(monkeypatch) -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=60)
|
||
|
|
params = _params(start_date)
|
||
|
|
entries, _, _ = CpSatEngine().collect_and_order(world, params)
|
||
|
|
original = cp_engine._line_day_buckets
|
||
|
|
|
||
|
|
def oversized_buckets(world_arg, line_ids, anchor, horizon):
|
||
|
|
buckets = original(world_arg, line_ids, anchor, horizon)
|
||
|
|
target = next(row for row in buckets if row["lineId"] == 1)
|
||
|
|
target["effectiveWindows"] = [
|
||
|
|
{"startMin": index, "endMin": index + 1, "shiftId": 1}
|
||
|
|
for index in range(50_001)
|
||
|
|
]
|
||
|
|
return buckets
|
||
|
|
|
||
|
|
monkeypatch.setattr(cp_engine, "_line_day_buckets", oversized_buckets)
|
||
|
|
with pytest.raises((ValueError, RuntimeError), match=r"C3.*(?:50,?000|50000)"):
|
||
|
|
optimize_line_assignment(
|
||
|
|
world,
|
||
|
|
entries,
|
||
|
|
params,
|
||
|
|
pipeline_label="C3-MODEL-LIMIT",
|
||
|
|
diagnostic_mode=True,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def test_short_rule_materialization_remains_calendar_compliant() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=60)
|
||
|
|
|
||
|
|
result = CpSatEngine().solve(world, _params(start_date), _counter())
|
||
|
|
|
||
|
|
version = next(row for row in world["scheduleVersions"] if row["id"] == result.versionId)
|
||
|
|
meta = version["solverMeta"]
|
||
|
|
validation = meta["materializedC3Validation"]
|
||
|
|
slot = _active_slots(meta)[0]
|
||
|
|
work_order = world["workOrders"][0]
|
||
|
|
assert result.solveStatus in {"OPTIMAL", "FEASIBLE"}
|
||
|
|
assert meta["placement"] == "cp-calendar-segmented-direct"
|
||
|
|
assert meta["directlyConsumedByMaterializer"] is True
|
||
|
|
assert meta["operationTimingValidation"]["passed"] is True
|
||
|
|
assert validation["checked"] is True
|
||
|
|
assert validation["passed"] is True
|
||
|
|
assert validation["cpTimingApplied"] is True
|
||
|
|
assert validation["exactAlignmentCount"] == 1
|
||
|
|
assert validation["violationCount"] == 0
|
||
|
|
assert work_order["cpTimingSource"] == "operationSlots"
|
||
|
|
assert work_order["logicalOperationKey"] == slot["logicalOperationKey"]
|
||
|
|
assert work_order["processingMinutes"] == slot["processingMinutes"]
|
||
|
|
assert work_order["elapsedSpanMinutes"] == slot["elapsedSpanMinutes"]
|
||
|
|
assert work_order["pauseMinutes"] == slot["pauseMinutes"]
|
||
|
|
assert work_order["segmentCount"] == len(work_order["plannedSegments"]) == 1
|
||
|
|
assert not [
|
||
|
|
row for row in world["conflicts"]
|
||
|
|
if row.get("versionId") == result.versionId and row.get("conflictType") == "CALENDAR"
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
def test_rule_materialization_crossing_break_creates_c3_hard_blocker() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=500, day_shift_only=True)
|
||
|
|
|
||
|
|
result = CpSatEngine().solve(world, _params(start_date), _counter())
|
||
|
|
|
||
|
|
version = next(row for row in world["scheduleVersions"] if row["id"] == result.versionId)
|
||
|
|
validation = version["solverMeta"]["materializedC3Validation"]
|
||
|
|
blockers = hard_blocking_conflicts(world, result.versionId)
|
||
|
|
calendar_blockers = [
|
||
|
|
row for row in blockers if row.get("constraintId") == "C3_calendar"
|
||
|
|
]
|
||
|
|
assert result.solveStatus in {"OPTIMAL", "FEASIBLE"}
|
||
|
|
assert validation["checked"] is True
|
||
|
|
assert validation["passed"] is True
|
||
|
|
assert validation["cpTimingApplied"] is True
|
||
|
|
assert validation["exactAlignmentCount"] == 1
|
||
|
|
assert validation["violationCount"] == 0
|
||
|
|
assert calendar_blockers == []
|
||
|
|
work_order = world["workOrders"][0]
|
||
|
|
segments = work_order["plannedSegments"]
|
||
|
|
assert work_order["segmentCount"] == len(segments) == 2
|
||
|
|
assert work_order["processingMinutes"] == sum(row["durationMin"] for row in segments) == 280
|
||
|
|
assert work_order["elapsedSpanMinutes"] == 340
|
||
|
|
assert work_order["pauseMinutes"] == 60
|
||
|
|
assert segments[0]["endTime"] < segments[1]["startTime"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_direct_materialization_rejects_segment_calendar_window_drift_before_write() -> None:
|
||
|
|
world, start_date = _single_line_world(quantity=500, day_shift_only=True)
|
||
|
|
params = _params(start_date)
|
||
|
|
entries, _, _ = CpSatEngine().collect_and_order(world, params)
|
||
|
|
ordered, meta = optimize_line_assignment(
|
||
|
|
world, entries, params, pipeline_label="C3-PARENT-PREFLIGHT",
|
||
|
|
)
|
||
|
|
meta["operationSlots"][0]["segments"][0]["calendarWindowId"] = "forged-window"
|
||
|
|
before_counts = {
|
||
|
|
key: len(world[key])
|
||
|
|
for key in ("scheduleVersions", "productionOrders", "workOrders", "conflicts")
|
||
|
|
}
|
||
|
|
|
||
|
|
with pytest.raises(SolverProcessError) as caught:
|
||
|
|
cp_engine._prepare_cp_operation_timing(world, entries, ordered, meta)
|
||
|
|
|
||
|
|
assert caught.value.code == "SOLVER_RESPONSE_INVALID"
|
||
|
|
assert before_counts == {
|
||
|
|
key: len(world[key])
|
||
|
|
for key in ("scheduleVersions", "productionOrders", "workOrders", "conflicts")
|
||
|
|
}
|