From d199e0cef7b40889864b18a2a8591f49ff3ed54c Mon Sep 17 00:00:00 2001 From: "z.zhang" Date: Wed, 26 Aug 2026 00:26:00 +0800 Subject: [PATCH] =?UTF-8?q?test:=20=E5=BA=B7=E5=B0=BC=E7=9C=9F=E5=AE=9E?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=94=9F=E4=BA=A7=E8=90=BD=E5=9C=B0=E4=B8=8E?= =?UTF-8?q?=20MES=20=E6=A1=A9=E9=97=AD=E7=8E=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 真实订单/工艺/BOM + 落地工位 + 823 张测试采购单，验证 trial=False 正式闭环 FEASIBLE、发布校验和 72 个 MES-WO 桩下发；保留客户目录 productionReady=false 门禁。 --- .../golden/test_kangni_production_landing.py | 431 ++++++++++++++++++ 1 file changed, 431 insertions(+) create mode 100644 tests/golden/test_kangni_production_landing.py diff --git a/tests/golden/test_kangni_production_landing.py b/tests/golden/test_kangni_production_landing.py new file mode 100644 index 0000000..202f7ad --- /dev/null +++ b/tests/golden/test_kangni_production_landing.py @@ -0,0 +1,431 @@ +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, +) + + +KANGNI_DATA_DIR = Path(r"D:\ItemSpace\14.工业智核\康尼\数据") +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) +