101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
# ============================================================
|
|
# OR-05 预测 / 长周期订单黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from server.aps_domain.forecast import apply_forecast_action, ensure_forecast_table, list_forecasts
|
|
from server.engines import get_engine
|
|
from server.engines.base import EngineParams
|
|
from server.state.seed import seed_world
|
|
from server.timeutil import add_minutes, fmt_date, today0
|
|
|
|
|
|
def _next_id_factory(world):
|
|
tables = {
|
|
"forecastOrder": "forecastOrders",
|
|
"salesOrder": "salesOrders",
|
|
"productionOrder": "productionOrders",
|
|
"workOrder": "workOrders",
|
|
"scheduleVersion": "scheduleVersions",
|
|
"conflict": "conflicts",
|
|
"log": "logs",
|
|
"audit": "auditEvents",
|
|
}
|
|
counters: dict[str, int] = {}
|
|
|
|
def next_id(kind: str):
|
|
if kind not in counters:
|
|
rows = world.get(tables.get(kind, kind + "s"), [])
|
|
counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
|
|
counters[kind] += 1
|
|
return counters[kind]
|
|
|
|
return next_id
|
|
|
|
|
|
def _run(world, *, include_forecast: bool = False):
|
|
params = EngineParams(
|
|
orderIds=[], engineType="RULE", strategyTemplate="COMPREHENSIVE",
|
|
planningHorizonDays=14,
|
|
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
|
includeForecast=include_forecast,
|
|
)
|
|
return get_engine("RULE").solve(world, params, _next_id_factory(world))
|
|
|
|
|
|
def test_forecast_excluded_by_default():
|
|
world = seed_world()
|
|
assert len(world["forecastOrders"]) == 3
|
|
active = [f for f in world["forecastOrders"] if f["status"] == "ACTIVE"]
|
|
assert len(active) == 2
|
|
result = _run(world, include_forecast=False)
|
|
assert result.orderCount == 7
|
|
|
|
|
|
def test_forecast_included_in_trial_run():
|
|
world = seed_world()
|
|
result = _run(world, include_forecast=True)
|
|
assert result.orderCount == 9 # 7 firm + 2 ACTIVE forecast
|
|
assert any(po.get("salesOrderNo", "").startswith("FC") for po in world["productionOrders"])
|
|
|
|
|
|
def test_forecast_convert_to_sales_order():
|
|
world = seed_world()
|
|
fc = next(f for f in world["forecastOrders"] if f["status"] == "ACTIVE")
|
|
before_so = len(world["salesOrders"])
|
|
applied = apply_forecast_action(world, _next_id_factory(world), "forecast.convert", {
|
|
"id": fc["id"],
|
|
})
|
|
assert applied["forecast"]["status"] == "CONSUMED"
|
|
assert applied["convertedOrderNo"]
|
|
assert len(world["salesOrders"]) == before_so + 1
|
|
order = applied["order"]
|
|
assert order["status"] == "APPROVED"
|
|
result = _run(world, include_forecast=False)
|
|
assert result.orderCount == 8
|
|
|
|
|
|
def test_forecast_upsert_and_list():
|
|
world = seed_world()
|
|
product_code = next(m["code"] for m in world["materials"] if m["type"] == "FINISHED_PRODUCT")
|
|
applied = apply_forecast_action(world, _next_id_factory(world), "forecast.upsert", {
|
|
"productCode": product_code,
|
|
"quantity": 120,
|
|
"bucket": "WEEK",
|
|
"status": "ACTIVE",
|
|
"confidence": 0.8,
|
|
})
|
|
assert applied["created"] is True
|
|
assert applied["forecast"]["forecastNo"].startswith("FC")
|
|
rows = list_forecasts(world)
|
|
assert any(r["id"] == applied["forecast"]["id"] for r in rows)
|
|
|
|
|
|
def test_ensure_forecast_table_on_legacy_world():
|
|
world = seed_world()
|
|
del world["forecastOrders"]
|
|
ensure_forecast_table(world)
|
|
assert world["forecastOrders"] == []
|
|
result = _run(world, include_forecast=True)
|
|
assert result.orderCount == 7
|