aps-agent/tests/golden/test_world_due_view.py

82 lines
3.8 KiB
Python
Raw Permalink Normal View History

# ============================================================
# 交期承诺看板的数据来源回退(固定产线通道 → 柔性通道)
# 现场现象:已经跑出柔性试排,但「交期分析」5 张订单整列显示"未排产"。
# 根因:due_view 只读 scheduleVersions / productionOrders,柔性版本不在这两张表里,
# 没有固定版本时 promised 恒为 None → risk 恒为 "none"。
# ============================================================
from __future__ import annotations
from server.aps_domain.views import due_view
def _so(order_id: int, order_no: str, delivery: str) -> dict:
"""最小销售订单行(due_view 只用到这几个字段)。"""
return {
"id": order_id, "orderNo": order_no, "customerName": "演示客户",
"customerLevel": "A", "isRush": False, "deliveryDate": delivery,
"status": "APPROVED",
}
def _world(**overrides) -> dict:
"""默认:既没有固定版本,也没有柔性版本。"""
world: dict = {
"salesOrders": [
_so(1, "SO-001", "2026-10-16"),
_so(2, "SO-002", "2026-10-10"),
_so(3, "SO-003", "2026-10-17"),
],
"scheduleVersions": [], "productionOrders": [],
"flexScheduleVersions": [], "flexVirtualLines": [],
}
world.update(overrides)
return world
def test_fixed_channel_still_wins_when_present():
"""固定通道有版本时行为不变:不因存在柔性版本而改取柔性时间。"""
rows = {r["orderNo"]: r for r in due_view(_world(
scheduleVersions=[{"id": 7, "versionNo": "FV-7"}],
productionOrders=[{"id": 1, "salesOrderId": 1, "schedulingVersionId": 7,
"plannedEndDate": "2026-10-15 12:00"}],
flexScheduleVersions=[{"id": 3, "versionNo": "FLEX-3"}],
flexVirtualLines=[{"versionId": 3, "orderNo": "SO-001",
"plannedEnd": "2026-10-09 08:00"}],
))}
assert rows["SO-001"]["promisedEnd"] == "2026-10-15 12:00"
assert rows["SO-002"]["risk"] == "none" # 该版本没有它的 PO
def test_flex_fallback_computes_commit_and_risk():
"""没有固定版本时回退柔性版本:承诺完成取该订单虚拟产线的最晚完成。"""
rows = {r["orderNo"]: r for r in due_view(_world(
flexScheduleVersions=[{"id": 3, "versionNo": "FLEX-3"}, {"id": 4, "versionNo": "FLEX-4"}],
flexVirtualLines=[
{"versionId": 4, "orderNo": "SO-001", "plannedEnd": "2026-10-14 09:00"},
{"versionId": 4, "orderNo": "SO-001", "plannedEnd": "2026-10-16 09:00"}, # 同订单取最晚
{"versionId": 3, "orderNo": "SO-002", "plannedEnd": "2026-10-01 09:00"}, # 旧版本不参与
{"versionId": 4, "orderNo": "SO-003", "plannedEnd": "2026-10-20 09:00"}, # 晚于交期 → 超期
],
))}
assert rows["SO-001"]["promisedEnd"] == "2026-10-16 09:00"
assert rows["SO-001"]["marginHours"] == 9.0 # 交期 10-16 18:00 前 9 小时
assert rows["SO-001"]["risk"] == "mid"
assert rows["SO-002"]["promisedEnd"] is None # 最新版本没排 → 仍是未排产
assert rows["SO-002"]["risk"] == "none"
assert rows["SO-003"]["marginHours"] == -63.0 # 10-20 09:00 比交期(10-17 18:00)晚 63 小时
assert rows["SO-003"]["risk"] == "high"
def test_no_schedule_anywhere_keeps_empty_board():
"""两个通道都没有版本时不报错、也不编造承诺完成。"""
rows = due_view(_world())
assert [r["risk"] for r in rows] == ["none", "none", "none"]
assert all(r["promisedEnd"] is None and r["marginHours"] is None for r in rows)
def test_cancelled_order_is_not_listed():
world = _world()
world["salesOrders"][1]["status"] = "CANCELLED"
rows = due_view(world)
assert [r["orderNo"] for r in rows] == ["SO-001", "SO-003"]