aps-agent/tests/golden/test_flex_atomic_schedule.py

from __future__ import annotations

import copy

import pytest

from server.engines.pool_engine import PoolEngine
from server.state.seed import build_demo_world

BUSINESS_DATE = "2026-08-02"


def _next_id_factory():
    counters: dict[str, int] = {}

    def next_id(kind: str) -> int:
        counters[kind] = counters.get(kind, 0) + 1
        return counters[kind]

    return next_id


def _single_order_world() -> tuple[dict, dict]:
    world = build_demo_world()
    order = next(item for item in world["flexOrders"] if item["productCode"] == "HV-HARNESS")
    world["flexOrders"] = [copy.deepcopy(order)]
    return world, world["flexOrders"][0]


def _solve(world: dict, *, enforce_teams: bool | None = None) -> dict:
    return PoolEngine().solve(
        world,
        _next_id_factory(),
        sort_mode="ASC",
        start_date=BUSINESS_DATE,
        enforce_teams=enforce_teams,
    )


@pytest.mark.parametrize(
    ("failure", "expected_conflict"),
    [
        ("routing", "NO_ROUTING"),
        ("capability", "NO_CAPABILITY"),
        ("team", "NO_TEAM"),
        ("mold", "NO_MOLD"),
    ],
)
def test_failed_order_never_leaves_partial_virtual_line_or_work_orders(
    failure: str,
    expected_conflict: str,
):
    world, order = _single_order_world()
    route = sorted(
        [row for row in world["flexRoutings"] if row["productCode"] == order["productCode"]],
        key=lambda row: row["seq"],
    )
    enforce_teams = None
    if failure == "routing":
        world["flexRoutings"] = [
            row for row in world["flexRoutings"] if row["productCode"] != order["productCode"]
        ]
    elif failure == "capability":
        final_operation = route[-1]["operationCode"]
        for equipment in world["flexEquipment"]:
            equipment["capabilities"] = [
                code for code in equipment.get("capabilities", []) if code != final_operation
            ]
    elif failure == "team":
        final_operation = route[-1]["operationCode"]
        world["flexTeams"] = [{
            "code": "TEAM-PARTIAL",
            "name": "Partial team",
            "memberCount": 20,
            "supportOps": [row["operationCode"] for row in route if row["operationCode"] != final_operation],
        }]
        enforce_teams = True
    elif failure == "mold":
        world["flexMolds"] = []

    molds_before = copy.deepcopy(world["flexMolds"])
    result = _solve(world, enforce_teams=enforce_teams)
    version_id = result["versionId"]
    version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id)
    work_orders = [item for item in world["flexWorkOrders"] if item["versionId"] == version_id]
    virtual_lines = [item for item in world["flexVirtualLines"] if item["versionId"] == version_id]
    conflict_types = {
        item["conflictType"]
        for item in world["flexConflicts"]
        if item["versionId"] == version_id
    }

    assert result["orderCount"] == 1
    assert result["vlCount"] == version["vlCount"] == len(virtual_lines) == 0
    assert result["woCount"] == version["woCount"] == len(work_orders) == 0
    assert expected_conflict in conflict_types
    assert world["flexMolds"] == molds_before


def test_successful_order_commits_one_virtual_line_and_exact_routing_work_orders():
    world, order = _single_order_world()
    expected_route = sorted(
        [
            (row["seq"], row["operationCode"])
            for row in world["flexRoutings"]
            if row["productCode"] == order["productCode"]
        ]
    )

    result = _solve(world)
    version_id = result["versionId"]
    version = next(item for item in world["flexScheduleVersions"] if item["id"] == version_id)
    virtual_lines = [item for item in world["flexVirtualLines"] if item["versionId"] == version_id]
    work_orders = [item for item in world["flexWorkOrders"] if item["versionId"] == version_id]

    assert result["vlCount"] == version["vlCount"] == len(virtual_lines) == 1
    assert result["woCount"] == version["woCount"] == len(work_orders) == len(expected_route)
    assert len({item["id"] for item in work_orders}) == len(work_orders)
    assert {item["vlId"] for item in work_orders} == {virtual_lines[0]["id"]}
    assert sorted((item["seq"], item["operationCode"]) for item in work_orders) == expected_route
    assert sorted(
        (item["seq"], item["operationCode"])
        for item in virtual_lines[0]["assignments"]
    ) == expected_route


def test_failed_candidate_releases_resources_before_next_order_is_scheduled():
    world = build_demo_world()
    failed = copy.deepcopy(next(item for item in world["flexOrders"] if item["productCode"] == "HV-HARNESS"))
    successful = copy.deepcopy(next(item for item in world["flexOrders"] if item["productCode"] == "PDU-UNIT"))
    failed["dueDate"] = "2026-08-03"
    successful["dueDate"] = "2026-08-04"
    world["flexOrders"] = [failed, successful]
    world["flexRoutings"].append({
        "productCode": failed["productCode"],
        "seq": 99,
        "operationCode": "OP-NO-CAPABILITY",
        "stdTimePerUnit": 1,
        "requireMold": False,
    })
    mold_before = copy.deepcopy(world["flexMolds"])

    result = _solve(world)
    version_id = result["versionId"]
    virtual_lines = [item for item in world["flexVirtualLines"] if item["versionId"] == version_id]
    work_orders = [item for item in world["flexWorkOrders"] if item["versionId"] == version_id]

    assert result["orderCount"] == 2
    assert result["vlCount"] == 1
    assert {item["orderNo"] for item in virtual_lines} == {successful["orderNo"]}
    assert all(item["flexOrderNo"] == successful["orderNo"] for item in work_orders)
    assert min(item["plannedStartTime"] for item in work_orders) == "2026-08-03 08:00"
    assert world["flexMolds"] == mold_before