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)