151 lines
8.3 KiB
Python
151 lines
8.3 KiB
Python
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from copy import deepcopy
|
|||
|
|
import json
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
|
|||
|
|
from server.aps_domain.closed_loop_problem import build_closed_loop_problem
|
|||
|
|
|
|||
|
|
BUSINESS_DATE = "2026-08-02"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _base_world() -> dict:
|
|||
|
|
return {
|
|||
|
|
"materials": [], "flexMaterials": [], "boms": [], "bomItems": [], "flexBom": [],
|
|||
|
|
"routings": [], "routingSteps": [], "routingOperations": [], "operations": [],
|
|||
|
|
"flexRoutings": [], "workstations": [], "workstationOperations": [],
|
|||
|
|
"equipment": [], "flexEquipment": [], "salesOrders": [],
|
|||
|
|
"purchaseOrders": [], "outsourceOrders": [],
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _ready_route(world: dict, product_id: int, routing_id: int, operation_id: int, code: str = "CUT") -> None:
|
|||
|
|
world["routings"].append({"id": routing_id, "productId": product_id, "isDefault": True, "status": "ACTIVE"})
|
|||
|
|
world["routingSteps"].append({"id": routing_id, "routingId": routing_id, "operationId": operation_id, "sequenceNo": 1, "isExternal": False})
|
|||
|
|
world["operations"].append({"id": operation_id, "code": code, "name": code})
|
|||
|
|
world["flexEquipment"].append({"id": operation_id, "code": f"EQ-{operation_id}", "status": "RUNNING", "capabilities": [code]})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_contract_global_stock_is_allocated_once_and_world_is_read_only():
|
|||
|
|
world = _base_world()
|
|||
|
|
world["materials"] = [
|
|||
|
|
{"id": 1, "code": "FG", "name": "Finished", "type": "FINISHED_PRODUCT", "unit": "PCS"},
|
|||
|
|
{"id": 2, "code": "RM", "name": "Raw", "type": "RAW_MATERIAL", "unit": "KG", "stock": 10, "safetyStock": 0},
|
|||
|
|
]
|
|||
|
|
world["boms"] = [{"id": 1, "productId": 1, "isDefault": True, "status": "ACTIVE"}]
|
|||
|
|
world["bomItems"] = [{"id": 1, "bomId": 1, "materialId": 2, "quantity": 8}]
|
|||
|
|
_ready_route(world, 1, 1, 1)
|
|||
|
|
world["salesOrders"] = [
|
|||
|
|
{"id": 1, "orderNo": "SO-1", "status": "APPROVED", "deliveryDate": "2026-08-04", "priority": 2,
|
|||
|
|
"items": [{"id": 11, "productId": 1, "productCode": "FG", "quantity": 1, "unit": "PCS"}]},
|
|||
|
|
{"id": 2, "orderNo": "SO-2", "status": "APPROVED", "deliveryDate": "2026-08-05", "priority": 1,
|
|||
|
|
"items": [{"id": 22, "productId": 1, "productCode": "FG", "quantity": 1, "unit": "PCS"}]},
|
|||
|
|
]
|
|||
|
|
before = deepcopy(world)
|
|||
|
|
|
|||
|
|
problem = build_closed_loop_problem(world, business_date=BUSINESS_DATE)
|
|||
|
|
payload = problem.to_dict()
|
|||
|
|
|
|||
|
|
assert world == before
|
|||
|
|
assert {"problemId", "manufacturingDemands", "supplyEvents", "blockers", "summary", "sourceHash"} <= payload.keys()
|
|||
|
|
assert payload["businessDate"] == BUSINESS_DATE
|
|||
|
|
stock = next(event for event in problem.supply_events if event.source == "STOCK")
|
|||
|
|
assert sum(allocation.quantity for allocation in stock.allocations) == 10
|
|||
|
|
assert sum(balance.shortage_quantity for balance in problem.demand_balances) == 6
|
|||
|
|
buy = [row for row in problem.requirements if row.sourcing_type == "BUY"]
|
|||
|
|
assert len(buy) == 2
|
|||
|
|
assert all(sum(peg.quantity for peg in row.pegging_refs) == row.quantity for row in buy)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_multilevel_make_buy_subcontract_pegging_and_blockers_are_explicit():
|
|||
|
|
world = _base_world()
|
|||
|
|
world["materials"] = [
|
|||
|
|
{"id": 1, "code": "FG", "name": "Finished", "type": "FINISHED_PRODUCT", "unit": "PCS"},
|
|||
|
|
{"id": 2, "code": "SFG", "name": "Semi", "type": "SEMI_FINISHED", "unit": "PCS"},
|
|||
|
|
{"id": 3, "code": "RM", "name": "Raw", "type": "RAW_MATERIAL", "unit": "KG"},
|
|||
|
|
]
|
|||
|
|
world["boms"] = [
|
|||
|
|
{"id": 1, "productId": 1, "isDefault": True, "status": "ACTIVE"},
|
|||
|
|
{"id": 2, "productId": 2, "isDefault": True, "status": "ACTIVE"},
|
|||
|
|
]
|
|||
|
|
world["bomItems"] = [
|
|||
|
|
{"id": 1, "bomId": 1, "materialId": 2, "quantity": 2},
|
|||
|
|
{"id": 2, "bomId": 2, "materialId": 3, "quantity": 3},
|
|||
|
|
]
|
|||
|
|
_ready_route(world, 1, 1, 1, "CUT")
|
|||
|
|
world["routings"].append({"id": 2, "productId": 2, "isDefault": True, "status": "ACTIVE"})
|
|||
|
|
world["routingSteps"].extend([
|
|||
|
|
{"id": 2, "routingId": 2, "operationId": 2, "sequenceNo": 1, "isExternal": False},
|
|||
|
|
{"id": 3, "routingId": 2, "operationId": 3, "sequenceNo": 2, "isExternal": True},
|
|||
|
|
])
|
|||
|
|
world["operations"].extend([
|
|||
|
|
{"id": 2, "code": "WELD", "name": "Weld"},
|
|||
|
|
{"id": 3, "code": "WZ01", "name": "External coat", "type": "EXTERNAL"},
|
|||
|
|
])
|
|||
|
|
world["flexEquipment"].append({"id": 2, "code": "EQ-W", "status": "RUNNING", "capabilities": ["WELD"]})
|
|||
|
|
world["salesOrders"] = [{"id": 1, "orderNo": "SO-MULTI", "status": "APPROVED", "deliveryDate": "2026-08-10",
|
|||
|
|
"items": [{"id": 1, "productId": 1, "productCode": "FG", "quantity": 1, "unit": "PCS"}]}]
|
|||
|
|
world["purchaseOrders"] = [{"id": 1, "orderNo": "PO-1", "materialId": 3, "quantity": 6, "expectedDate": "2026-08-09", "status": "CONFIRMED"}]
|
|||
|
|
world["outsourceOrders"] = [{"id": 1, "orderNo": "OUT-1", "productCode": "SFG", "operationCode": "WZ01", "sequenceNo": 2, "quantity": 2, "expectedDate": "2026-08-09", "status": "CONFIRMED"}]
|
|||
|
|
|
|||
|
|
problem = build_closed_loop_problem(world, business_date=BUSINESS_DATE)
|
|||
|
|
|
|||
|
|
assert problem.summary["makeCount"] == 2
|
|||
|
|
assert problem.summary["buyCount"] == 1
|
|||
|
|
assert problem.summary["subcontractCount"] == 1
|
|||
|
|
non_roots = [row for row in problem.requirements if row.parent_requirement_ids]
|
|||
|
|
assert all(sum(peg.quantity for peg in row.pegging_refs) == row.quantity for row in non_roots)
|
|||
|
|
assert not any(blocker.code == "SUPPLY_SHORTAGE" for blocker in problem.blockers)
|
|||
|
|
assert all(demand.release_status == "READY_FOR_SCHEDULING" for demand in problem.manufacturing_demands)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_draft_or_late_supply_is_not_used_and_cycle_is_blocked():
|
|||
|
|
world = _base_world()
|
|||
|
|
world["materials"] = [
|
|||
|
|
{"id": 1, "code": "A", "name": "A", "type": "FINISHED_PRODUCT"},
|
|||
|
|
{"id": 2, "code": "B", "name": "B", "type": "SEMI_FINISHED"},
|
|||
|
|
{"id": 3, "code": "RM", "name": "RM", "type": "RAW_MATERIAL"},
|
|||
|
|
]
|
|||
|
|
world["boms"] = [{"id": 1, "productId": 1, "isDefault": True}, {"id": 2, "productId": 2, "isDefault": True}]
|
|||
|
|
world["bomItems"] = [
|
|||
|
|
{"id": 1, "bomId": 1, "materialId": 2, "quantity": 1},
|
|||
|
|
{"id": 2, "bomId": 1, "materialId": 3, "quantity": 8},
|
|||
|
|
{"id": 3, "bomId": 2, "materialId": 1, "quantity": 1},
|
|||
|
|
]
|
|||
|
|
world["flexRoutings"] = [{"productCode": "A", "seq": 1, "operationCode": "ASM", "stdTimeSource": "模板", "isExternal": False}]
|
|||
|
|
world["salesOrders"] = [{"id": 1, "orderNo": "SO-BLOCK", "status": "APPROVED", "deliveryDate": "2026-08-04", "items": [{"id": 1, "productId": 1, "productCode": "A", "quantity": 1}]}]
|
|||
|
|
world["purchaseOrders"] = [
|
|||
|
|
{"id": 1, "materialId": 3, "quantity": 8, "expectedDate": "2026-08-03", "status": "DRAFT"},
|
|||
|
|
{"id": 2, "materialId": 3, "quantity": 8, "expectedDate": "2026-08-05", "status": "CONFIRMED"},
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
problem = build_closed_loop_problem(world, business_date=BUSINESS_DATE)
|
|||
|
|
codes = {blocker.code for blocker in problem.blockers}
|
|||
|
|
|
|||
|
|
assert {"BOM_CYCLE", "TEMPLATE_ROUTING_UNCONFIRMED", "MISSING_RESOURCE_CAPABILITY", "MISSING_ROUTING", "UNTRUSTED_DRAFT_SUPPLY", "SUPPLY_SHORTAGE"} <= codes
|
|||
|
|
assert all(not event.allocations for event in problem.supply_events if event.event_id.startswith("PO:"))
|
|||
|
|
assert all(demand.release_status == "BLOCKED" for demand in problem.manufacturing_demands)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def test_current_mom_data_is_read_only_and_only_yields_requirements_and_blockers():
|
|||
|
|
path = Path("server/data/world-proj_712276ba.json")
|
|||
|
|
if not path.exists():
|
|||
|
|
pytest.skip("local MOM project world is not present")
|
|||
|
|
world = json.loads(path.read_text(encoding="utf-8"))
|
|||
|
|
if not any(row.get("orderNo") == "MOM--00280" for row in world.get("salesOrders", [])):
|
|||
|
|
pytest.skip("MOM--00280 is not present in this local project world")
|
|||
|
|
before_orders = deepcopy({key: world.get(key) for key in ("purchaseOrders", "outsourceOrders", "productionOrders", "workOrders", "flexWorkOrders")})
|
|||
|
|
|
|||
|
|
problem = build_closed_loop_problem(world, business_date=BUSINESS_DATE, order_nos=["MOM--00280"])
|
|||
|
|
|
|||
|
|
assert problem.summary["makeCount"] == 106
|
|||
|
|
assert problem.summary["buyCount"] == 124
|
|||
|
|
assert problem.summary["subcontractCount"] == 0
|
|||
|
|
assert problem.summary["blockerCounts"]["MISSING_ROUTING"] == 39
|
|||
|
|
assert problem.summary["blockerCounts"]["TEMPLATE_ROUTING_UNCONFIRMED"] == 67
|
|||
|
|
assert problem.summary["blockerCounts"]["MISSING_RESOURCE_CAPABILITY"] == 67
|
|||
|
|
assert all(demand.release_status == "BLOCKED" for demand in problem.manufacturing_demands)
|
|||
|
|
assert before_orders == {key: world.get(key) for key in before_orders}
|