65 lines
2.2 KiB
Python
65 lines
2.2 KiB
Python
# ============================================================
|
||
# 工程目录试排（trial）只走 PoolEngine，不改变默认闭环路径
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.aps_domain.flex import run_flex_schedule
|
||
from server.state.seed import empty_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, 1000) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def _site_world() -> dict:
|
||
world = empty_world()
|
||
world["flexOrders"] = [{
|
||
"id": 1, "orderNo": "SO-TRIAL", "productCode": "FG", "productName": "成品",
|
||
"quantity": 2, "dueDate": "2026-08-20", "priority": 5, "status": "RELEASED",
|
||
}]
|
||
world["flexMaterials"] = [{
|
||
"id": 1, "code": "FG", "name": "成品", "type": "FINISHED_PRODUCT",
|
||
"unit": "件", "stock": 0, "inTransit": 0, "safetyStock": 0,
|
||
"procurementLeadTime": 0, "sourcingType": "MAKE",
|
||
}]
|
||
world["flexOperations"] = [{"id": 1, "code": "CUT", "name": "下料", "changeoverMin": 10}]
|
||
world["flexRoutings"] = [{
|
||
"id": 1, "productCode": "FG", "productName": "成品", "seq": 1,
|
||
"operationCode": "CUT", "operationName": "下料", "stdTimePerUnit": 5.0,
|
||
}]
|
||
world["flexEquipment"] = [{
|
||
"id": 1, "code": "EQ-1", "name": "切割机", "status": "RUNNING",
|
||
"capabilities": ["CUT"], "availabilityRate": 1.0,
|
||
}]
|
||
return world
|
||
|
||
|
||
def test_trial_pool_schedule_produces_virtual_lines():
|
||
store = _MemStore(_site_world())
|
||
result = run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test", trial=True)
|
||
|
||
assert result["executionMode"] == "POOL_TRIAL"
|
||
assert result["vlCount"] == 1
|
||
assert result["woCount"] == 1
|
||
version_id = result["versionId"]
|
||
assert len([
|
||
row for row in store.data["flexVirtualLines"]
|
||
if row.get("versionId") == version_id
|
||
]) == 1
|
||
|
||
|
||
def test_default_run_keeps_closed_loop_mode():
|
||
store = _MemStore(_site_world())
|
||
result = run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
||
|
||
assert result["executionMode"] == "CLOSED_LOOP_V1"
|