698 lines
29 KiB
Python
698 lines
29 KiB
Python
# ============================================================
|
||
# 只读分析 → P2 采用：SQL 数据包 / MOM 主数据表
|
||
# SQL 用可解析的最小数据包；MOM 解析器按契约打桩（本机无专家数据），
|
||
# 门禁、冻结与写入路径全部走真实实现。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import uuid
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from server.state.seed import empty_world
|
||
|
||
MINI_SQL = """
|
||
CREATE TABLE `pl_order` (
|
||
`id` int, `code` varchar(64), `business_code` varchar(64),
|
||
`customer_business_code` varchar(64), `material_code` varchar(64),
|
||
`material_name` varchar(64), `parent_code` varchar(64), `craftl_code` varchar(64),
|
||
`machine_model` varchar(64), `dept` varchar(64), `people` varchar(64),
|
||
`drawing_code` varchar(64), `quantity` decimal(10,2), `man_quantity` decimal(10,2),
|
||
`all_quantity` decimal(10,2), `status` varchar(8), `stock_status` varchar(8),
|
||
`distribution_status` varchar(8), `type` varchar(8), `description` varchar(64),
|
||
`planned_end_time` datetime, `is_delete` varchar(8), `level` int
|
||
) ENGINE=InnoDB;
|
||
CREATE TABLE `r_production_craftl` (
|
||
`id` int, `craftl_code` varchar(64), `craftl_name` varchar(64),
|
||
`procedure_code` varchar(64), `procedure_name` varchar(64),
|
||
`output_material_code` varchar(64), `output_material_name` varchar(64),
|
||
`working_hours` decimal(10,2), `sort_no` int, `is_delete` varchar(8)
|
||
) ENGINE=InnoDB;
|
||
CREATE TABLE `md_equipment` (
|
||
`id` int, `code` varchar(64), `name` varchar(64), `is_delete` varchar(8)
|
||
) ENGINE=InnoDB;
|
||
CREATE TABLE `md_material` (
|
||
`id` int, `code` varchar(64), `name` varchar(64), `type` varchar(32),
|
||
`unit` varchar(8), `is_delete` varchar(8)
|
||
) ENGINE=InnoDB;
|
||
INSERT INTO `pl_order` VALUES
|
||
(1,'ORD1',NULL,'C1','FG001','Product 1',NULL,'CRAFT1',NULL,NULL,NULL,NULL,10,NULL,NULL,'0','0','0','S',NULL,'2026-08-01 00:00:00','0',5);
|
||
INSERT INTO `r_production_craftl` VALUES
|
||
(1,'CRAFT1','Product 1','OP10','Cut','WIP-X','Semi',0.5,10,'0'),
|
||
(2,'CRAFT1','Product 1','OP20','Finish','FG001','Product 1',0.8,20,'0');
|
||
INSERT INTO `md_equipment` VALUES (1,'EQ1','Cutter','0');
|
||
INSERT INTO `md_material` VALUES (1,'FG001','Product 1','Product','pc','0');
|
||
"""
|
||
|
||
|
||
MOM_PACK = {
|
||
"filename": "MOM主数据收集表.xlsx",
|
||
"modelSheetHits": 4,
|
||
"stats": {"materials": 1, "bom": 1, "equipment": 1, "orders": 1,
|
||
"routings": 1, "rootProduct": "MOM-FG"},
|
||
"flexMaterials": [
|
||
{"code": "MOM-FG", "name": "样例成品", "type": "FINISHED_PRODUCT",
|
||
"unit": "件", "stock": 5},
|
||
],
|
||
"flexBom": [{"productCode": "MOM-FG", "materialCode": "MOM-RM", "quantity": 1}],
|
||
"flexEquipment": [{"code": "MOM-EQ", "name": "示例设备"}],
|
||
"flexOrders": [{"orderNo": "MOM-1", "productCode": "MOM-FG",
|
||
"quantity": 3, "dueDate": "2026-08-30"}],
|
||
"flexRoutings": [{"productCode": "MOM-FG", "seq": 10, "operationCode": "CUT",
|
||
"operationName": "Cut", "stdTimePerUnit": 1.0}],
|
||
}
|
||
|
||
|
||
class _Store:
|
||
def __init__(self) -> None:
|
||
self.data = empty_world()
|
||
self.saved = 0
|
||
self._ids: dict[str, int] = {}
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
self._ids[kind] = self._ids.get(kind, 0) + 1
|
||
return self._ids[kind]
|
||
|
||
def save(self) -> None:
|
||
self.saved += 1
|
||
|
||
|
||
async def _run_analyze(store: _Store, session_id: str):
|
||
from server.aps_domain.workflow import handle_intent
|
||
from server.contracts import IntentResult
|
||
|
||
return await handle_intent(
|
||
store,
|
||
session_id,
|
||
IntentResult(intent="folder.analyze", params={}, confidence=1.0, source="LLM"),
|
||
actor="test",
|
||
)
|
||
|
||
|
||
def _prepare_env(tmp_path: Path, monkeypatch, *, project_id: str, session_id: str):
|
||
from server.auth.context import IdentityContext, bind_identity
|
||
from server.knowledge.assets import KnowledgeStore
|
||
|
||
class FakePS:
|
||
def snapshot(self, include_messages=False):
|
||
return {
|
||
"projects": [{"id": project_id, "name": "采用项目",
|
||
"workDir": str(tmp_path)}],
|
||
"sessions": [{"id": session_id, "projectId": project_id}],
|
||
"files": [],
|
||
}
|
||
|
||
def active_world_key(self) -> str:
|
||
return project_id
|
||
|
||
kb = KnowledgeStore(path=str(tmp_path / "knowledge.json"))
|
||
kb.assets = []
|
||
kb._write()
|
||
monkeypatch.setattr("server.state.projects.get_project_store", lambda: FakePS())
|
||
monkeypatch.setattr("server.knowledge.assets.get_knowledge", lambda: kb)
|
||
monkeypatch.setattr(
|
||
"server.aps_domain.project_analyze._project_ctx_readonly",
|
||
lambda _session_id: (
|
||
{"id": project_id, "name": "采用项目", "workDir": str(tmp_path)},
|
||
str(tmp_path),
|
||
),
|
||
)
|
||
identity = IdentityContext(
|
||
user_id=99,
|
||
username="adopt_owner",
|
||
fullname="采用用户",
|
||
tenant_uuid=f"tenant-adopt-{uuid.uuid4().hex[:10]}",
|
||
roles=("planner",),
|
||
)
|
||
token = bind_identity(identity)
|
||
return _Store(), kb, token
|
||
|
||
|
||
def _adoption_card(reply):
|
||
return next(block for block in reply.blocks if block.type == "confirm-card")
|
||
|
||
|
||
_MASTER_KEYS = ("flexMaterials", "flexOrders")
|
||
|
||
|
||
def _seed_master_data(store: _Store) -> dict[str, str]:
|
||
"""既有业务主数据（采用整表替换时绝不允许被清空）。"""
|
||
store.data["flexMaterials"] = [
|
||
{"code": "KEEP-RM", "name": "既有物料", "unit": "件"},
|
||
]
|
||
store.data["flexOrders"] = [
|
||
{"orderNo": "KEEP-1", "productCode": "KEEP-RM", "quantity": 1},
|
||
]
|
||
return _master_snapshot(store)
|
||
|
||
|
||
def _master_snapshot(store: _Store) -> dict[str, str]:
|
||
return {
|
||
key: json.dumps(store.data.get(key), ensure_ascii=False, sort_keys=True, default=str)
|
||
for key in _MASTER_KEYS
|
||
}
|
||
|
||
|
||
def test_sql_analysis_stages_p2_adoption_and_writes_after_approval(tmp_path: Path, monkeypatch):
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
|
||
(tmp_path / "mini.sql").write_text(MINI_SQL, encoding="utf-8")
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-sql", session_id="s-sql",
|
||
)
|
||
try:
|
||
reply = asyncio.run(_run_analyze(store, "s-sql"))
|
||
|
||
# 只读分析：确认前不得写入任何排产数据，但必须给出采用卡。
|
||
assert store.data["flexOrders"] == []
|
||
assert store.data["flexEquipment"] == []
|
||
assert store.data["flexRoutings"] == []
|
||
card = _adoption_card(reply)
|
||
assert card.props["action"] == "import.commit"
|
||
|
||
message = execute_confirmed(store, card.props["confirmId"], True, "approver")
|
||
assert "采用" in message
|
||
assert store.data["flexOrders"]
|
||
assert store.data["flexEquipment"]
|
||
assert store.data["flexRoutings"]
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_mom_analysis_adoption_is_gated_and_fails_closed_on_changed_file(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
|
||
mom_path = tmp_path / "MOM主数据收集表.xlsx"
|
||
mom_path.write_bytes(b"mom-source-v1")
|
||
monkeypatch.setattr("server.importers.mom_pack.is_mom_workbook", lambda _path: True)
|
||
monkeypatch.setattr(
|
||
"server.importers.mom_pack.parse_mom_workbook", lambda _path: dict(MOM_PACK),
|
||
)
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-mom", session_id="s-mom",
|
||
)
|
||
try:
|
||
reply = asyncio.run(_run_analyze(store, "s-mom"))
|
||
assert store.data["flexMaterials"] == []
|
||
assert store.data["flexOrders"] == []
|
||
card = _adoption_card(reply)
|
||
assert card.props["action"] == "import.commit"
|
||
|
||
message = execute_confirmed(store, card.props["confirmId"], True, "approver")
|
||
assert "采用" in message
|
||
assert any(row.get("code") == "MOM-FG" for row in store.data["flexMaterials"])
|
||
|
||
# 出卡后来源文件被改写：批准必须 fail closed，世界保持最后一次采用结果。
|
||
stale = asyncio.run(_run_analyze(store, "s-mom"))
|
||
stale_card = _adoption_card(stale)
|
||
before = json.dumps(store.data.get("flexMaterials"), sort_keys=True, default=str)
|
||
mom_path.write_bytes(b"mom-source-v2")
|
||
with pytest.raises(PermissionError):
|
||
execute_confirmed(store, stale_card.props["confirmId"], True, "approver")
|
||
after = json.dumps(store.data.get("flexMaterials"), sort_keys=True, default=str)
|
||
assert before == after
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_empty_mom_workbook_is_not_adoptable_and_never_clears_master_data(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
"""空表/仅表头（含只是文件名带 MOM 的普通工作簿）不得升级成整表替换的采用卡。"""
|
||
from server.auth.context import reset_identity
|
||
|
||
mom_path = tmp_path / "MOM2026资料.xlsx"
|
||
mom_path.write_bytes(b"empty-mom-template")
|
||
monkeypatch.setattr("server.importers.mom_pack.is_mom_workbook", lambda _path: True)
|
||
monkeypatch.setattr(
|
||
"server.importers.mom_pack.parse_mom_workbook",
|
||
lambda _path: {
|
||
"filename": mom_path.name,
|
||
"modelSheetHits": 2,
|
||
"stats": {"materials": 0, "bom": 0, "equipment": 0, "orders": 0,
|
||
"routings": 0},
|
||
"flexMaterials": [], "flexBom": [], "flexEquipment": [],
|
||
"flexOrders": [], "flexRoutings": [],
|
||
},
|
||
)
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-empty-mom", session_id="s-empty-mom",
|
||
)
|
||
try:
|
||
before = _seed_master_data(store)
|
||
reply = asyncio.run(_run_analyze(store, "s-empty-mom"))
|
||
assert not [block for block in reply.blocks if block.type == "confirm-card"]
|
||
assert _master_snapshot(store) == before
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_sql_adoption_rejects_world_drift_after_card_is_staged(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
"""出卡后目标主数据被改动：批准必须拒绝整包替换（卡片同时被消费，需重新分析）。"""
|
||
from server.agent_core import harness
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
|
||
(tmp_path / "mini.sql").write_text(MINI_SQL, encoding="utf-8")
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-sql-drift", session_id="s-sql-drift",
|
||
)
|
||
try:
|
||
reply = asyncio.run(_run_analyze(store, "s-sql-drift"))
|
||
card = _adoption_card(reply)
|
||
frozen = harness.pending_confirmation(card.props["confirmId"])
|
||
assert frozen["params"].get("targetWorldFingerprint")
|
||
|
||
store.data["flexMaterials"].append({"code": "LATE-EDIT", "name": "审批期间新增"})
|
||
with pytest.raises(PermissionError):
|
||
execute_confirmed(store, card.props["confirmId"], True, "approver")
|
||
|
||
# 拒绝后世界保持拒绝前状态：SQL 冻结载荷没有被写入。
|
||
assert [row.get("code") for row in store.data["flexMaterials"]] == ["LATE-EDIT"]
|
||
assert store.data["flexOrders"] == []
|
||
assert store.data["flexEquipment"] == []
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_mom_adoption_rejects_world_drift_after_card_is_staged(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
from server.agent_core import harness
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
|
||
mom_path = tmp_path / "MOM主数据收集表.xlsx"
|
||
mom_path.write_bytes(b"mom-source-v1")
|
||
monkeypatch.setattr("server.importers.mom_pack.is_mom_workbook", lambda _path: True)
|
||
monkeypatch.setattr(
|
||
"server.importers.mom_pack.parse_mom_workbook", lambda _path: dict(MOM_PACK),
|
||
)
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-mom-drift", session_id="s-mom-drift",
|
||
)
|
||
try:
|
||
before = _seed_master_data(store)
|
||
reply = asyncio.run(_run_analyze(store, "s-mom-drift"))
|
||
card = _adoption_card(reply)
|
||
frozen = harness.pending_confirmation(card.props["confirmId"])
|
||
assert frozen["params"].get("targetWorldFingerprint")
|
||
|
||
store.data["flexOrders"].append(
|
||
{"orderNo": "LATE-EDIT", "productCode": "KEEP-RM", "quantity": 2})
|
||
with pytest.raises(PermissionError):
|
||
execute_confirmed(store, card.props["confirmId"], True, "approver")
|
||
|
||
snapshot = _master_snapshot(store)
|
||
assert snapshot["flexMaterials"] == before["flexMaterials"]
|
||
assert "LATE-EDIT" in snapshot["flexOrders"]
|
||
assert "MOM-FG" not in snapshot["flexMaterials"]
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_zero_row_mom_card_fails_closed_at_execution(tmp_path: Path, monkeypatch):
|
||
"""即使出卡环节被绕过（直接 stage），零业务行的 MOM 卡在批准时也必须拒绝写入。"""
|
||
from server.agent_core import harness
|
||
from server.agent_core.fallback_highrisk import file_sha256
|
||
from server.aps_domain.folder_pack import folder_schedule_world_fingerprint
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
|
||
mom_path = tmp_path / "MOM主数据收集表.xlsx"
|
||
mom_path.write_bytes(b"mom-empty-v1")
|
||
monkeypatch.setattr(
|
||
"server.importers.mom_pack.parse_mom_workbook",
|
||
lambda _path: {"filename": mom_path.name, "modelSheetHits": 2, "stats": {},
|
||
"flexMaterials": [], "flexOrders": []},
|
||
)
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-mom-zero", session_id="s-mom-zero",
|
||
)
|
||
try:
|
||
before = _seed_master_data(store)
|
||
block = harness.stage_confirmation(
|
||
"s-mom-zero", "import.commit",
|
||
{"filename": mom_path.name, "momPath": str(mom_path),
|
||
"momSha256": file_sha256(mom_path),
|
||
"targetWorldFingerprint": folder_schedule_world_fingerprint(store.data)},
|
||
title="采用分析资料（测试）", summary_lines=["零业务行必须拒绝"],
|
||
)
|
||
with pytest.raises(PermissionError):
|
||
execute_confirmed(store, block.props["confirmId"], True, "approver")
|
||
assert _master_snapshot(store) == before
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_sql_adoption_uses_frozen_payload_after_source_file_is_removed(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
"""SQL 采用以冻结载荷为准：出卡后来源文件删除不影响已批准的写入。"""
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
|
||
source = tmp_path / "mini.sql"
|
||
source.write_text(MINI_SQL, encoding="utf-8")
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-sql-frozen", session_id="s-sql-frozen",
|
||
)
|
||
try:
|
||
reply = asyncio.run(_run_analyze(store, "s-sql-frozen"))
|
||
card = _adoption_card(reply)
|
||
source.unlink()
|
||
message = execute_confirmed(store, card.props["confirmId"], True, "approver")
|
||
assert "采用" in message
|
||
assert any(row.get("code") == "FG001" for row in store.data["flexMaterials"])
|
||
assert store.data["flexOrders"]
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_real_parser_ignores_empty_mom_named_workbook(tmp_path: Path, monkeypatch):
|
||
"""真实解析器路径（不打桩）：仅文件名带 MOM、无业务行的工作簿不得出采用卡。
|
||
|
||
解析器对任何工作簿都会生成 5 条钣金默认模板工序，因此业务行口径必须排除
|
||
flexOperations，否则空表会重新变成可触发整表替换的采用来源。
|
||
"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.auth.context import reset_identity
|
||
from server.importers.mom_pack import mom_pack_row_count, parse_mom_workbook
|
||
|
||
mom_path = tmp_path / "MOM2026资料.xlsx"
|
||
wb = Workbook()
|
||
wb.active.title = "数据说明"
|
||
wb.active["A1"] = "本工作簿没有任何 MOM 主数据行"
|
||
wb.save(mom_path)
|
||
|
||
pack = parse_mom_workbook(str(mom_path))
|
||
assert mom_pack_row_count(pack) == 0 # 模板占位工序不算业务行
|
||
assert pack.get("flexOperations") # 但解析器确实会生成模板占位工序
|
||
assert pack.get("modelSheetHits") == 0 # 也没有任何模型表结构证据
|
||
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-empty-real-mom", session_id="s-empty-real-mom",
|
||
)
|
||
try:
|
||
before = _seed_master_data(store)
|
||
reply = asyncio.run(_run_analyze(store, "s-empty-real-mom"))
|
||
assert not [block for block in reply.blocks if block.type == "confirm-card"]
|
||
assert _master_snapshot(store) == before
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_real_parser_zone_only_workbook_cannot_clear_master_data(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
"""只有 01-生产模型 一行车间记录的工作簿：辅助台账不算主干，不得出卡、不得整表替换。"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.agent_core import harness
|
||
from server.agent_core.fallback_highrisk import file_sha256
|
||
from server.aps_domain.folder_pack import folder_schedule_world_fingerprint
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
from server.importers.mom_pack import mom_pack_row_count, parse_mom_workbook
|
||
|
||
mom_path = tmp_path / "MOM车间清单.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "01-生产模型"
|
||
ws.append(["车间管理"])
|
||
ws.append(["", "", "车间名称", "", "装配车间"])
|
||
ws.append(["", "", "装配车间", "", "WZ-01"])
|
||
wb.save(mom_path)
|
||
|
||
pack = parse_mom_workbook(str(mom_path))
|
||
assert pack.get("flexZones") # 解析器确实产出车间（辅助台账）行
|
||
assert pack.get("flexOperations") # 以及固定模板工序
|
||
assert mom_pack_row_count(pack) == 0 # 但主干行数为 0
|
||
assert pack.get("modelSheetHits") == 1 # 且只有 1 张模型表
|
||
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-zone-mom", session_id="s-zone-mom",
|
||
)
|
||
try:
|
||
before = _seed_master_data(store)
|
||
reply = asyncio.run(_run_analyze(store, "s-zone-mom"))
|
||
assert not [block for block in reply.blocks if block.type == "confirm-card"]
|
||
assert _master_snapshot(store) == before
|
||
|
||
# 绕过出卡直接 stage 的卡也必须被执行期门槛拒绝。
|
||
block = harness.stage_confirmation(
|
||
"s-zone-mom", "import.commit",
|
||
{"filename": mom_path.name, "momPath": str(mom_path),
|
||
"momSha256": file_sha256(mom_path),
|
||
"targetWorldFingerprint": folder_schedule_world_fingerprint(store.data)},
|
||
title="采用分析资料（测试）", summary_lines=["辅助台账必须拒绝"],
|
||
)
|
||
with pytest.raises(PermissionError):
|
||
execute_confirmed(store, block.props["confirmId"], True, "approver")
|
||
assert _master_snapshot(store) == before
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_real_parser_rejects_duplicate_header_rows_and_single_sheet_mom(
|
||
tmp_path: Path, monkeypatch,
|
||
):
|
||
"""重复表头不得造出「伪物料」；只有 1 张模型表的结构证据也不足以背书整表替换。"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.agent_core import harness
|
||
from server.agent_core.fallback_highrisk import file_sha256
|
||
from server.aps_domain.folder_pack import folder_schedule_world_fingerprint
|
||
from server.aps_domain.workflow import execute_confirmed
|
||
from server.auth.context import reset_identity
|
||
from server.importers.mom_pack import (
|
||
mom_pack_is_adoptable,
|
||
mom_pack_row_count,
|
||
parse_mom_workbook,
|
||
)
|
||
|
||
mom_path = tmp_path / "MOM重复表头.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "02-工艺模型"
|
||
ws.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws.append(["物料编码", "物料名称", "1", "1", "件"]) # 重复表头（畸形来源）
|
||
ws.append(["物料编码", "物料名称", "1", "1", "件"])
|
||
ws.append(["FG-001", "演示成品", "1", "1", "件"]) # 唯一真实主干行
|
||
wb.save(mom_path)
|
||
|
||
pack = parse_mom_workbook(str(mom_path))
|
||
codes = {row.get("code") for row in pack.get("flexMaterials") or []}
|
||
assert "物料编码" not in codes # 表头文本被过滤，不会造出伪物料
|
||
assert "FG-001" in codes
|
||
assert mom_pack_row_count(pack) > 0 # 主干行确实存在（物料 + 根订单）
|
||
assert pack.get("modelSheetHits") == 1 # 结构证据不足（只有 1 张模型表）
|
||
assert mom_pack_is_adoptable(pack) is False
|
||
|
||
store, _kb, token = _prepare_env(
|
||
tmp_path, monkeypatch, project_id="p-header-mom", session_id="s-header-mom",
|
||
)
|
||
try:
|
||
before = _seed_master_data(store)
|
||
reply = asyncio.run(_run_analyze(store, "s-header-mom"))
|
||
# 退化到普通表格采用（逐行预览、合并语义）可以；升级成 MOM 整表替换卡不行。
|
||
cards = [block for block in reply.blocks if block.type == "confirm-card"]
|
||
assert not any(
|
||
((harness.pending_confirmation(card.props["confirmId"]) or {}).get("params") or {})
|
||
.get("momPath")
|
||
for card in cards
|
||
)
|
||
assert _master_snapshot(store) == before
|
||
|
||
block = harness.stage_confirmation(
|
||
"s-header-mom", "import.commit",
|
||
{"filename": mom_path.name, "momPath": str(mom_path),
|
||
"momSha256": file_sha256(mom_path),
|
||
"targetWorldFingerprint": folder_schedule_world_fingerprint(store.data)},
|
||
title="采用分析资料（测试）", summary_lines=["结构证据不足必须拒绝"],
|
||
)
|
||
with pytest.raises(PermissionError):
|
||
execute_confirmed(store, block.props["confirmId"], True, "approver")
|
||
assert _master_snapshot(store) == before
|
||
finally:
|
||
reset_identity(token)
|
||
|
||
|
||
def test_real_parser_counts_sheets_not_hints_and_filters_material_model_headers(tmp_path: Path):
|
||
"""结构证据按工作表张数计；04-物料模型的别名重复表头同样不得造出伪物料。"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.importers.mom_pack import (
|
||
mom_pack_is_adoptable,
|
||
parse_mom_workbook,
|
||
)
|
||
|
||
# 一个复合名称的单表：命中 2 个 hint，但只有 1 张工作表，不构成结构证据。
|
||
composite = tmp_path / "MOM复合表名.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "生产模型-工艺模型"
|
||
ws.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws.append(["FG-001", "演示成品", "1", "1", "件"])
|
||
wb.save(composite)
|
||
pack = parse_mom_workbook(str(composite))
|
||
assert pack.get("modelSheetHits") == 1
|
||
assert mom_pack_is_adoptable(pack) is False
|
||
|
||
# 两张模型表（结构证据成立），但 04-物料模型 只有别名重复表头、没有真实数据。
|
||
dual = tmp_path / "MOM双表.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "02-工艺模型"
|
||
ws.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws.append(["FG-001", "演示成品", "1", "1", "件"])
|
||
ws2 = wb.create_sheet("04-物料模型")
|
||
ws2.append(["品号", "物料名称", "物料组", "单位", "图号"])
|
||
ws2.append(["品号", "物料名称", "物料组", "单位", "图号"]) # 重复表头
|
||
wb.save(dual)
|
||
pack2 = parse_mom_workbook(str(dual))
|
||
codes = {row.get("code") for row in pack2.get("flexMaterials") or []}
|
||
assert codes == {"FG-001"} # 表头标签没有被当成物料
|
||
assert pack2.get("modelSheetHits") == 2
|
||
assert mom_pack_is_adoptable(pack2) is True # 真实主干行 + 结构证据
|
||
|
||
|
||
def test_real_parser_ignores_template_sentinel_rows(tmp_path: Path):
|
||
"""两张模型表但只有模板示例行：结构证据成立、真实主数据为 0，不得准入。"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.importers.mom_pack import (
|
||
mom_pack_is_adoptable,
|
||
mom_pack_row_count,
|
||
parse_mom_workbook,
|
||
)
|
||
|
||
mom_path = tmp_path / "MOM模板示例.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "01-生产模型"
|
||
ws.append(["工厂信息", "说明"])
|
||
ws2 = wb.create_sheet("02-工艺模型")
|
||
ws2.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws2.append(["示例", "示例", "1", "1", "件"]) # 模板哨兵行
|
||
wb.save(mom_path)
|
||
|
||
pack = parse_mom_workbook(str(mom_path))
|
||
assert pack.get("modelSheetHits") == 2 # 结构证据成立
|
||
assert not pack.get("flexMaterials") # 但没有任何真实物料
|
||
assert mom_pack_row_count(pack) == 0 # 也没有自动生成的订单
|
||
assert mom_pack_is_adoptable(pack) is False
|
||
|
||
|
||
def test_real_parser_ignores_instruction_rows_as_evidence(tmp_path: Path):
|
||
"""「填写说明」类填写指引行：两张模型表但只有说明文字，不得准入、不得造伪数据。"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.importers.mom_pack import (
|
||
mom_pack_is_adoptable,
|
||
mom_pack_row_count,
|
||
parse_mom_workbook,
|
||
)
|
||
|
||
# 1) 典型模板指引行：命中填写指引词 → 解析期就跳过（不入库，也不派生假订单）。
|
||
guide_path = tmp_path / "MOM填写指引.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "01-生产模型"
|
||
ws.append(["工厂信息", "说明"])
|
||
ws2 = wb.create_sheet("02-工艺模型")
|
||
ws2.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws2.append(["填写说明", "请填写真实物料信息", "1", "1", "件"])
|
||
wb.save(guide_path)
|
||
|
||
pack = parse_mom_workbook(str(guide_path))
|
||
assert pack.get("modelSheetHits") == 2
|
||
assert not pack.get("flexMaterials")
|
||
assert mom_pack_row_count(pack) == 0
|
||
assert mom_pack_is_adoptable(pack) is False
|
||
|
||
# 2) 不含指引词、也不是业务编码形态的中文占位文本：主干证据不成立（证据口径兜底）。
|
||
vague_path = tmp_path / "MOM中文占位.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "01-生产模型"
|
||
ws.append(["工厂信息", "说明"])
|
||
ws2 = wb.create_sheet("02-工艺模型")
|
||
ws2.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws2.append(["内容", "内容", "1", "1", "件"])
|
||
wb.save(vague_path)
|
||
|
||
pack2 = parse_mom_workbook(str(vague_path))
|
||
assert mom_pack_row_count(pack2) == 0 # 无字母数字编码 → 不算主干证据
|
||
assert mom_pack_is_adoptable(pack2) is False
|
||
|
||
# 3) 真实形态不受影响：中文物料名 + 字母数字编码照常准入。
|
||
real_path = tmp_path / "MOM中文物料名.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "02-工艺模型"
|
||
ws.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws.append(["FG-001", "垫片", "1", "1", "件"])
|
||
ws2 = wb.create_sheet("04-物料模型")
|
||
ws2.append(["品号", "物料名称", "物料组", "单位", "图号"])
|
||
ws2.append(["RM-0001", "钢带", "原材料", "件", ""])
|
||
wb.save(real_path)
|
||
|
||
pack3 = parse_mom_workbook(str(real_path))
|
||
assert mom_pack_row_count(pack3) > 0
|
||
assert mom_pack_is_adoptable(pack3) is True
|
||
|
||
# 4) 常见演示/测试编码（TEST-001）：同样不算主干证据，不得出「整表替换」采用卡。
|
||
test_path = tmp_path / "MOM测试编码.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "01-生产模型"
|
||
ws.append(["工厂信息", "说明"])
|
||
ws2 = wb.create_sheet("02-工艺模型")
|
||
ws2.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws2.append(["TEST-001", "测试物料", "1", "1", "件"])
|
||
wb.save(test_path)
|
||
|
||
pack4 = parse_mom_workbook(str(test_path))
|
||
assert mom_pack_row_count(pack4) == 0
|
||
assert mom_pack_is_adoptable(pack4) is False
|
||
|
||
|
||
def test_real_parser_keeps_real_rows_whose_text_merely_looks_template_like(tmp_path: Path):
|
||
"""「说明书」「测试夹具」这类真实物料：占位判定是前缀/整值匹配，不得被误杀。"""
|
||
from openpyxl import Workbook
|
||
|
||
from server.importers.mom_pack import (
|
||
mom_pack_is_adoptable,
|
||
mom_pack_row_count,
|
||
parse_mom_workbook,
|
||
)
|
||
|
||
mom_path = tmp_path / "MOM真实物料.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "02-工艺模型"
|
||
ws.append(["物料编码", "物料名称", "层级", "数量", "单位"])
|
||
ws.append(["DOC-01", "说明书", "1", "1", "件"])
|
||
ws.append(["FX-001", "测试夹具", "2", "2", "件"])
|
||
ws2 = wb.create_sheet("04-物料模型")
|
||
ws2.append(["品号", "物料名称", "物料组", "单位"])
|
||
ws2.append(["RM-0001", "钢板", "原材料", "件"])
|
||
wb.save(mom_path)
|
||
|
||
pack = parse_mom_workbook(str(mom_path))
|
||
codes = {row.get("code") for row in pack.get("flexMaterials") or []}
|
||
assert {"DOC-01", "FX-001", "RM-0001"} <= codes
|
||
assert mom_pack_row_count(pack) > 0
|
||
assert mom_pack_is_adoptable(pack) is True
|