127 lines
5.2 KiB
Python
127 lines
5.2 KiB
Python
# ============================================================
|
|
# EX-04 甘特拖拽调程黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from server.aps_domain.adjust import commit_flex_adjust, preview_flex_adjust, snap_dt
|
|
from server.aps_domain.flex import run_flex_schedule
|
|
from server.timeutil import add_minutes, parse_dt
|
|
from server.state.seed import seed_world
|
|
|
|
|
|
class _MemStore:
|
|
def __init__(self, data):
|
|
self.data = data
|
|
|
|
def next_id(self, kind: str) -> int:
|
|
key = f"_c_{kind}"
|
|
self.data[key] = self.data.get(key, 3000) + 1
|
|
return self.data[key]
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
def test_snap_dt_15min():
|
|
dt = parse_dt("2026-07-23 08:07")
|
|
assert snap_dt(dt).minute in (0, 15)
|
|
assert snap_dt(parse_dt("2026-07-23 08:08")).minute == 15
|
|
|
|
|
|
def test_preview_ok_and_overlap_fail():
|
|
store = _MemStore(seed_world())
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
wos = [w for w in store.data["flexWorkOrders"]
|
|
if w["versionId"] == store.data["flexScheduleVersions"][-1]["id"]]
|
|
assert wos
|
|
from server.timeutil import fmt_dt
|
|
# 找一条能后移 15 分钟的工单
|
|
ok = None
|
|
for wo in wos:
|
|
later = fmt_dt(add_minutes(parse_dt(wo["plannedStartTime"]), 15))
|
|
cand = preview_flex_adjust(store.data, wo["id"], later)
|
|
if cand["ok"] and cand["moves"]:
|
|
ok = cand
|
|
break
|
|
assert ok is not None, "种子排产后应至少有一条可平移 15 分钟的工单"
|
|
# 故意拖到同设备另一工单起点 → 应报不可行（若存在同伴）
|
|
wo0 = next(w for w in wos if w["id"] == ok["woId"])
|
|
same_eq = [w for w in wos if w["equipmentId"] == wo0["equipmentId"] and w["id"] != wo0["id"]]
|
|
if same_eq:
|
|
bad = preview_flex_adjust(store.data, wo0["id"], same_eq[0]["plannedStartTime"])
|
|
assert bad["ok"] is False or bad["reasons"] is not None
|
|
|
|
|
|
def test_commit_creates_new_version():
|
|
store = _MemStore(seed_world())
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
before = len(store.data["flexScheduleVersions"])
|
|
from server.timeutil import fmt_dt
|
|
wos = [w for w in store.data["flexWorkOrders"]
|
|
if w["versionId"] == store.data["flexScheduleVersions"][-1]["id"]]
|
|
target = None
|
|
later = None
|
|
for wo in wos:
|
|
cand_start = fmt_dt(add_minutes(parse_dt(wo["plannedStartTime"]), 15))
|
|
prev = preview_flex_adjust(store.data, wo["id"], cand_start)
|
|
if prev["ok"] and prev["moves"]:
|
|
target, later = wo, cand_start
|
|
break
|
|
assert target is not None
|
|
r = commit_flex_adjust(store, target["id"], later, actor="test")
|
|
assert len(store.data["flexScheduleVersions"]) == before + 1
|
|
assert r["versionNo"]
|
|
latest = store.data["flexScheduleVersions"][-1]["id"]
|
|
assert any(w.get("adjustedFrom") for w in store.data["flexWorkOrders"]
|
|
if w["versionId"] == latest)
|
|
|
|
|
|
def test_preview_does_not_mutate():
|
|
store = _MemStore(seed_world())
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
wo = next(w for w in store.data["flexWorkOrders"]
|
|
if w["versionId"] == store.data["flexScheduleVersions"][-1]["id"])
|
|
old = wo["plannedStartTime"]
|
|
from server.timeutil import fmt_dt
|
|
preview_flex_adjust(store.data, wo["id"],
|
|
fmt_dt(add_minutes(parse_dt(old), 60)))
|
|
assert wo["plannedStartTime"] == old
|
|
assert len(store.data["flexScheduleVersions"]) == 1
|
|
|
|
|
|
def test_fixed_preview_and_commit():
|
|
from server.aps_domain.adjust import commit_fixed_adjust, preview_fixed_adjust
|
|
from server.engines import get_engine
|
|
from server.engines.base import EngineParams
|
|
from server.timeutil import fmt_date, fmt_dt, today0
|
|
store = _MemStore(seed_world())
|
|
ep = EngineParams(
|
|
orderIds=[], engineType="RULE", strategyTemplate="COMPREHENSIVE",
|
|
planningHorizonDays=14,
|
|
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
|
)
|
|
get_engine("RULE").solve(store.data, ep, store.next_id)
|
|
before = len(store.data["scheduleVersions"])
|
|
latest = store.data["scheduleVersions"][-1]
|
|
po_ids = {p["id"] for p in store.data["productionOrders"]
|
|
if p.get("schedulingVersionId") == latest["id"]}
|
|
wos = [w for w in store.data["workOrders"] if w["productionOrderId"] in po_ids]
|
|
assert wos
|
|
target = None
|
|
later = None
|
|
for wo in wos:
|
|
cand = fmt_dt(add_minutes(parse_dt(wo["plannedStartTime"]), 15))
|
|
prev = preview_fixed_adjust(store.data, wo["id"], cand)
|
|
if prev["ok"] and prev["moves"]:
|
|
target, later = wo, cand
|
|
break
|
|
assert target is not None, "应至少有一条可平移 15 分钟的固定轨工单"
|
|
r = commit_fixed_adjust(store, target["id"], later, actor="test")
|
|
assert len(store.data["scheduleVersions"]) == before + 1
|
|
assert r["versionNo"].startswith("V")
|
|
new_vid = store.data["scheduleVersions"][-1]["id"]
|
|
new_po_ids = {p["id"] for p in store.data["productionOrders"]
|
|
if p.get("schedulingVersionId") == new_vid}
|
|
assert any(w.get("adjustedFrom") for w in store.data["workOrders"]
|
|
if w["productionOrderId"] in new_po_ids)
|