332 lines
13 KiB
Python
332 lines
13 KiB
Python
# ============================================================
|
|
# EX-05 / EX-09 MES 下发与报工黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from server.agent_core import harness
|
|
from server.aps_domain.flex import run_flex_schedule
|
|
from server.aps_domain.mes import (
|
|
apply_dispatch,
|
|
apply_report,
|
|
list_execution,
|
|
preview_dispatch,
|
|
validate_dispatchable_version,
|
|
)
|
|
from server.engines import get_engine
|
|
from server.engines.base import EngineParams
|
|
from server.integrations.mes_stub import reset_mes_client
|
|
from server.state.checkpoints import CheckpointStore
|
|
from server.state.seed import seed_world
|
|
from server.timeutil import add_minutes, fmt_date, today0
|
|
|
|
|
|
class _MemStore:
|
|
def __init__(self, data, checkpoint_path: Path):
|
|
self.data = data
|
|
self.checkpoints = CheckpointStore(str(checkpoint_path))
|
|
|
|
def next_id(self, kind: str) -> int:
|
|
key = f"_c_{kind}"
|
|
self.data[key] = self.data.get(key, 4000) + 1
|
|
return self.data[key]
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
def _approved_grant(
|
|
track: str,
|
|
version_id: int,
|
|
evidence_refs: list[str],
|
|
) -> tuple[str, str]:
|
|
from server.auth.context import IdentityContext, bind_identity, reset_identity
|
|
|
|
block = harness.stage_confirmation(
|
|
"test-mes",
|
|
"mes.dispatch",
|
|
{"track": track, "versionId": version_id, "evidenceRefs": evidence_refs},
|
|
title="MES 下发",
|
|
summary_lines=["测试"],
|
|
)
|
|
confirm_id = str(block.props["confirmId"])
|
|
first = harness.take_confirmation(confirm_id, approve=True)
|
|
token = bind_identity(IdentityContext(
|
|
2002, "approver-2", "Approver 2", "platform", roles=("planner",),
|
|
))
|
|
try:
|
|
second = harness.take_confirmation(confirm_id, approve=True)
|
|
finally:
|
|
reset_identity(token)
|
|
assert first and first["needsSecondConfirm"] is True
|
|
assert second and second["executionGrant"]
|
|
return confirm_id, str(second["executionGrant"])
|
|
|
|
|
|
def _dispatch_evidence_refs(world: dict, track: str, version_id: int) -> list[str]:
|
|
validation = validate_dispatchable_version(world, track, version_id)
|
|
assert validation["dispatchReady"] is True
|
|
return [f"schedule-version:{version_id}", str(validation["evidenceRef"])]
|
|
|
|
|
|
def _dispatch(
|
|
store: _MemStore,
|
|
*,
|
|
track: str = "flex",
|
|
version_id: int | None = None,
|
|
actor: str = "test",
|
|
):
|
|
versions = store.data["flexScheduleVersions"] if track == "flex" else store.data["scheduleVersions"]
|
|
version_id = int(version_id if version_id is not None else versions[-1]["id"])
|
|
evidence_refs = _dispatch_evidence_refs(store.data, track, version_id)
|
|
confirm_id, grant = _approved_grant(track, version_id, evidence_refs)
|
|
checkpoint = store.checkpoints.create(
|
|
store.data,
|
|
label="MES 下发前基线",
|
|
reason="auto:mes.dispatch",
|
|
)
|
|
return apply_dispatch(
|
|
store,
|
|
track=track,
|
|
actor=actor,
|
|
confirm_id=confirm_id,
|
|
execution_grant=grant,
|
|
version_id=version_id,
|
|
before_snapshot=str(checkpoint["pairId"]),
|
|
evidence_refs=evidence_refs,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
|
|
|
|
def test_dispatch_idempotent(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
version = store.data["flexScheduleVersions"][-1]
|
|
version["status"] = "PUBLISHED"
|
|
preview = preview_dispatch(store.data, "flex")
|
|
assert preview["newCount"] >= 1
|
|
|
|
first = _dispatch(store)
|
|
assert first["created"]
|
|
assert len(first["duplicates"]) == 0
|
|
work_order = next(item for item in store.data["flexWorkOrders"] if item.get("mesExternalId"))
|
|
assert work_order["mesExternalId"].startswith("MES-WO-")
|
|
assert version["status"] == "DISPATCHED"
|
|
|
|
second = _dispatch(store)
|
|
assert len(second["created"]) == 0
|
|
assert len(second["duplicates"]) >= len(first["created"])
|
|
audit = store.data["auditEvents"][-1]
|
|
assert store.checkpoints.get(audit["beforeSnapshot"]) is not None
|
|
assert audit["evidenceRefs"][0] == f"schedule-version:{version['id']}"
|
|
assert audit["evidenceRefs"][1].startswith("schedule-evidence:")
|
|
assert audit["rationale"]["confirmId"]
|
|
|
|
|
|
def test_dispatch_requires_gate_snapshot_and_evidence(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
version_id = store.data["flexScheduleVersions"][-1]["id"]
|
|
store.data["flexScheduleVersions"][-1]["status"] = "PUBLISHED"
|
|
evidence_refs = _dispatch_evidence_refs(store.data, "flex", version_id)
|
|
checkpoint = store.checkpoints.create(store.data, label="valid", reason="test")
|
|
|
|
with pytest.raises(PermissionError, match="执行凭据"):
|
|
apply_dispatch(
|
|
store,
|
|
track="flex",
|
|
actor="test",
|
|
confirm_id="confirm-test",
|
|
execution_grant="",
|
|
version_id=version_id,
|
|
before_snapshot=str(checkpoint["pairId"]),
|
|
evidence_refs=evidence_refs,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
|
|
confirm_id, grant = _approved_grant("flex", version_id, evidence_refs)
|
|
with pytest.raises(PermissionError, match="前置快照不存在"):
|
|
apply_dispatch(
|
|
store,
|
|
track="flex",
|
|
actor="test",
|
|
confirm_id=confirm_id,
|
|
execution_grant=grant,
|
|
version_id=version_id,
|
|
before_snapshot="missing-pair",
|
|
evidence_refs=evidence_refs,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
|
|
|
|
def test_report_progress_and_order_complete(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
store.data["flexScheduleVersions"][-1]["status"] = "PUBLISHED"
|
|
_dispatch(store)
|
|
execution = list_execution(store.data, "flex")
|
|
assert execution["total"] >= 1
|
|
row = execution["rows"][0]
|
|
result = apply_report(store, row["woId"], track="flex", progress_pct=50, actor="test")
|
|
assert result["progressPct"] == 50
|
|
work_order = next(item for item in store.data["flexWorkOrders"] if item["id"] == row["woId"])
|
|
assert work_order["status"] == "RUNNING"
|
|
|
|
order_no = row["orderNo"]
|
|
version_id = store.data["flexScheduleVersions"][-1]["id"]
|
|
siblings = [
|
|
item for item in store.data["flexWorkOrders"]
|
|
if item.get("flexOrderNo") == order_no and item.get("versionId") == version_id
|
|
]
|
|
for sibling in siblings:
|
|
apply_report(store, sibling["id"], track="flex", finish=True, actor="test")
|
|
flex_order = next(item for item in store.data["flexOrders"] if item["orderNo"] == order_no)
|
|
assert flex_order["status"] == "COMPLETED"
|
|
|
|
|
|
def test_dispatch_uses_the_version_frozen_at_confirmation(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
approved_version = store.data["flexScheduleVersions"][-1]["id"]
|
|
store.data["flexScheduleVersions"][-1]["status"] = "PUBLISHED"
|
|
evidence_refs = _dispatch_evidence_refs(store.data, "flex", approved_version)
|
|
confirm_id, grant = _approved_grant("flex", approved_version, evidence_refs)
|
|
checkpoint = store.checkpoints.create(store.data, label="frozen", reason="test")
|
|
|
|
run_flex_schedule(store, sort_mode="FORWARD", actor="test")
|
|
assert store.data["flexScheduleVersions"][-1]["id"] != approved_version
|
|
result = apply_dispatch(
|
|
store,
|
|
track="flex",
|
|
actor="test",
|
|
confirm_id=confirm_id,
|
|
execution_grant=grant,
|
|
version_id=approved_version,
|
|
before_snapshot=str(checkpoint["pairId"]),
|
|
evidence_refs=evidence_refs,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
|
|
assert result["created"]
|
|
assert {link["versionId"] for link in store.data["mesLinks"]} == {approved_version}
|
|
assert store.data["auditEvents"][-1]["evidenceRefs"] == evidence_refs
|
|
|
|
|
|
def test_dispatch_grant_rejects_substituted_version_and_evidence(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
|
run_flex_schedule(store, sort_mode="BOTTLENECK", actor="test")
|
|
approved_version = store.data["flexScheduleVersions"][-1]["id"]
|
|
store.data["flexScheduleVersions"][-1]["status"] = "PUBLISHED"
|
|
approved_evidence = _dispatch_evidence_refs(store.data, "flex", approved_version)
|
|
confirm_id, grant = _approved_grant("flex", approved_version, approved_evidence)
|
|
checkpoint = store.checkpoints.create(store.data, label="frozen", reason="test")
|
|
|
|
run_flex_schedule(store, sort_mode="FORWARD", actor="test")
|
|
substituted_version = store.data["flexScheduleVersions"][-1]["id"]
|
|
store.data["flexScheduleVersions"][-1]["status"] = "PUBLISHED"
|
|
substituted_evidence = _dispatch_evidence_refs(store.data, "flex", substituted_version)
|
|
|
|
with pytest.raises(PermissionError, match="漂移"):
|
|
apply_dispatch(
|
|
store,
|
|
track="flex",
|
|
actor="test",
|
|
confirm_id=confirm_id,
|
|
execution_grant=grant,
|
|
version_id=substituted_version,
|
|
before_snapshot=str(checkpoint["pairId"]),
|
|
evidence_refs=substituted_evidence,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
|
|
assert store.data.get("mesLinks") in (None, [])
|
|
|
|
approved = apply_dispatch(
|
|
store,
|
|
track="flex",
|
|
actor="test",
|
|
confirm_id=confirm_id,
|
|
execution_grant=grant,
|
|
version_id=approved_version,
|
|
before_snapshot=str(checkpoint["pairId"]),
|
|
evidence_refs=approved_evidence,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
assert approved["created"]
|
|
assert {link["versionId"] for link in store.data["mesLinks"]} == {approved_version}
|
|
|
|
|
|
def test_fixed_dispatch_resolves_work_orders_through_production_orders(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store = _MemStore(seed_world(), tmp_path / "checkpoints.json")
|
|
params = EngineParams(
|
|
orderIds=[],
|
|
engineType="RULE",
|
|
strategyTemplate="COMPREHENSIVE",
|
|
planningHorizonDays=14,
|
|
startDate=fmt_date(add_minutes(today0(), 24 * 60)),
|
|
)
|
|
result = get_engine("RULE").solve(store.data, params, store.next_id)
|
|
version = next(v for v in store.data["scheduleVersions"] if v["id"] == result.versionId)
|
|
version["status"] = "PUBLISHED"
|
|
for conflict in store.data.get("conflicts") or []:
|
|
if conflict.get("versionId") == result.versionId:
|
|
conflict["isResolved"] = True
|
|
|
|
preview = preview_dispatch(store.data, "fixed")
|
|
assert preview["versionId"] == result.versionId
|
|
assert len(preview["items"]) == result.woCount
|
|
dispatched = _dispatch(store, track="fixed", version_id=result.versionId)
|
|
assert len(dispatched["created"]) == result.woCount
|
|
execution = list_execution(store.data, "fixed")
|
|
assert execution["total"] == result.woCount
|
|
|
|
|
|
def test_fixed_dispatch_executes_all_work_orders_beyond_preview_limit(tmp_path: Path):
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
world = seed_world()
|
|
version_id = 9001
|
|
production_order_id = 9101
|
|
world["scheduleVersions"] = [{
|
|
"id": version_id,
|
|
"versionNo": "SV-9001",
|
|
"status": "PUBLISHED",
|
|
}]
|
|
world["productionOrders"] = [{
|
|
"id": production_order_id,
|
|
"productionOrderNo": "PO-9001",
|
|
"schedulingVersionId": version_id,
|
|
}]
|
|
world["workOrders"] = [{
|
|
"id": 1000 + index,
|
|
"productionOrderId": production_order_id,
|
|
"productionOrderNo": "PO-9001",
|
|
"operationCode": f"OP-{index:03d}",
|
|
"status": "PENDING",
|
|
} for index in range(81)]
|
|
store = _MemStore(world, tmp_path / "checkpoints.json")
|
|
|
|
preview = preview_dispatch(store.data, "fixed", version_id=version_id)
|
|
assert len(preview["items"]) == 80
|
|
assert preview["totalCount"] == 81
|
|
assert preview["newCount"] == 81
|
|
assert preview["truncated"] is True
|
|
|
|
dispatched = _dispatch(store, track="fixed", version_id=version_id)
|
|
assert len(dispatched["created"]) == 81
|
|
assert dispatched["fullyDispatched"] is True
|
|
assert len(store.data["mesLinks"]) == 81
|
|
assert all(work_order.get("mesExternalId") for work_order in store.data["workOrders"])
|
|
assert world["scheduleVersions"][0]["mesDispatched"] is True
|
|
|
|
after = preview_dispatch(store.data, "fixed", version_id=version_id)
|
|
assert after["newCount"] == 0
|