430 lines
16 KiB
Python
430 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from server.agent_core.async_jobs import JobQueue
|
|
from server.aps_domain.cp_rhs import run_cp_rhs_resolve
|
|
from server.aps_domain.masterdata import master_overview
|
|
from server.engines.base import EngineParams
|
|
from server.engines.cp_engine import CpSatEngine, _line_day_buckets
|
|
from server.engines.solver_process import (
|
|
run_cp_constraint_diagnostic,
|
|
run_cp_rhs_diagnostic,
|
|
)
|
|
from server.engines.solver_worker import _validate_request
|
|
from server.state.seed import build_demo_world
|
|
from server.timeutil import add_minutes, fmt_dt, parse_dt
|
|
from tests.auth_provider import install_test_auth
|
|
from tests.golden.test_cp_rhs import _worker_request
|
|
|
|
|
|
def _tight_c7_world(*, orders: int = 2, quantity: int = 840) -> tuple[dict, str]:
|
|
world = build_demo_world()
|
|
start_date = min(
|
|
row["date"] for row in world["shiftCalendar"]
|
|
if row["lineId"] == 1 and row["isWorking"]
|
|
)
|
|
selected = [
|
|
row for row in world["salesOrders"]
|
|
if row["items"][0]["productId"] == 1
|
|
][:orders]
|
|
for order in selected:
|
|
order["deliveryDate"] = start_date
|
|
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
|
|
]
|
|
world["scheduleParams"]["freezeWindowHours"] = 0
|
|
world["scheduleParams"]["deliveryBufferRatio"] = 1.0
|
|
return world, start_date
|
|
|
|
|
|
def _entries_params(world: dict, start_date: str, **overrides):
|
|
values = {
|
|
"orderIds": [], "engineType": "CP", "strategyTemplate": "COMPREHENSIVE",
|
|
"planningHorizonDays": 14, "startDate": start_date, "timeLimitSeconds": 4,
|
|
"constraints": {"capacity": True},
|
|
}
|
|
values.update(overrides)
|
|
params = EngineParams(**values)
|
|
entries, _, _ = CpSatEngine().collect_and_order(world, params)
|
|
return entries, params
|
|
|
|
|
|
def _counter():
|
|
counters: dict[str, int] = {}
|
|
|
|
def next_id(kind: str) -> int:
|
|
counters[kind] = counters.get(kind, 0) + 1
|
|
return counters[kind]
|
|
|
|
return next_id
|
|
|
|
|
|
def _add_frozen_line_load(world: dict, start_date: str, minutes: int = 480) -> None:
|
|
world["scheduleVersions"].append({
|
|
"id": 900, "versionNo": "FROZEN", "status": "PUBLISHED",
|
|
})
|
|
world["productionOrders"].append({
|
|
"id": 900, "orderNo": "PO-FROZEN", "schedulingVersionId": 900,
|
|
})
|
|
start = parse_dt(start_date + " 08:00")
|
|
world["workOrders"].append({
|
|
"id": 900,
|
|
"orderNo": "WO-FROZEN",
|
|
"productionOrderId": 900,
|
|
"lineId": 1,
|
|
"workstationId": 2,
|
|
"plannedStartTime": fmt_dt(start),
|
|
"plannedEndTime": fmt_dt(add_minutes(start, minutes)),
|
|
"isFrozen": True,
|
|
})
|
|
|
|
|
|
def test_real_c7_baseline_is_active_and_never_exceeds_line_day_rhs():
|
|
world, start_date = _tight_c7_world()
|
|
entries, params = _entries_params(world, start_date)
|
|
_, meta = run_cp_rhs_diagnostic(world, entries, params, pipeline_label="C7-BASELINE")
|
|
assert meta["status"] == "OPTIMAL"
|
|
assert meta["constraintInstanceCounts"]["C7_capacity"] > 0
|
|
assert "C7_capacity" in meta["activeAssumptionConstraints"]
|
|
assert meta["lineDailyCapacity"]["accountingMethod"] == "start-day-full-duration.v1"
|
|
assert meta["lineDailyCapacity"]["isMaterializedShiftModel"] is False
|
|
used = [row for row in meta["lineDailyCapacity"]["resources"] if row["usedMinutes"]]
|
|
assert used
|
|
assert all(row["usedMinutes"] <= row["capacityMinutes"] for row in used)
|
|
|
|
|
|
def test_c7_default_14_day_model_performance_baseline():
|
|
world = build_demo_world()
|
|
start_date = min(
|
|
row["date"] for row in world["shiftCalendar"]
|
|
if row["lineId"] == 1 and row["isWorking"]
|
|
)
|
|
entries, params = _entries_params(world, start_date, timeLimitSeconds=4)
|
|
_, meta = run_cp_rhs_diagnostic(world, entries, params, pipeline_label="C7-PERF")
|
|
assert meta["status"] in {"OPTIMAL", "FEASIBLE"}
|
|
assert meta["wallTimeSec"] < 10
|
|
assert 1 <= len(meta["lineDailyCapacity"]["resources"]) <= 90
|
|
|
|
|
|
def test_c7_line_day_plus_60_has_exact_objective_improvement_and_instance_cost():
|
|
world, start_date = _tight_c7_world()
|
|
before = copy.deepcopy(world)
|
|
instance_id = f"line-day:1:{start_date}"
|
|
report = run_cp_rhs_resolve(
|
|
world,
|
|
start_date=start_date,
|
|
parameter_ids=["C7_line_day_capacity_minutes"],
|
|
instance_increments={instance_id: 60},
|
|
instance_cost_rates={instance_id: 2.0},
|
|
)
|
|
assert world == before
|
|
assert report["status"] == "completed"
|
|
row = report["rows"][0]
|
|
assert row["status"] == "available"
|
|
assert row["rhsInstanceId"] == instance_id
|
|
assert row["lineCode"] == "L001"
|
|
assert row["baselineRhs"] == 869
|
|
assert row["perturbedRhs"] == 929
|
|
assert row["objectiveImprovement"] > 0
|
|
assert row["costScope"] == "rhs-instance"
|
|
assert row["estimatedCost"] == 120.0
|
|
|
|
|
|
def test_c7_unconfigured_instance_cost_is_all_null():
|
|
world, start_date = _tight_c7_world()
|
|
instance_id = f"line-day:1:{start_date}"
|
|
row = run_cp_rhs_resolve(
|
|
world,
|
|
start_date=start_date,
|
|
parameter_ids=["C7_line_day_capacity_minutes"],
|
|
instance_increments={instance_id: 60},
|
|
)["rows"][0]
|
|
assert row["costStatus"] == "not_configured"
|
|
for field in (
|
|
"costRate", "costRateUnit", "costCurrency", "estimatedCost", "costScope", "costSource",
|
|
):
|
|
assert row[field] is None
|
|
|
|
|
|
def test_frozen_existing_load_consumes_c7_and_instance_increment_can_relieve_it():
|
|
world, start_date = _tight_c7_world(orders=1, quantity=840)
|
|
_add_frozen_line_load(world, start_date, 480)
|
|
entries, params = _entries_params(world, start_date)
|
|
_, baseline = run_cp_rhs_diagnostic(
|
|
world, entries, params, pipeline_label="C7-FROZEN-BASELINE",
|
|
)
|
|
day0 = next(
|
|
row for row in baseline["lineDailyCapacity"]["resources"]
|
|
if row["lineId"] == 1 and row["bucketDate"] == start_date
|
|
)
|
|
assert day0["fixedFrozenLoadMinutes"] == 480
|
|
assert day0["usedMinutes"] == 480
|
|
|
|
instance_id = f"line-day:1:{start_date}"
|
|
report = run_cp_rhs_resolve(
|
|
world,
|
|
start_date=start_date,
|
|
parameter_ids=["C7_line_day_capacity_minutes"],
|
|
instance_increments={instance_id: 61},
|
|
)
|
|
row = report["rows"][0]
|
|
assert row["baselineRhs"] == 869
|
|
assert row["perturbedRhs"] == 930
|
|
assert row["objectiveImprovement"] > 0
|
|
|
|
|
|
def test_c7_whole_removal_keeps_protocol_separate_and_improves_objective():
|
|
world, start_date = _tight_c7_world()
|
|
entries, params = _entries_params(world, start_date)
|
|
_, baseline = run_cp_constraint_diagnostic(
|
|
world, entries, params, pipeline_label="C7-BASELINE",
|
|
)
|
|
_, relaxed = run_cp_constraint_diagnostic(
|
|
world, entries, params, pipeline_label="C7-REMOVAL",
|
|
relaxed_constraint_id="C7_capacity",
|
|
)
|
|
assert baseline["status"] == relaxed["status"] == "OPTIMAL"
|
|
assert relaxed["objective"] < baseline["objective"]
|
|
assert relaxed["relaxedConstraintIds"] == ["C7_capacity"]
|
|
|
|
|
|
def test_cp_materialization_exact_c7_drift_is_explicit_and_keeps_capacity_blockers():
|
|
world, start_date = _tight_c7_world()
|
|
_, params = _entries_params(world, start_date)
|
|
result = CpSatEngine().solve(world, params, _counter())
|
|
version = next(row for row in world["scheduleVersions"] if row["id"] == result.versionId)
|
|
validation = version["solverMeta"]["materializedC7Validation"]
|
|
assert result.solveStatus == "OPTIMAL"
|
|
assert validation["checked"] is True
|
|
assert validation["passed"] is True
|
|
assert validation["accountingMethod"] == "start-day-full-duration.v1"
|
|
assert validation["violations"] == []
|
|
blockers = [
|
|
row for row in world["conflicts"]
|
|
if row["versionId"] == result.versionId and row["conflictType"] == "CAPACITY"
|
|
]
|
|
assert blockers == []
|
|
assert validation["conflictIds"] == []
|
|
direct_work_orders = [
|
|
row for row in world["workOrders"] if row.get("cpTimingSource") == "operationSlots"
|
|
]
|
|
processing_by_start_day: dict[str, int] = {}
|
|
for work_order in direct_work_orders:
|
|
start_day = work_order["plannedStartTime"][:10]
|
|
processing_by_start_day[start_day] = (
|
|
processing_by_start_day.get(start_day, 0) + work_order["processingMinutes"]
|
|
)
|
|
assert work_order["elapsedSpanMinutes"] == (
|
|
work_order["processingMinutes"] + work_order["pauseMinutes"]
|
|
)
|
|
resources = version["solverMeta"]["lineDailyCapacity"]["resources"]
|
|
for bucket_date, processing_minutes in processing_by_start_day.items():
|
|
bucket = next(row for row in resources if row["bucketDate"] == bucket_date)
|
|
assert bucket["usedMinutes"] == processing_minutes
|
|
|
|
|
|
def test_long_operation_crossing_midnight_is_charged_wholly_to_start_day():
|
|
world, start_date = _tight_c7_world(orders=1, quantity=1140)
|
|
entries, params = _entries_params(world, start_date, freezeWindowHours=15.0)
|
|
_, meta = run_cp_rhs_diagnostic(world, entries, params, pipeline_label="C7-LONG")
|
|
slot = meta["operationSlots"][0]
|
|
assert slot["startMin"] >= 15 * 60
|
|
assert slot["endMin"] > 16 * 60
|
|
resources = meta["lineDailyCapacity"]["resources"]
|
|
day0 = next(row for row in resources if row["lineId"] == 1 and row["bucketDate"] == start_date)
|
|
next_day = next(row for row in resources if row["lineId"] == 1 and row["bucketDate"] > start_date)
|
|
assert day0["usedMinutes"] == slot["durationMin"] == 600
|
|
assert next_day["usedMinutes"] == 0
|
|
|
|
|
|
def test_c7_calendar_capacity_does_not_multiply_line_efficiency_again():
|
|
world, start_date = _tight_c7_world(orders=1, quantity=400)
|
|
anchor = parse_dt(start_date + " 08:00")
|
|
baseline = _line_day_buckets(world, {1}, anchor, 2 * 24 * 60)[0]
|
|
next(line for line in world["lines"] if line["id"] == 1)["efficiencyFactor"] = 0.5
|
|
changed = _line_day_buckets(world, {1}, anchor, 2 * 24 * 60)[0]
|
|
assert baseline["baseCapacityMinutes"] == changed["baseCapacityMinutes"] == 869
|
|
|
|
|
|
def test_c7_missing_calendar_coverage_fails_closed():
|
|
world, start_date = _tight_c7_world(orders=1)
|
|
world["shiftCalendar"] = [
|
|
row for row in world["shiftCalendar"]
|
|
if not (row["lineId"] == 1 and row["date"] == start_date)
|
|
]
|
|
with pytest.raises(ValueError, match="覆盖不完整"):
|
|
_line_day_buckets(world, {1}, parse_dt(start_date + " 08:00"), 2 * 24 * 60)
|
|
|
|
|
|
def test_c7_explicit_nonworking_day_has_zero_capacity():
|
|
world, start_date = _tight_c7_world(orders=1)
|
|
for row in world["shiftCalendar"]:
|
|
if row["lineId"] == 1 and row["date"] == start_date:
|
|
row["isWorking"] = False
|
|
bucket = _line_day_buckets(
|
|
world, {1}, parse_dt(start_date + " 08:00"), 2 * 24 * 60,
|
|
)[0]
|
|
assert bucket["baseCapacityMinutes"] == 0
|
|
assert bucket["shiftIds"] == []
|
|
|
|
|
|
def test_master_projection_exposes_only_working_dates_for_c7_picker():
|
|
world, start_date = _tight_c7_world(orders=1)
|
|
row = next(item for item in master_overview(world)["calendar"] if item["lineId"] == 1)
|
|
assert start_date in row["workingDates"]
|
|
nonworking = {
|
|
item["date"] for item in world["shiftCalendar"]
|
|
if item["lineId"] == 1 and not item["isWorking"]
|
|
}
|
|
assert nonworking.isdisjoint(row["workingDates"])
|
|
|
|
|
|
def test_c7_cross_midnight_and_overlapping_shifts_fail_closed():
|
|
world, start_date = _tight_c7_world(orders=1)
|
|
first_row = next(
|
|
row for row in world["shiftCalendar"]
|
|
if row["lineId"] == 1 and row["date"] == start_date and row["isWorking"]
|
|
)
|
|
shift = next(row for row in world["shifts"] if row["id"] == first_row["shiftId"])
|
|
shift["endTime"] = shift["startTime"]
|
|
with pytest.raises(ValueError, match="跨午夜"):
|
|
_line_day_buckets(world, {1}, parse_dt(start_date + " 08:00"), 2 * 24 * 60)
|
|
|
|
|
|
def test_worker_accepts_strict_c7_line_day_rhs_schema():
|
|
perturbation = {
|
|
"parameterId": "C7_line_day_capacity_minutes",
|
|
"lineId": 1,
|
|
"bucketDate": "2026-08-18",
|
|
"increment": 60,
|
|
}
|
|
validated = _validate_request(
|
|
_worker_request("diagnose_rhs_perturbation", perturbation)
|
|
)
|
|
assert validated[8] == perturbation
|
|
assert validated[-1] is True
|
|
|
|
|
|
def test_worker_rejects_invalid_c7_line_day_rhs_schema():
|
|
with pytest.raises(ValueError, match="C7"):
|
|
_validate_request(_worker_request("diagnose_rhs_perturbation", {
|
|
"parameterId": "C7_line_day_capacity_minutes",
|
|
"lineId": 1,
|
|
"bucketDate": "20260818",
|
|
"increment": 60,
|
|
}))
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("instance_increments", "instance_cost_rates"),
|
|
[
|
|
({}, {}),
|
|
({"bad": 60}, {}),
|
|
({"line-day:1:2026-08-18": True}, {}),
|
|
({"line-day:1:2026-08-18": 1441}, {}),
|
|
({"line-day:1:2026-08-18": 60}, {"other": 1.0}),
|
|
({"line-day:1:2026-08-18": 60}, {"line-day:1:2026-08-18": float("nan")}),
|
|
],
|
|
)
|
|
def test_c7_domain_rejects_invalid_instance_contract(instance_increments, instance_cost_rates):
|
|
world, start_date = _tight_c7_world()
|
|
with pytest.raises((TypeError, ValueError)):
|
|
run_cp_rhs_resolve(
|
|
world,
|
|
start_date=start_date,
|
|
parameter_ids=["C7_line_day_capacity_minutes"],
|
|
instance_increments=instance_increments,
|
|
instance_cost_rates=instance_cost_rates,
|
|
)
|
|
|
|
|
|
class _Store:
|
|
def __init__(self, world: dict) -> None:
|
|
self.data = world
|
|
|
|
|
|
class _Projects:
|
|
def active_world_key(self) -> str:
|
|
return "project-c7"
|
|
|
|
def require_active_write(self) -> None:
|
|
return None
|
|
|
|
|
|
@pytest.fixture
|
|
def c7_client(monkeypatch):
|
|
import server.gateway.app as gateway
|
|
from server.agent_core import async_jobs
|
|
from server.state import projects
|
|
|
|
world, start_date = _tight_c7_world()
|
|
install_test_auth(monkeypatch, "tenant-c7")
|
|
monkeypatch.setattr(gateway, "get_store", lambda: _Store(world))
|
|
monkeypatch.setattr(projects, "get_project_store", lambda: _Projects())
|
|
queue = JobQueue()
|
|
monkeypatch.setattr(async_jobs, "_queue", queue)
|
|
client = TestClient(gateway.create_app())
|
|
assert client.post(
|
|
"/api/auth/login", json={"username": "planner", "password": "test"},
|
|
).status_code == 200
|
|
return client, queue, start_date
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutator",
|
|
[
|
|
lambda start: {},
|
|
lambda start: {"instanceIncrements": {"bad": 60}},
|
|
lambda start: {"instanceIncrements": {f"line-day:1:{start}": 0}},
|
|
lambda start: {
|
|
"instanceIncrements": {f"line-day:1:{start}": 60},
|
|
"instanceCostRates": {"other": 1},
|
|
},
|
|
],
|
|
)
|
|
def test_gateway_rejects_invalid_c7_before_queue(c7_client, mutator):
|
|
client, queue, start_date = c7_client
|
|
params = {
|
|
"startDate": start_date,
|
|
"parameterIds": ["C7_line_day_capacity_minutes"],
|
|
**mutator(start_date),
|
|
}
|
|
response = client.post("/api/jobs", json={"kind": "cp-rhs.recompute", "params": params})
|
|
assert response.status_code == 422, response.text
|
|
assert queue.stats()["total"] == 0
|
|
|
|
|
|
def test_gateway_real_c7_submit_and_poll(c7_client):
|
|
client, queue, start_date = c7_client
|
|
instance_id = f"line-day:1:{start_date}"
|
|
response = client.post("/api/jobs", json={
|
|
"kind": "cp-rhs.recompute",
|
|
"params": {
|
|
"startDate": start_date,
|
|
"parameterIds": ["C7_line_day_capacity_minutes"],
|
|
"instanceIncrements": {instance_id: 60},
|
|
"instanceCostRates": {instance_id: 2},
|
|
},
|
|
})
|
|
assert response.status_code == 200, response.text
|
|
record = queue.wait(response.json()["jobId"], timeout=20)
|
|
assert record["status"] == "done"
|
|
row = record["result"]["rows"][0]
|
|
assert row["rhsInstanceId"] == instance_id
|
|
assert row["estimatedCost"] == 120.0
|