144 lines
6.3 KiB
Python
144 lines
6.3 KiB
Python
# ============================================================
|
|
# PL-01 时间分桶计划黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from server.aps_domain.planning import build_bucket_skeleton, build_plan_buckets, daily_capacity
|
|
from server.state.seed import seed_world
|
|
from server.timeutil import add_minutes, fmt_date, today0
|
|
|
|
|
|
def test_hybrid_skeleton_has_day_week_month():
|
|
start = fmt_date(today0())
|
|
buckets = build_bucket_skeleton(mode="HYBRID", start_date=start, horizon_days=90)
|
|
kinds = {b["kind"] for b in buckets}
|
|
assert kinds == {"DAY", "WEEK", "MONTH"}
|
|
assert sum(1 for b in buckets if b["kind"] == "DAY") == 7
|
|
|
|
|
|
def test_plan_buckets_seed_summary():
|
|
world = seed_world()
|
|
report = build_plan_buckets(world, mode="WEEK", horizon_days=30, include_forecast=True)
|
|
assert report["mode"] == "WEEK"
|
|
assert report["dailyCapacity"] == daily_capacity(world)
|
|
assert report["summary"]["firmQty"] > 0
|
|
assert report["summary"]["forecastQty"] > 0 # ACTIVE forecasts in horizon
|
|
assert report["summary"]["bucketCount"] == len(report["buckets"])
|
|
assert all("loadRatio" in b and "status" in b for b in report["buckets"])
|
|
|
|
|
|
def test_plan_buckets_exclude_forecast():
|
|
world = seed_world()
|
|
with_fc = build_plan_buckets(world, mode="MONTH", horizon_days=90, include_forecast=True)
|
|
no_fc = build_plan_buckets(world, mode="MONTH", horizon_days=90, include_forecast=False)
|
|
assert with_fc["summary"]["forecastQty"] >= no_fc["summary"]["forecastQty"]
|
|
assert no_fc["summary"]["forecastQty"] == 0
|
|
assert no_fc["summary"]["firmQty"] == with_fc["summary"]["firmQty"]
|
|
|
|
|
|
def test_day_mode_bucket_count():
|
|
start = fmt_date(add_minutes(today0(), 0))
|
|
report = build_plan_buckets(seed_world(), mode="DAY", start_date=start, horizon_days=14,
|
|
include_forecast=False)
|
|
assert len(report["buckets"]) == 14
|
|
assert all(b["kind"] == "DAY" for b in report["buckets"])
|
|
|
|
|
|
def test_infinite_capacity_no_over_status():
|
|
world = seed_world()
|
|
finite = build_plan_buckets(world, mode="WEEK", horizon_days=30, capacity_mode="FINITE")
|
|
infinite = build_plan_buckets(world, mode="WEEK", horizon_days=30, capacity_mode="INFINITE")
|
|
assert infinite["capacityMode"] == "INFINITE"
|
|
assert infinite["summary"]["overCount"] == 0
|
|
assert all(b["capacity"] is None and b["capacityUnlimited"] for b in infinite["buckets"])
|
|
assert infinite["summary"]["peakRequiredDaily"] >= 0
|
|
# 有限参照应保留缺口信息
|
|
assert infinite["summary"]["finiteOverCount"] == finite["summary"]["overCount"]
|
|
|
|
|
|
def test_rccp_compare_delta():
|
|
from server.aps_domain.planning import build_rccp_compare
|
|
cmp_ = build_rccp_compare(seed_world(), mode="HYBRID", horizon_days=90)
|
|
assert "finite" in cmp_ and "infinite" in cmp_ and "delta" in cmp_
|
|
assert cmp_["finite"]["weightedDemand"] == cmp_["infinite"]["weightedDemand"]
|
|
assert cmp_["delta"]["capacityGapDaily"] >= 0
|
|
assert cmp_["report"]["capacityMode"] == "FINITE"
|
|
|
|
|
|
def test_feasibility_portfolio_and_orders():
|
|
from server.aps_domain.planning import build_feasibility
|
|
world = seed_world()
|
|
report = build_feasibility(world, horizon_days=90, include_forecast=True)
|
|
assert report["verdict"] in ("FEASIBLE", "AT_RISK", "INFEASIBLE")
|
|
assert report["summary"]["all"] == len(report["orders"])
|
|
assert report["summary"]["all"] >= 7 # firm SOs
|
|
# 累计需求单调不减
|
|
prev = 0.0
|
|
for row in report["orders"]:
|
|
assert row["cumulativeDemand"] >= prev
|
|
prev = row["cumulativeDemand"]
|
|
assert row["status"] in ("FEASIBLE", "AT_RISK", "INFEASIBLE", "LATE")
|
|
|
|
|
|
def test_feasibility_exclude_forecast_fewer_rows():
|
|
from server.aps_domain.planning import build_feasibility
|
|
world = seed_world()
|
|
with_fc = build_feasibility(world, include_forecast=True)
|
|
no_fc = build_feasibility(world, include_forecast=False)
|
|
assert no_fc["summary"]["all"] < with_fc["summary"]["all"]
|
|
assert all(o["source"] == "FIRM" for o in no_fc["orders"])
|
|
|
|
|
|
def test_leveling_reduces_peak_or_moves():
|
|
from server.aps_domain.planning import build_leveling, build_plan_buckets
|
|
world = seed_world()
|
|
# 压低日产能制造超载，便于验证削峰
|
|
for line in world["lines"]:
|
|
line["capacityPerDay"] = 50
|
|
base = build_plan_buckets(world, mode="WEEK", horizon_days=60, capacity_mode="FINITE")
|
|
assert base["summary"]["overCount"] >= 1
|
|
report = build_leveling(world, mode="WEEK", horizon_days=60, target_load=0.85)
|
|
assert report["before"]["overCount"] >= 1
|
|
assert report["improvement"]["moveCount"] >= 1
|
|
assert report["after"]["peakLoad"] <= report["before"]["peakLoad"] + 1e-6
|
|
assert len(report["buckets"]) == len(base["buckets"])
|
|
|
|
|
|
def test_leveling_no_overload_empty_moves():
|
|
from server.aps_domain.planning import build_leveling
|
|
world = seed_world()
|
|
# 放大产能 → 无超载
|
|
for line in world["lines"]:
|
|
line["capacityPerDay"] = 100000
|
|
report = build_leveling(world, mode="WEEK", horizon_days=30)
|
|
assert report["before"]["overCount"] == 0
|
|
assert report["moves"] == []
|
|
assert report["improvement"]["movedQty"] == 0
|
|
|
|
|
|
def test_supply_need_when_capacity_crushed():
|
|
from server.aps_domain.planning import build_supply_decisions
|
|
world = seed_world()
|
|
# 极低压产能：削峰也吃不完 → NEED_SUPPLY
|
|
for line in world["lines"]:
|
|
line["capacityPerDay"] = 20
|
|
report = build_supply_decisions(world, mode="WEEK", horizon_days=60, target_load=0.85)
|
|
assert report["verdict"] == "NEED_SUPPLY"
|
|
assert report["summary"]["residualGap"] > 0
|
|
assert report["summary"]["optionCount"] >= 1
|
|
kinds = {o["kind"] for o in report["options"]}
|
|
assert "OVERTIME" in kinds or "OUTSOURCE" in kinds or "EXPAND" in kinds
|
|
assert report["options"][0]["costIndex"] <= max(o["costIndex"] for o in report["options"])
|
|
|
|
|
|
def test_supply_balanced_when_capacity_ample():
|
|
from server.aps_domain.planning import build_supply_decisions
|
|
world = seed_world()
|
|
for line in world["lines"]:
|
|
line["capacityPerDay"] = 100000
|
|
report = build_supply_decisions(world, mode="WEEK", horizon_days=30)
|
|
assert report["verdict"] == "BALANCED"
|
|
assert report["summary"]["residualGap"] == 0
|
|
assert report["options"] == []
|
|
assert report["gaps"] == []
|