174 lines
7.9 KiB
Python
174 lines
7.9 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from server.aps_domain.closed_loop_problem import build_closed_loop_problem
|
|
from server.aps_domain.closed_loop_runtime import closed_loop_to_problem_v2, run_closed_loop_candidate
|
|
from server.engines.external_engine import ExternalEngine, run_external_flex
|
|
from server.integrations.algo_skill_stub import solve_problem
|
|
|
|
BUSINESS_DATE = "2026-08-05"
|
|
|
|
|
|
class _MemStore:
|
|
def __init__(self, data: dict):
|
|
self.data = data
|
|
self.world_key = "activity-input-consumer"
|
|
self._counters: dict[str, int] = {}
|
|
|
|
def next_id(self, kind: str) -> int:
|
|
self._counters[kind] = self._counters.get(kind, 0) + 1
|
|
return self._counters[kind]
|
|
|
|
def save(self) -> None:
|
|
return None
|
|
|
|
|
|
def _multi_step_bom_world() -> dict:
|
|
return {
|
|
"businessDate": BUSINESS_DATE,
|
|
"salesOrders": [{
|
|
"id": 2, "orderNo": "SO-MULTI", "status": "APPROVED",
|
|
"deliveryDate": "2026-08-12", "priority": 1,
|
|
"items": [{"id": 21, "productId": 1, "productCode": "FG", "quantity": 1, "unit": "PCS"}],
|
|
}],
|
|
"materials": [
|
|
{"id": 1, "code": "FG", "name": "Finished", "type": "FINISHED_PRODUCT", "unit": "PCS"},
|
|
{"id": 2, "code": "RM-A", "name": "Raw A", "type": "RAW_MATERIAL", "unit": "PCS", "stock": 1},
|
|
{"id": 3, "code": "RM-B", "name": "Raw B", "type": "RAW_MATERIAL", "unit": "PCS", "stock": 2},
|
|
],
|
|
"flexMaterials": [],
|
|
"boms": [{"id": 1, "productId": 1, "isDefault": True, "status": "ACTIVE"}],
|
|
"bomItems": [
|
|
{"id": 1, "bomId": 1, "materialId": 2, "quantity": 1},
|
|
{"id": 2, "bomId": 1, "materialId": 3, "quantity": 2},
|
|
],
|
|
"flexBom": [],
|
|
"routings": [{"id": 1, "productId": 1, "isDefault": True, "status": "ACTIVE", "version": "V1"}],
|
|
"routingSteps": [
|
|
{"id": 1, "routingId": 1, "operationId": 1, "sequenceNo": 1, "runTimePerUnit": 10, "isExternal": False},
|
|
{"id": 2, "routingId": 1, "operationId": 2, "sequenceNo": 2, "runTimePerUnit": 10, "isExternal": False},
|
|
{"id": 3, "routingId": 1, "operationId": 3, "sequenceNo": 3, "runTimePerUnit": 10, "isExternal": False},
|
|
],
|
|
"operations": [
|
|
{"id": 1, "code": "CUT", "name": "Cut"},
|
|
{"id": 2, "code": "WELD", "name": "Weld"},
|
|
{"id": 3, "code": "ASM", "name": "Assembly"},
|
|
],
|
|
"flexRoutings": [],
|
|
"flexEquipment": [
|
|
{"id": 1, "code": "EQ-CUT", "name": "Cutter", "status": "RUNNING", "capabilities": ["CUT"]},
|
|
{"id": 2, "code": "EQ-WELD", "name": "Welder", "status": "RUNNING", "capabilities": ["WELD"]},
|
|
{"id": 3, "code": "EQ-ASM", "name": "Assembler", "status": "RUNNING", "capabilities": ["ASM"]},
|
|
],
|
|
"flexCalendar": [{"startTime": "08:00", "endTime": "17:00", "breaks": [], "workdays": [1, 2, 3, 4, 5]}],
|
|
"purchaseOrders": [], "outsourceOrders": [], "maintenance": [], "scheduleVersions": [],
|
|
"flexParams": {"afterDays": 7, "weights": {"weightedTardiness": 1.0}},
|
|
}
|
|
|
|
|
|
def _assert_consumer_alignment(problem) -> None:
|
|
make = next(row for row in problem.requirements if row.sourcingType.value == "MAKE")
|
|
activities = sorted(
|
|
(row for row in problem.activities if row.requirementId == make.requirementId),
|
|
key=lambda row: row.sequence,
|
|
)
|
|
assert len(activities) == 3
|
|
first, *later = activities
|
|
assert len(first.inputRequirementIds) == 2
|
|
assert all(not row.inputRequirementIds for row in later)
|
|
requirement_by_id = {row.requirementId: row for row in problem.requirements}
|
|
for child_id in first.inputRequirementIds:
|
|
child = requirement_by_id[child_id]
|
|
assert child.requiredByActivityIds == (first.activityId,)
|
|
assert child.requiredByActivityId == first.activityId
|
|
|
|
|
|
def test_consumer_alignment_survives_mapping_and_pool_solve() -> None:
|
|
world = _multi_step_bom_world()
|
|
closed_loop = build_closed_loop_problem(world, business_date=BUSINESS_DATE, strict=True)
|
|
_assert_consumer_alignment(closed_loop_to_problem_v2(world, closed_loop))
|
|
store = _MemStore(world)
|
|
result = run_closed_loop_candidate(
|
|
world, store.next_id, business_date=BUSINESS_DATE, order_nos=["SO-MULTI"]
|
|
)
|
|
codes = {row["code"] for row in result["validation"]["hardViolations"]}
|
|
assert "ACTIVITY_INPUT_CONSUMER_MISMATCH" not in codes
|
|
assert result["solveStatus"] == "FEASIBLE"
|
|
assert result["woCount"] == 3
|
|
|
|
|
|
def test_execute_decomposition_persists_consumer_aligned_problem(monkeypatch) -> None:
|
|
import server.gateway.app as gateway_module
|
|
from tests.auth_provider import install_test_auth
|
|
|
|
world = _multi_step_bom_world()
|
|
store = _MemStore(world)
|
|
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
|
install_test_auth(monkeypatch, "tenant-activity-input-consumer")
|
|
client = TestClient(gateway_module.create_app())
|
|
assert client.post("/api/auth/login", json={"username": "planner"}).status_code == 200
|
|
response = client.post("/api/mrp/decompose", json={"sessionId": "consumer-regression", "orderNo": "SO-MULTI"})
|
|
assert response.status_code == 200
|
|
problem = response.json()["result"]["closedLoop"]["schedulingProblemV2"]
|
|
activities = sorted(problem["activities"], key=lambda row: row["sequence"])
|
|
first, *later = activities
|
|
assert len(first["inputRequirementIds"]) == 2
|
|
assert all(not row["inputRequirementIds"] for row in later)
|
|
requirement_by_id = {row["requirementId"]: row for row in problem["requirements"]}
|
|
for child_id in first["inputRequirementIds"]:
|
|
assert requirement_by_id[child_id]["requiredByActivityIds"] == [first["activityId"]]
|
|
assert world["closedLoopProblems"][-1]["schedulingProblemV2"] == problem
|
|
|
|
|
|
def test_external_solve_preserves_consumer_alignment(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("APS_SKILLS_PATH", str(tmp_path / "skills.json"))
|
|
import server.agent_core.skills as skills_module
|
|
|
|
skills_module._registry = None
|
|
world = _multi_step_bom_world()
|
|
store = _MemStore(world)
|
|
|
|
def sequential_solution(self, skill, problem):
|
|
solution = solve_problem(problem)
|
|
cursor = datetime(2026, 8, 6, 8, 0)
|
|
operations = []
|
|
for operation in sorted(solution.operations, key=lambda row: (row.orderId, row.seq)):
|
|
start = cursor
|
|
end = start + timedelta(minutes=float(operation.runMin))
|
|
operations.append(operation.model_copy(update={
|
|
"start": start.strftime("%Y-%m-%d %H:%M"),
|
|
"end": end.strftime("%Y-%m-%d %H:%M"),
|
|
}))
|
|
cursor = end
|
|
return solution.model_copy(update={"operations": tuple(operations)})
|
|
|
|
monkeypatch.setattr(ExternalEngine, "_call_skill", sequential_solution)
|
|
summary = run_external_flex(store, skill_id="algo.stub")
|
|
assert summary["solveStatus"] == "FEASIBLE"
|
|
assert summary["validation"]["valid"] is True
|
|
assert "ACTIVITY_INPUT_CONSUMER_MISMATCH" not in {
|
|
row["code"] for row in summary["validation"]["hardViolations"]
|
|
}
|
|
|
|
|
|
def test_database_projection_preserves_consumer_graph(tmp_path, monkeypatch) -> None:
|
|
monkeypatch.setenv("APS_DB_PATH", str(tmp_path / "consumer-projection.db"))
|
|
monkeypatch.delenv("APS_DB_DISABLED", raising=False)
|
|
from server.db.database import reset_engine
|
|
from server.db.sync import db_to_world, world_to_db
|
|
|
|
reset_engine()
|
|
try:
|
|
world_to_db(_multi_step_bom_world(), project_code="consumer-regression")
|
|
projected: dict = {}
|
|
db_to_world(projected, project_code="consumer-regression")
|
|
closed_loop = build_closed_loop_problem(
|
|
projected, business_date=BUSINESS_DATE, order_nos=["SO-MULTI"], strict=True
|
|
)
|
|
_assert_consumer_alignment(closed_loop_to_problem_v2(projected, closed_loop))
|
|
finally:
|
|
reset_engine()
|