300 lines
12 KiB
Python
300 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from server.aps_domain.attribution import build_attribution_report
|
|
from server.aps_domain.shadow_price import analyze_diagnostic_lp
|
|
from server.engines import get_engine
|
|
from server.engines.base import EngineParams
|
|
from server.state.seed import build_demo_world
|
|
from tests.auth_provider import install_test_auth
|
|
|
|
|
|
def _row(
|
|
conflict_id: int,
|
|
constraint_id: str,
|
|
kind: str,
|
|
value: float,
|
|
*,
|
|
relief: list[str] | None = None,
|
|
resource: str = "L1",
|
|
when: str = "2026-08-18 08:00",
|
|
) -> dict:
|
|
return {
|
|
"conflictId": conflict_id,
|
|
"constraintId": constraint_id,
|
|
"constraintName": constraint_id,
|
|
"reliefConstraintIds": relief or [constraint_id],
|
|
"resourceName": resource,
|
|
"conflictTimeStart": when,
|
|
"orderNos": [f"SO-{conflict_id}"],
|
|
"degree": {"kind": kind, "value": value},
|
|
}
|
|
|
|
|
|
def test_local_finite_difference_and_one_minute_benefit_are_distinct_at_breakpoint():
|
|
report = analyze_diagnostic_lp([
|
|
_row(1, "C7_capacity", "overload_minutes", 0.5, resource="L1"),
|
|
_row(2, "C7_capacity", "overload_minutes", 10.0, resource="L2"),
|
|
])
|
|
row = report["constraints"]["C7_capacity"]
|
|
assert row["status"] == "available"
|
|
assert row["localStepMinutes"] == 0.25
|
|
assert row["localMarginalBenefit"] == 2.0
|
|
assert row["capDualEvidence"] == 2.0
|
|
assert row["oneMinuteBenefit"] == 1.5
|
|
assert row["dualConsistencyPassed"] is True
|
|
|
|
|
|
def test_non_minute_constraint_and_count_evidence_are_unsupported():
|
|
report = analyze_diagnostic_lp([
|
|
_row(1, "C5_capability", "overload_minutes", 30.0),
|
|
_row(2, "C11_freeze", "count", 1.0),
|
|
])
|
|
for constraint_id in ("C5_capability", "C11_freeze"):
|
|
row = report["constraints"][constraint_id]
|
|
assert row["status"] == "unsupported"
|
|
assert row["localMarginalBenefit"] is None
|
|
assert row["unsupportedCount"] == 1
|
|
|
|
|
|
def test_shared_relief_is_marked_non_additive_and_not_reinterpreted_across_units():
|
|
report = analyze_diagnostic_lp([
|
|
_row(
|
|
7,
|
|
"C7_capacity",
|
|
"overload_minutes",
|
|
12.0,
|
|
relief=["C7_capacity", "C8_due_date"],
|
|
),
|
|
])
|
|
capacity = report["constraints"]["C7_capacity"]
|
|
due_date = report["constraints"]["C8_due_date"]
|
|
assert report["counterfactualMode"] == "one-constraint-at-a-time"
|
|
assert report["nonAdditiveAcrossConstraints"] is True
|
|
assert capacity["status"] == "available"
|
|
assert due_date["status"] == "unsupported"
|
|
assert capacity["sharedReliefEvidenceRefs"] == ["conflict#7"]
|
|
assert due_date["sharedReliefEvidenceRefs"] == ["conflict#7"]
|
|
|
|
|
|
def test_duplicate_evidence_and_same_physical_rhs_are_not_double_counted():
|
|
first = _row(1, "C7_capacity", "overload_minutes", 10.0)
|
|
duplicate_id = copy.deepcopy(first)
|
|
duplicate_rhs = _row(2, "C7_capacity", "overload_minutes", 12.0)
|
|
report = analyze_diagnostic_lp([first, duplicate_id, duplicate_rhs])
|
|
row = report["constraints"]["C7_capacity"]
|
|
assert row["evidenceCount"] == 2
|
|
assert row["quantifiedCount"] == 1
|
|
assert row["deduplicatedCount"] == 1
|
|
assert row["deduplicatedEvidenceRefs"] == ["conflict#1"]
|
|
assert row["duplicateEvidenceIdCount"] == 1
|
|
assert row["localMarginalBenefit"] == 1.0
|
|
assert row["baselineObjectiveMinutes"] == 12.0
|
|
|
|
|
|
def test_input_permutation_is_deterministic():
|
|
rows = [
|
|
_row(3, "C8_due_date", "hours_late", 1.5),
|
|
_row(1, "C7_capacity", "overload_minutes", 10.0, resource="L2"),
|
|
_row(2, "C7_capacity", "overload_minutes", 20.0, resource="L1"),
|
|
]
|
|
assert analyze_diagnostic_lp(rows) == analyze_diagnostic_lp(list(reversed(rows)))
|
|
|
|
|
|
def test_conflicting_duplicate_evidence_id_fails_closed_independent_of_order():
|
|
first = _row(1, "C7_capacity", "overload_minutes", 1.0)
|
|
conflicting = {**first, "degree": {"kind": "overload_minutes", "value": 100.0}}
|
|
forward = analyze_diagnostic_lp([first, conflicting])
|
|
reverse = analyze_diagnostic_lp([conflicting, first])
|
|
assert forward == reverse
|
|
row = forward["constraints"]["C7_capacity"]
|
|
assert row["status"] == "unsupported"
|
|
assert row["conflictingDuplicateEvidenceRefs"] == ["conflict#1"]
|
|
assert "失败关闭" in row["reason"]
|
|
|
|
|
|
def test_conflicting_duplicate_id_is_detected_globally_before_constraint_split():
|
|
capacity = _row(1, "C7_capacity", "overload_minutes", 10.0)
|
|
due_date = _row(1, "C8_due_date", "hours_late", 2.0)
|
|
report = analyze_diagnostic_lp([capacity, due_date])
|
|
assert report["inputConsistencyPassed"] is False
|
|
assert report["dualConsistencyPassed"] is False
|
|
assert report["conflictingDuplicateEvidenceRefs"] == ["conflict#1"]
|
|
for constraint_id in ("C7_capacity", "C8_due_date"):
|
|
row = report["constraints"][constraint_id]
|
|
assert row["status"] == "unsupported"
|
|
assert row["conflictingDuplicateEvidenceRefs"] == ["conflict#1"]
|
|
|
|
|
|
def test_conflicting_duplicate_id_fails_closed_even_with_other_valid_evidence():
|
|
first = _row(1, "C7_capacity", "overload_minutes", 1.0, resource="L1")
|
|
conflicting = {**first, "degree": {"kind": "overload_minutes", "value": 100.0}}
|
|
valid = _row(2, "C7_capacity", "overload_minutes", 20.0, resource="L2")
|
|
row = analyze_diagnostic_lp([first, conflicting, valid])["constraints"]["C7_capacity"]
|
|
assert row["status"] == "unsupported"
|
|
assert row["localMarginalBenefit"] is None
|
|
assert row["conflictingDuplicateEvidenceRefs"] == ["conflict#1"]
|
|
assert "失败关闭" in row["reason"]
|
|
|
|
|
|
def test_hours_are_converted_to_minutes_with_log_precision_metadata():
|
|
report = analyze_diagnostic_lp([
|
|
_row(1, "C8_due_date", "hours_late", 1.5),
|
|
])
|
|
row = report["constraints"]["C8_due_date"]
|
|
assert row["baselineObjectiveMinutes"] == 90.0
|
|
assert row["inputSources"] == ["solve-log-description"]
|
|
assert row["inputPrecisionMinutes"] == 6.0
|
|
|
|
|
|
def test_non_finite_zero_and_negative_values_do_not_emit_a_rate():
|
|
rows = [
|
|
_row(1, "C7_capacity", "overload_minutes", float("nan"), resource="L1"),
|
|
_row(2, "C7_capacity", "overload_minutes", float("inf"), resource="L2"),
|
|
_row(3, "C7_capacity", "overload_minutes", 0.0, resource="L3"),
|
|
_row(4, "C7_capacity", "overload_minutes", -1.0, resource="L4"),
|
|
]
|
|
result = analyze_diagnostic_lp(rows)["constraints"]["C7_capacity"]
|
|
assert result["status"] == "unsupported"
|
|
assert result["quantifiedCount"] == 0
|
|
assert result["unsupportedCount"] == 4
|
|
|
|
|
|
def test_constraint_and_degree_kind_must_match_minute_rhs_semantics():
|
|
rows = [
|
|
_row(1, "C4_maintenance", "hours_late", 1.0),
|
|
_row(2, "C7_capacity", "overlap_minutes", 10.0),
|
|
_row(3, "C8_due_date", "overload_minutes", 10.0),
|
|
]
|
|
report = analyze_diagnostic_lp(rows)
|
|
assert all(row["status"] == "unsupported" for row in report["constraints"].values())
|
|
|
|
|
|
def test_missing_physical_rhs_identity_is_unsupported_instead_of_merged():
|
|
capacity = _row(1, "C7_capacity", "overload_minutes", 10.0, resource="")
|
|
capacity["conflictTimeStart"] = ""
|
|
maintenance = _row(2, "C4_maintenance", "overlap_minutes", 10.0, resource="")
|
|
maintenance["conflictTimeStart"] = ""
|
|
due_date = _row(3, "C8_due_date", "hours_late", 1.0)
|
|
due_date["orderNos"] = []
|
|
report = analyze_diagnostic_lp([capacity, maintenance, due_date])
|
|
assert all(row["status"] == "unsupported" for row in report["constraints"].values())
|
|
|
|
|
|
def test_invalid_capacity_date_and_blank_due_order_are_unsupported():
|
|
capacity = _row(1, "C7_capacity", "overload_minutes", 10.0)
|
|
capacity["conflictTimeStart"] = "xxxxxxxxxx"
|
|
due_date = _row(2, "C8_due_date", "hours_late", 1.0)
|
|
due_date["orderNos"] = [" "]
|
|
report = analyze_diagnostic_lp([capacity, due_date])
|
|
assert report["constraints"]["C7_capacity"]["status"] == "unsupported"
|
|
assert report["constraints"]["C8_due_date"]["status"] == "unsupported"
|
|
|
|
|
|
def test_solver_failure_does_not_emit_a_dual(monkeypatch):
|
|
from server.aps_domain import shadow_price
|
|
|
|
monkeypatch.setattr(
|
|
shadow_price,
|
|
"_solve_relaxation",
|
|
lambda _degrees, _cap: {
|
|
"status": "SOLVER_STATUS_6", "objective": None, "capDual": None,
|
|
},
|
|
)
|
|
row = analyze_diagnostic_lp([
|
|
_row(1, "C7_capacity", "overload_minutes", 10.0),
|
|
])["constraints"]["C7_capacity"]
|
|
assert row["status"] == "solver_error"
|
|
assert row["localMarginalBenefit"] is None
|
|
assert row["capDualEvidence"] is None
|
|
|
|
|
|
def test_solver_error_is_separate_from_unsupported_and_fails_global_consistency(monkeypatch):
|
|
from server.aps_domain import shadow_price
|
|
|
|
original = shadow_price._solve_relaxation
|
|
|
|
def selective_failure(degrees, cap):
|
|
if degrees == [90.0]:
|
|
return {"status": "SOLVER_STATUS_6", "objective": None, "capDual": None}
|
|
return original(degrees, cap)
|
|
|
|
monkeypatch.setattr(shadow_price, "_solve_relaxation", selective_failure)
|
|
report = analyze_diagnostic_lp([
|
|
_row(1, "C7_capacity", "overload_minutes", 10.0),
|
|
_row(2, "C8_due_date", "hours_late", 1.5),
|
|
])
|
|
assert report["availableConstraints"] == 1
|
|
assert report["unsupportedConstraints"] == 0
|
|
assert report["solverErrorConstraints"] == 1
|
|
assert report["dualConsistencyPassed"] is False
|
|
|
|
|
|
def _scheduled_world() -> tuple[dict, int]:
|
|
world = build_demo_world()
|
|
result = get_engine("RULE").solve(
|
|
world,
|
|
EngineParams(
|
|
orderIds=[],
|
|
engineType="RULE",
|
|
strategyTemplate="COMPREHENSIVE",
|
|
planningHorizonDays=14,
|
|
startDate="2026-08-03",
|
|
),
|
|
lambda kind: len(world.get(f"{kind}s", [])) + 1,
|
|
)
|
|
return world, result.versionId
|
|
|
|
|
|
def test_attribution_report_adds_diagnostic_lp_without_replacing_proxy_contract():
|
|
world, version_id = _scheduled_world()
|
|
before = copy.deepcopy(world)
|
|
report = build_attribution_report(world, version_id=version_id)
|
|
assert world == before
|
|
analysis = report["diagnosticLpAnalysis"]
|
|
assert analysis["modelScope"] == "diagnostic-conflict-relaxation"
|
|
assert analysis["uniformAcrossEvidenceRows"] is True
|
|
assert analysis["exactForLpModel"] is True
|
|
assert analysis["exactForCpSat"] is False
|
|
assert analysis["dualConsistencyPassed"] is True
|
|
assert report["masterControl"]
|
|
for row in report["masterControl"]:
|
|
assert row["isShadowPriceProxy"] is True
|
|
assert "diagnosticLp" in row
|
|
|
|
|
|
class _Store:
|
|
def __init__(self) -> None:
|
|
self.data, self.version_id = _scheduled_world()
|
|
|
|
|
|
def test_attribution_api_exposes_version_bound_diagnostic_lp(monkeypatch):
|
|
import server.gateway.app as gateway
|
|
|
|
install_test_auth(monkeypatch, "tenant-shadow-price")
|
|
store = _Store()
|
|
monkeypatch.setattr(gateway, "get_store", lambda: store)
|
|
client = TestClient(gateway.create_app())
|
|
assert client.post(
|
|
"/api/auth/login", json={"username": "planner", "password": "test"},
|
|
).status_code == 200
|
|
response = client.get(f"/api/attribution?track=fixed&versionId={store.version_id}")
|
|
assert response.status_code == 200, response.text
|
|
body = response.json()
|
|
assert body["schemaVersion"] == "schedule-attribution.v1"
|
|
assert body["versionId"] == store.version_id
|
|
assert body["diagnosticLpAnalysis"]["exactForCpSat"] is False
|
|
assert all("diagnosticLp" in row for row in body["masterControl"])
|
|
|
|
|
|
def test_native_sidecar_probe_executes_glop_dual(capsys):
|
|
from server.sidecar import PROBE_OK_PREFIX, probe_native_dependencies
|
|
|
|
probe_native_dependencies()
|
|
output = capsys.readouterr().out
|
|
assert PROBE_OK_PREFIX in output
|
|
assert "glop=GLOP" in output
|