432 lines
18 KiB
Python
432 lines
18 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from server.agent_core import harness
|
|
from server.agent_core.approval_store import ApprovalStore
|
|
from server.aps_domain.closed_loop_problem import build_closed_loop_problem
|
|
from server.aps_domain.flex import run_flex_schedule
|
|
from server.aps_domain.kangni_intake import (
|
|
apply_site_payload_to_world,
|
|
build_site_payload_from_data_dir,
|
|
)
|
|
from server.aps_domain.mes import apply_dispatch, validate_dispatchable_version
|
|
from server.integrations.mes_stub import get_mes_client, reset_mes_client
|
|
from server.state.checkpoints import CheckpointStore
|
|
from server.timeutil import parse_dt
|
|
from tests.golden.test_mes import _approved_grant, _dispatch_evidence_refs
|
|
from server.auth.context import IdentityContext
|
|
from tests.golden.test_folder_schedule_security import (
|
|
_CheckpointStore,
|
|
_ProjectStore,
|
|
_Store,
|
|
_approve,
|
|
_stage,
|
|
)
|
|
from tests.external_data import external_dir
|
|
|
|
KANGNI_DATA_DIR = external_dir("KANGNI_DATA_DIR", "kangni")
|
|
SOURCE_WORKBOOKS = (
|
|
"订单.xlsx",
|
|
"工艺路线.xlsx",
|
|
"工时.xlsx",
|
|
"BOM.xlsx",
|
|
"设备.xlsx",
|
|
"模具.xlsx",
|
|
"物料.xlsx",
|
|
)
|
|
EQUIP_HEADER = [
|
|
"设备编号", "设备名称", "可执行工序编号", "单件工时", "可动率",
|
|
"是否可移动", "移动耗时(分钟)", "区域编码", "适配模具编号", "状态", "备注",
|
|
]
|
|
MOLD_HEADER = [
|
|
"模具编号", "模具名称", "适用工序编号", "适配设备编号",
|
|
"寿命上限", "已用寿命", "区域编码", "换型耗时(分钟)", "状态", "备注",
|
|
]
|
|
|
|
|
|
def _sha256(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def _source_hashes() -> dict[str, str]:
|
|
return {name: _sha256(KANGNI_DATA_DIR / name) for name in SOURCE_WORKBOOKS}
|
|
|
|
|
|
def _assert_no_equipment_overlap(work_orders: list[dict[str, Any]]) -> None:
|
|
by_eq: dict[str, list[tuple[Any, Any, Any]]] = {}
|
|
for wo in work_orders:
|
|
key = str(wo.get("equipmentCode") or wo.get("equipmentId") or "")
|
|
start = parse_dt(wo["plannedStartTime"])
|
|
end = parse_dt(wo["plannedEndTime"])
|
|
assert end > start, f"工单时间非法 {wo.get('id')}: {start} >= {end}"
|
|
by_eq.setdefault(key, []).append((start, end, wo.get("id")))
|
|
for eq_id, ivs in by_eq.items():
|
|
ivs.sort()
|
|
for (_s1, e1, first_id), (s2, _e2, second_id) in zip(ivs, ivs[1:]):
|
|
assert e1 <= s2, f"设备 {eq_id} 双占：{first_id} {e1} > {second_id} {s2}"
|
|
|
|
|
|
def _assert_landing_schedule_quality(work_orders: list[dict[str, Any]]) -> None:
|
|
assert work_orders
|
|
codes = {str(row.get("equipmentCode") or "") for row in work_orders}
|
|
assert codes <= {"WS-LANDING-01", "WS-LANDING-02"}
|
|
window_start = parse_dt("2026-01-01 08:00")
|
|
window_end = parse_dt("2026-09-18 23:59")
|
|
for wo in work_orders:
|
|
start = parse_dt(wo["plannedStartTime"])
|
|
end = parse_dt(wo["plannedEndTime"])
|
|
assert window_start <= start < end <= window_end, wo
|
|
assert start.weekday() < 5
|
|
assert start.hour >= 8
|
|
_assert_no_equipment_overlap(work_orders)
|
|
|
|
|
|
def _real_operation_codes() -> list[str]:
|
|
from openpyxl import load_workbook
|
|
|
|
workbook = load_workbook(KANGNI_DATA_DIR / "工时.xlsx", data_only=True, read_only=True)
|
|
codes: set[str] = set()
|
|
try:
|
|
for sheet_name in workbook.sheetnames:
|
|
sheet = workbook[sheet_name]
|
|
for row in sheet.iter_rows(min_row=2, values_only=True):
|
|
code = str(row[1]).strip() if row and row[1] is not None else ""
|
|
if code:
|
|
codes.add(code)
|
|
finally:
|
|
workbook.close()
|
|
return sorted(codes)
|
|
|
|
|
|
def _write_landing_maps(root: Path, operation_codes: list[str]) -> None:
|
|
from openpyxl import Workbook
|
|
|
|
caps = ",".join(operation_codes)
|
|
std = ";".join(f"{code}:5" for code in operation_codes)
|
|
equip = Workbook()
|
|
sheet = equip.active
|
|
sheet.title = "设备能力"
|
|
sheet.append(EQUIP_HEADER)
|
|
sheet.append([
|
|
"WS-LANDING-01", "城轨机构装配工位1", caps, std, 0.95, "否", 0,
|
|
"ZONE-CG", "", "RUNNING", "落地测试工位，非现场确认设备",
|
|
])
|
|
sheet.append([
|
|
"WS-LANDING-02", "城轨机构装配工位2", caps, std, 0.95, "否", 0,
|
|
"ZONE-CG", "", "RUNNING", "落地测试工位，非现场确认设备",
|
|
])
|
|
equip.save(root / "设备能力映射模板.xlsx")
|
|
|
|
mold = Workbook()
|
|
sheet = mold.active
|
|
sheet.title = "模具适配"
|
|
sheet.append(MOLD_HEADER)
|
|
mold.save(root / "模具适配映射模板.xlsx")
|
|
|
|
|
|
@pytest.fixture
|
|
def landing_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
|
|
missing = [name for name in SOURCE_WORKBOOKS if not (KANGNI_DATA_DIR / name).is_file()]
|
|
if missing:
|
|
pytest.skip(f"康尼现场只读数据不完整：{missing}")
|
|
|
|
source_hashes = _source_hashes()
|
|
upload_dir = tmp_path / "uploaded"
|
|
upload_dir.mkdir()
|
|
for name in SOURCE_WORKBOOKS:
|
|
(upload_dir / name).write_bytes((KANGNI_DATA_DIR / name).read_bytes())
|
|
_write_landing_maps(upload_dir, _real_operation_codes())
|
|
|
|
tenant = "tenant-kangni-landing"
|
|
identity = IdentityContext(
|
|
user_id=2201,
|
|
username="planner",
|
|
fullname="Planner",
|
|
tenant_uuid=tenant,
|
|
roles=("planner", "approver", "admin"),
|
|
)
|
|
project_store = _ProjectStore(upload_dir)
|
|
checkpoints = _CheckpointStore()
|
|
store = _Store(tenant)
|
|
approval_store = ApprovalStore(str(tmp_path / "approvals-landing.json"))
|
|
original_approval_store = harness._approval_store
|
|
harness.configure_approval_store(store=approval_store)
|
|
|
|
import server.aps_domain.workflow as workflow_module
|
|
from server.agent_core import plan_orchestration
|
|
|
|
monkeypatch.setattr("server.state.projects.get_project_store", lambda: project_store)
|
|
monkeypatch.setattr(workflow_module, "get_checkpoints", lambda: checkpoints)
|
|
monkeypatch.setattr(plan_orchestration, "stage_plan_node", lambda **_kwargs: None)
|
|
monkeypatch.setattr(plan_orchestration, "decide_plan_node", lambda **_kwargs: None)
|
|
try:
|
|
yield {
|
|
"store": store,
|
|
"identity": identity,
|
|
"approvals": approval_store,
|
|
"uploadDir": upload_dir,
|
|
"sourceHashes": source_hashes,
|
|
}
|
|
finally:
|
|
harness.configure_approval_store(store=original_approval_store)
|
|
|
|
|
|
def test_landing_mappings_make_production_ready(landing_env: dict[str, Any]):
|
|
payload = build_site_payload_from_data_dir(landing_env["uploadDir"])
|
|
quality = payload["meta"]["resourceQuality"]
|
|
equipment = payload["flex"]["flexEquipment"]
|
|
assert quality["productionReady"] is True
|
|
assert quality["sharedPlaceholder"] is False
|
|
assert quality["equipmentCount"] == 2
|
|
assert quality["equipmentWithoutCapability"] == 0
|
|
assert quality["moldCount"] == 0
|
|
assert [row["code"] for row in equipment] == ["WS-LANDING-01", "WS-LANDING-02"]
|
|
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True)
|
|
assert "EQ-SHARED-CG-01" not in encoded
|
|
assert str(KANGNI_DATA_DIR) not in encoded
|
|
assert _source_hashes() == landing_env["sourceHashes"]
|
|
|
|
|
|
def test_customer_source_dir_is_not_field_production_ready(landing_env: dict[str, Any]):
|
|
payload = build_site_payload_from_data_dir(KANGNI_DATA_DIR)
|
|
quality = payload["meta"]["resourceQuality"]
|
|
material_header = [
|
|
"序号", "物料代码", "物料名称", "物料描述", "物料英文描述", "ABC分类", "单位",
|
|
"次要单位", "图号", "采购类型", "特殊采购类型", "规格型号", "是否关键件",
|
|
"是否批次管理", "仓库编号", "仓库名称", "最小包装数量", "容器包装规格",
|
|
"厂内拉动类型", "物料组", "物料类型", "生成管理员", "MRP控制者",
|
|
"是否为序列号管理", "旧图号", "齐套检查", "工厂编号", "工厂名称", "原因",
|
|
]
|
|
from openpyxl import load_workbook
|
|
workbook = load_workbook(KANGNI_DATA_DIR / "物料.xlsx", data_only=True, read_only=True)
|
|
try:
|
|
header = [str(cell).strip() if cell is not None else "" for cell in next(workbook.active.iter_rows(min_row=1, max_row=1, values_only=True))]
|
|
finally:
|
|
workbook.close()
|
|
|
|
assert header == material_header
|
|
assert "库存" not in "".join(header)
|
|
assert "在途" not in "".join(header)
|
|
assert quality["productionReady"] is False
|
|
assert quality["sharedPlaceholder"] is True
|
|
assert quality["equipmentCount"] == 41
|
|
assert quality["equipmentWithoutCapability"] == 40
|
|
assert quality["moldCount"] == 40
|
|
assert quality["moldsWithoutOperation"] == 40
|
|
assert quality["moldsWithoutEquipment"] == 40
|
|
assert quality["moldsWithoutLifeTotal"] == 40
|
|
assert _source_hashes() == landing_env["sourceHashes"]
|
|
|
|
|
|
def test_landing_folder_trial_marks_production_ready_without_mes(
|
|
landing_env: dict[str, Any],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
):
|
|
from server.aps_domain import mes
|
|
|
|
store = landing_env["store"]
|
|
identity = landing_env["identity"]
|
|
approvals = landing_env["approvals"]
|
|
|
|
def fail_mes(*_args, **_kwargs):
|
|
raise AssertionError("production landing folder.schedule must not dispatch MES")
|
|
|
|
monkeypatch.setattr(mes, "apply_dispatch", fail_mes)
|
|
confirm_id, reply = _stage(store, identity)
|
|
params = approvals.pending[confirm_id]["params"]
|
|
folder_block = next(block for block in reply.blocks if block.type == "folder-pack")
|
|
|
|
assert params["productionReady"] is True
|
|
assert params["trialReady"] is True
|
|
assert folder_block.props["productionReady"] is True
|
|
assert "productionReady=true" in reply.text
|
|
assert "共享占位" not in reply.text
|
|
assert "不得生产发布或下发 MES" not in reply.text
|
|
|
|
message = _approve(store, identity, confirm_id)
|
|
version = store.data["flexScheduleVersions"][-1]
|
|
version_id = version["id"]
|
|
assert version["productionReady"] is True
|
|
assert version["trialOnly"] is True
|
|
assert version["resourceQuality"]["sharedPlaceholder"] is False
|
|
assert len(store.data.get("salesOrders") or []) == 10
|
|
assert len(store.data.get("flexRoutings") or []) == 72
|
|
assert [row["code"] for row in store.data.get("flexEquipment") or []] == [
|
|
"WS-LANDING-01",
|
|
"WS-LANDING-02",
|
|
]
|
|
assert len([row for row in store.data["flexVirtualLines"] if row["versionId"] == version_id]) == 10
|
|
assert len([row for row in store.data["flexWorkOrders"] if row["versionId"] == version_id]) == 72
|
|
assert "productionReady=true" in message
|
|
assert "不得生产发布或下发 MES" not in message
|
|
actions = {str(row.get("action") or "") for row in store.data.get("auditEvents") or []}
|
|
assert "mes.dispatch" not in actions
|
|
assert "schedule.publish" not in actions
|
|
assert _source_hashes() == landing_env["sourceHashes"]
|
|
|
|
|
|
|
|
|
|
def _prepare_formal_landing_world(world: dict[str, Any], operation_codes: list[str]) -> None:
|
|
for row in world.get("salesOrders") or []:
|
|
row["source"] = "SITE"
|
|
for table in ("materials", "flexMaterials"):
|
|
for row in world.get(table) or []:
|
|
row["stock"] = 0
|
|
row["inTransit"] = 0
|
|
row["safetyStock"] = 0
|
|
for row in world.get("flexRoutings") or []:
|
|
row["stdTimePerUnit"] = 5
|
|
world["inventoryBalances"] = []
|
|
world["inventories"] = []
|
|
world["businessDate"] = "2026-01-01"
|
|
world.setdefault("flexParams", {})["afterDays"] = 260
|
|
world["flexCalendar"] = [{
|
|
"startTime": "08:00",
|
|
"endTime": "23:59",
|
|
"breaks": [],
|
|
"workdays": [1, 2, 3, 4, 5],
|
|
}]
|
|
for key in ("factories", "workshops", "lines", "workstations"):
|
|
for row in world.get(key) or []:
|
|
row["capacityMinutesPerDay"] = 20000
|
|
row["status"] = "ACTIVE"
|
|
for row in world.get("flexTeams") or []:
|
|
row["status"] = "ACTIVE"
|
|
row["memberCount"] = max(int(row.get("memberCount") or 1), 8)
|
|
row["supportOps"] = list(operation_codes)
|
|
|
|
|
|
def _inject_confirmed_purchase_orders(world: dict[str, Any]) -> int:
|
|
problem = build_closed_loop_problem(world, business_date=str(world["businessDate"])[:10])
|
|
needed: dict[tuple[Any, Any], dict[str, Any]] = {}
|
|
for req in problem.requirements:
|
|
if req.sourcing_type != "BUY":
|
|
continue
|
|
sales_order_id: Any = int(req.sales_order_id) if str(req.sales_order_id).isdigit() else req.sales_order_id
|
|
key = (sales_order_id, req.material_id)
|
|
info = needed.setdefault(key, {"qty": 0.0, "code": req.material_code})
|
|
info["qty"] += float(req.quantity or 0)
|
|
info["code"] = req.material_code
|
|
world["purchaseOrders"] = [
|
|
{
|
|
"id": 80_000 + index,
|
|
"orderNo": f"KN-LANDING-PO-{index:04d}",
|
|
"salesOrderId": sales_order_id,
|
|
"materialId": material_id,
|
|
"materialCode": info["code"],
|
|
"quantity": info["qty"],
|
|
"expectedDate": "2025-12-01",
|
|
"status": "CONFIRMED",
|
|
"note": "landing-test trusted supply, not customer snapshot",
|
|
}
|
|
for index, ((sales_order_id, material_id), info) in enumerate(
|
|
sorted(needed.items(), key=lambda item: (str(item[0][0]), str(item[0][1]))),
|
|
start=1,
|
|
)
|
|
]
|
|
return len(world["purchaseOrders"])
|
|
|
|
def test_formal_closed_loop_still_blocks_on_missing_trusted_supply(
|
|
landing_env: dict[str, Any],
|
|
):
|
|
from server.aps_domain.kangni_intake import apply_site_payload_to_world
|
|
|
|
payload = build_site_payload_from_data_dir(landing_env["uploadDir"])
|
|
store = landing_env["store"]
|
|
apply_site_payload_to_world(store.data, payload, clear_all=True)
|
|
result = run_flex_schedule(store, sort_mode="BOTTLENECK", actor="landing", trial=False)
|
|
summary = ((result.get("planning") or {}).get("summary") or {})
|
|
version = store.data["flexScheduleVersions"][-1]
|
|
validation = validate_dispatchable_version(store.data, "flex", int(version["id"]))
|
|
|
|
assert payload["meta"]["resourceQuality"]["productionReady"] is True
|
|
assert result.get("solveStatus") == "BLOCKED"
|
|
assert int(result.get("woCount") or 0) == 0
|
|
assert summary.get("blockerCounts", {}).get("SUPPLY_SHORTAGE") == 1646
|
|
assert validation["publishReady"] is False
|
|
assert validation["dispatchReady"] is False
|
|
assert any(
|
|
str(reason.get("code")) == "V2_SOLUTION_NOT_FEASIBLE"
|
|
for reason in (validation.get("publishBlockingReasons") or [])
|
|
)
|
|
assert _source_hashes() == landing_env["sourceHashes"]
|
|
|
|
|
|
def test_formal_closed_loop_with_trusted_supply_dispatches_mes_stub(
|
|
landing_env: dict[str, Any],
|
|
tmp_path: Path,
|
|
):
|
|
payload = build_site_payload_from_data_dir(landing_env["uploadDir"])
|
|
store = landing_env["store"]
|
|
apply_site_payload_to_world(store.data, payload, clear_all=True)
|
|
_prepare_formal_landing_world(store.data, _real_operation_codes())
|
|
injected = _inject_confirmed_purchase_orders(store.data)
|
|
reset_mes_client(tmp_path / "mes_mirror.json")
|
|
store.checkpoints = CheckpointStore(str(tmp_path / "landing-checkpoints.json"))
|
|
|
|
result = run_flex_schedule(store, sort_mode="BOTTLENECK", actor="landing", trial=False)
|
|
version = store.data["flexScheduleVersions"][-1]
|
|
version_id = int(version["id"])
|
|
work_orders = [row for row in store.data.get("flexWorkOrders") or [] if row.get("versionId") == version_id]
|
|
virtual_lines = [row for row in store.data.get("flexVirtualLines") or [] if row.get("versionId") == version_id]
|
|
|
|
assert payload["meta"]["resourceQuality"]["productionReady"] is True
|
|
assert injected == 823
|
|
assert result.get("executionMode") == "CLOSED_LOOP_V1"
|
|
assert result.get("solveStatus") == "FEASIBLE"
|
|
assert len(store.data.get("salesOrders") or []) == 10
|
|
assert len(virtual_lines) == 10
|
|
assert len(work_orders) == 72
|
|
assert [row["code"] for row in store.data.get("flexEquipment") or []] == [
|
|
"WS-LANDING-01",
|
|
"WS-LANDING-02",
|
|
]
|
|
assert all(str(row.get("orderNo") or "").startswith("KN-LANDING-PO-") for row in store.data.get("purchaseOrders") or [])
|
|
_assert_landing_schedule_quality(work_orders)
|
|
|
|
draft_gate = validate_dispatchable_version(store.data, "flex", version_id)
|
|
assert draft_gate["publishReady"] is True
|
|
assert draft_gate["dispatchReady"] is False
|
|
version["status"] = "PUBLISHED"
|
|
validation = validate_dispatchable_version(store.data, "flex", version_id)
|
|
assert validation["publishReady"] is False
|
|
assert validation["dispatchReady"] is True
|
|
|
|
evidence_refs = _dispatch_evidence_refs(store.data, "flex", version_id)
|
|
confirm_id, grant = _approved_grant("flex", version_id, evidence_refs)
|
|
checkpoint = store.checkpoints.create(store.data, label="康尼落地 MES 桩", reason="auto:mes.dispatch")
|
|
dispatched = apply_dispatch(
|
|
store,
|
|
track="flex",
|
|
actor="landing",
|
|
confirm_id=confirm_id,
|
|
execution_grant=grant,
|
|
version_id=version_id,
|
|
before_snapshot=str(checkpoint["pairId"]),
|
|
evidence_refs=evidence_refs,
|
|
checkpoint_store=store.checkpoints,
|
|
)
|
|
mes_ids = [row.get("mesExternalId") for row in store.data.get("flexWorkOrders") or [] if row.get("mesExternalId")]
|
|
mes_status = get_mes_client().status()
|
|
|
|
assert dispatched["created"]
|
|
assert len(mes_ids) == 72
|
|
assert all(str(item).startswith("MES-WO-") for item in mes_ids)
|
|
assert version["status"] == "DISPATCHED"
|
|
assert mes_status.get("mode") == "stub"
|
|
assert int(mes_status.get("woCount") or 0) == 72
|
|
assert "http" not in json.dumps(mes_status, ensure_ascii=False).lower()
|
|
mirror = json.loads((tmp_path / "mes_mirror.json").read_text(encoding="utf-8"))
|
|
assert len(mirror.get("workOrders") or []) == 72
|
|
assert all(str(row.get("id") or "").startswith("MES-WO-") for row in mirror.get("workOrders") or [])
|
|
assert _source_hashes() == landing_env["sourceHashes"]
|
|
assert str(KANGNI_DATA_DIR) not in json.dumps(payload, ensure_ascii=False)
|
|
|