172 lines
5.2 KiB
Python
172 lines
5.2 KiB
Python
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.closed_loop_problem import build_closed_loop_problem
|
|
from server.aps_domain.mrp import decompose_orders
|
|
|
|
|
|
BUSINESS_DATE = "2026-08-02"
|
|
|
|
|
|
def _provenance_world() -> dict:
|
|
return {
|
|
"businessDate": BUSINESS_DATE,
|
|
"materials": [
|
|
{
|
|
"id": 1,
|
|
"code": "FG-PROV",
|
|
"name": "Provenance finished good",
|
|
"type": "FINISHED_PRODUCT",
|
|
"procurementType": "MAKE",
|
|
"unit": "PCS",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
},
|
|
{
|
|
"id": 2,
|
|
"code": "RM-PROV-A",
|
|
"name": "Provenance raw A",
|
|
"type": "RAW_MATERIAL",
|
|
"procurementType": "BUY",
|
|
"unit": "PCS",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
"procurementLeadTime": 2,
|
|
},
|
|
{
|
|
"id": 3,
|
|
"code": "RM-PROV-B",
|
|
"name": "Provenance raw B",
|
|
"type": "RAW_MATERIAL",
|
|
"procurementType": "BUY",
|
|
"unit": "PCS",
|
|
"stock": 0,
|
|
"inTransit": 0,
|
|
"procurementLeadTime": 4,
|
|
},
|
|
],
|
|
"boms": [{"id": 1, "productId": 1, "isDefault": True}],
|
|
"bomItems": [
|
|
{"id": 1, "bomId": 1, "materialId": 2, "quantity": 2, "isKeyMaterial": True},
|
|
{"id": 2, "bomId": 1, "materialId": 3, "quantity": 1, "isKeyMaterial": False},
|
|
],
|
|
"salesOrders": [
|
|
{
|
|
"id": 10,
|
|
"orderNo": "SO-PROVENANCE-001",
|
|
"status": "APPROVED",
|
|
"deliveryDate": "2026-08-10",
|
|
"items": [
|
|
{
|
|
"id": 11,
|
|
"productId": 1,
|
|
"productCode": "FG-PROV",
|
|
"productName": "Provenance finished good",
|
|
"quantity": 3,
|
|
"unit": "PCS",
|
|
"status": "APPROVED",
|
|
}
|
|
],
|
|
}
|
|
],
|
|
"routings": [],
|
|
"routingSteps": [],
|
|
"routingOperations": [],
|
|
"operations": [],
|
|
"flexRoutings": [],
|
|
"workstations": [],
|
|
"workstationOperations": [],
|
|
"equipment": [],
|
|
"flexEquipment": [],
|
|
"inventory": [],
|
|
"purchaseOrders": [],
|
|
"outsourceOrders": [],
|
|
"makeSuggestions": [],
|
|
"flexOrders": [],
|
|
}
|
|
|
|
|
|
def _next_id_factory(world: dict):
|
|
tables = {
|
|
"purchaseOrder": "purchaseOrders",
|
|
"outsourceOrder": "outsourceOrders",
|
|
}
|
|
counters: dict[str, int] = {}
|
|
|
|
def next_id(kind: str) -> int:
|
|
if kind not in counters:
|
|
rows = world.get(tables.get(kind, kind + "s"), [])
|
|
counters[kind] = max(
|
|
(int(row.get("id") or 0) for row in rows if isinstance(row, dict)),
|
|
default=100,
|
|
)
|
|
counters[kind] += 1
|
|
return counters[kind]
|
|
|
|
return next_id
|
|
|
|
|
|
def _problem(world: dict):
|
|
return build_closed_loop_problem(
|
|
world,
|
|
business_date=BUSINESS_DATE,
|
|
order_nos=["SO-PROVENANCE-001"],
|
|
strict=True,
|
|
)
|
|
|
|
|
|
def test_repeated_mrp_decomposition_keeps_planning_identity_stable():
|
|
world = _provenance_world()
|
|
next_id = _next_id_factory(world)
|
|
|
|
first = decompose_orders(world, next_id, "SO-PROVENANCE-001")
|
|
assert len(first["purchase"]) == 2
|
|
for row in world["purchaseOrders"]:
|
|
row["createdAt"] = "2026-08-02 08:00"
|
|
first_technical_identity = {
|
|
(row["id"], row["orderNo"], row["createdAt"])
|
|
for row in world["purchaseOrders"]
|
|
}
|
|
first_problem = _problem(world)
|
|
|
|
second = decompose_orders(world, next_id, "SO-PROVENANCE-001")
|
|
assert len(second["purchase"]) == 2
|
|
for row in world["purchaseOrders"]:
|
|
row["createdAt"] = "2026-08-03 09:30"
|
|
row["updatedAt"] = "2026-08-03 09:31"
|
|
world["purchaseOrders"].reverse()
|
|
second_technical_identity = {
|
|
(row["id"], row["orderNo"], row["createdAt"])
|
|
for row in world["purchaseOrders"]
|
|
}
|
|
second_problem = _problem(world)
|
|
|
|
assert second_technical_identity.isdisjoint(first_technical_identity)
|
|
assert second_problem.sourceHash == first_problem.sourceHash
|
|
assert second_problem.problemId == first_problem.problemId
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"mutate",
|
|
[
|
|
lambda row: row.__setitem__("quantity", float(row["quantity"]) + 1),
|
|
lambda row: row.__setitem__("materialCode", "RM-PROV-CHANGED"),
|
|
lambda row: row.__setitem__("status", "APPROVED"),
|
|
],
|
|
ids=["quantity", "material", "supply-status"],
|
|
)
|
|
def test_business_supply_semantics_change_planning_source_hash(mutate):
|
|
world = _provenance_world()
|
|
decompose_orders(world, _next_id_factory(world), "SO-PROVENANCE-001")
|
|
baseline = _problem(world)
|
|
|
|
changed = deepcopy(world)
|
|
mutate(changed["purchaseOrders"][0])
|
|
changed_problem = _problem(changed)
|
|
|
|
assert changed_problem.sourceHash != baseline.sourceHash
|
|
assert changed_problem.problemId != baseline.problemId
|