196 lines
6.9 KiB
Python
196 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import sys
|
|
from collections.abc import Callable
|
|
from types import ModuleType
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from server.engines.base import EngineParams
|
|
from server.engines.cp_engine import CpSatEngine, HybridEngine, optimize_line_assignment
|
|
from server.engines.solver_process import _validate_operation_slots
|
|
from server.state.seed import seed_world
|
|
from server.timeutil import add_minutes, fmt_date, today0
|
|
|
|
|
|
def _next_id_factory() -> Callable[[str], int]:
|
|
counters: dict[str, int] = {}
|
|
|
|
def next_id(kind: str) -> int:
|
|
counters[kind] = counters.get(kind, 0) + 1
|
|
return counters[kind]
|
|
|
|
return next_id
|
|
|
|
|
|
def _params(engine_type: str) -> EngineParams:
|
|
return EngineParams(
|
|
orderIds=[],
|
|
engineType=engine_type,
|
|
strategyTemplate="COMPREHENSIVE",
|
|
planningHorizonDays=14,
|
|
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
|
timeLimitSeconds=8.0,
|
|
)
|
|
|
|
|
|
def _install_solver_module(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
runner: Callable[..., tuple[list[dict[str, Any]], dict[str, Any]]],
|
|
) -> type[RuntimeError]:
|
|
module = ModuleType("server.engines.solver_process")
|
|
|
|
class FakeSolverProcessError(RuntimeError):
|
|
def __init__(self, code: str, message: str) -> None:
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
module.SolverProcessError = FakeSolverProcessError
|
|
module._validate_operation_slots = _validate_operation_slots
|
|
module.run_cp_assignment = runner
|
|
monkeypatch.setitem(sys.modules, "server.engines.solver_process", module)
|
|
return FakeSolverProcessError
|
|
|
|
|
|
def test_cp_engine_delegates_only_assignment_and_preserves_solver_metadata(monkeypatch: pytest.MonkeyPatch):
|
|
world = seed_world()
|
|
captured: dict[str, Any] = {}
|
|
|
|
def runner(**kwargs):
|
|
captured.update(kwargs)
|
|
return optimize_line_assignment(
|
|
kwargs["world"],
|
|
kwargs["entries"],
|
|
kwargs["params"],
|
|
warm_start=kwargs.get("warm_start"),
|
|
pipeline_label=kwargs["pipeline_label"],
|
|
)
|
|
|
|
_install_solver_module(monkeypatch, runner)
|
|
result = CpSatEngine().solve(world, _params("CP"), _next_id_factory())
|
|
|
|
assert captured["world"] is world
|
|
assert captured["params"].engineType == "CP"
|
|
assert captured["pipeline_label"] == "CP-SAT→shift-slot"
|
|
assert "warm_start" not in captured
|
|
assert result.engineType == "CP"
|
|
assert result.poCount == 7
|
|
assert result.woCount == 30
|
|
meta = world["scheduleVersions"][-1]["solverMeta"]
|
|
assert meta["backend"] == "OR-Tools CP-SAT"
|
|
assert meta["pipeline"] == "CP-SAT→shift-slot"
|
|
assert "gap" in meta
|
|
assert "cumulative" in meta
|
|
assert meta.get("operationSlots")
|
|
|
|
|
|
def test_hybrid_passes_rule_warm_start_and_materializes_in_parent(monkeypatch: pytest.MonkeyPatch):
|
|
world = seed_world()
|
|
captured: dict[str, Any] = {}
|
|
|
|
def runner(**kwargs):
|
|
captured.update(kwargs)
|
|
return optimize_line_assignment(
|
|
kwargs["world"],
|
|
kwargs["entries"],
|
|
kwargs["params"],
|
|
warm_start=kwargs["warm_start"],
|
|
pipeline_label=kwargs["pipeline_label"],
|
|
)
|
|
|
|
_install_solver_module(monkeypatch, runner)
|
|
result = HybridEngine().solve(world, _params("HYBRID"), _next_id_factory())
|
|
|
|
assert captured["world"] is world
|
|
assert captured["pipeline_label"] == "RULE→CP-SAT→shift-slot"
|
|
assert len(captured["warm_start"]) == len(captured["entries"])
|
|
assert result.engineType == "HYBRID"
|
|
assert result.poCount == 7
|
|
assert result.woCount == 30
|
|
meta = world["scheduleVersions"][-1]["solverMeta"]
|
|
assert meta["warmStart"] == "RULE"
|
|
assert meta["pipeline"] == "RULE→CP-SAT→shift-slot"
|
|
assert meta["status"] in {"OPTIMAL", "FEASIBLE", "UNKNOWN", "INFEASIBLE"}
|
|
|
|
|
|
@pytest.mark.parametrize("engine_cls,engine_type", [(CpSatEngine, "CP"), (HybridEngine, "HYBRID")])
|
|
@pytest.mark.parametrize(
|
|
"error_code",
|
|
["SOLVER_NATIVE_FATAL", "SOLVER_PROCESS_TIMEOUT", "SOLVER_RESPONSE_INVALID"],
|
|
)
|
|
def test_solver_process_error_is_fail_closed_and_world_atomic(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
engine_cls: type[CpSatEngine],
|
|
engine_type: str,
|
|
error_code: str,
|
|
):
|
|
world = seed_world()
|
|
before = copy.deepcopy(world)
|
|
|
|
def runner(**_kwargs):
|
|
raise solver_error(error_code, f"injected {error_code}")
|
|
|
|
solver_error = _install_solver_module(monkeypatch, runner)
|
|
result = engine_cls().solve(world, _params(engine_type), _next_id_factory())
|
|
|
|
assert result.engineType == engine_type
|
|
assert result.solveStatus == "UNAVAILABLE"
|
|
assert result.poCount == 0
|
|
assert result.woCount == 0
|
|
assert result.status == "DRAFT"
|
|
assert f"solver-error:{error_code}" in result.evidenceRefs
|
|
|
|
for key, value in before.items():
|
|
if key not in {"scheduleVersions", "conflicts"}:
|
|
assert world[key] == value, f"solver failure mutated world[{key!r}]"
|
|
|
|
assert len(world["scheduleVersions"]) == len(before["scheduleVersions"]) + 1
|
|
blocked = world["scheduleVersions"][-1]
|
|
assert blocked["status"] == "DRAFT"
|
|
assert blocked["publishReady"] is False
|
|
assert blocked["dispatchReady"] is False
|
|
assert blocked["poCount"] == 0
|
|
assert blocked["woCount"] == 0
|
|
assert blocked["solverMeta"]["status"] == "UNAVAILABLE"
|
|
assert blocked["solverMeta"]["errorCode"] == error_code
|
|
assert "DEGRADED_RULE" not in str(blocked)
|
|
|
|
assert len(world["conflicts"]) == len(before["conflicts"]) + 1
|
|
conflict = world["conflicts"][-1]
|
|
assert conflict["versionId"] == blocked["id"]
|
|
assert conflict["conflictType"] == "ENGINE_UNAVAILABLE"
|
|
assert error_code in conflict["description"]
|
|
|
|
|
|
@pytest.mark.parametrize("engine_cls,engine_type", [(CpSatEngine, "CP"), (HybridEngine, "HYBRID")])
|
|
def test_runtime_unsafe_failure_preserves_only_sanitized_reasons(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
engine_cls: type[CpSatEngine],
|
|
engine_type: str,
|
|
) -> None:
|
|
world = seed_world()
|
|
|
|
def runner(**_kwargs):
|
|
error = solver_error("SOLVER_RUNTIME_UNSAFE", "unsafe runtime")
|
|
error.details = {
|
|
"runtimeIdentity": {
|
|
"reasons": ["ortools:user-site", "python:not-isolated"],
|
|
"executable": "C:/secret/python.exe",
|
|
"packages": {"ortools": {"origin": "C:/secret/site-packages"}},
|
|
}
|
|
}
|
|
raise error
|
|
|
|
solver_error = _install_solver_module(monkeypatch, runner)
|
|
result = engine_cls().solve(world, _params(engine_type), _next_id_factory())
|
|
|
|
assert result.solveStatus == "UNAVAILABLE"
|
|
meta = world["scheduleVersions"][-1]["solverMeta"]
|
|
assert meta["runtimeSafetyReasons"] == ["ortools:user-site", "python:not-isolated"]
|
|
assert meta["runtimeExecutableMatchesParent"] is False
|
|
assert meta["runtimePrefixMatchesParent"] is False
|
|
assert meta["runtimePackageAvailability"] == {"ortools": True}
|
|
assert "C:/secret" not in str(meta)
|