473 lines
20 KiB
Python
473 lines
20 KiB
Python
# ============================================================
|
||
# MD-04 Excel/CSV 导入管线黄金测试
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.aps_domain.importers import apply_import_commit, preview_file, validate_batch
|
||
from server.state.seed import empty_world, seed_world
|
||
|
||
|
||
def _next_id_factory(world):
|
||
tables = {
|
||
"audit": "auditEvents",
|
||
"salesOrder": "salesOrders",
|
||
"material": "materials",
|
||
}
|
||
counters: dict[str, int] = {}
|
||
|
||
def next_id(kind: str) -> int:
|
||
if kind not in counters:
|
||
rows = world.get(tables.get(kind, kind + "s"), [])
|
||
counters[kind] = max((r.get("id", 0) for r in rows if isinstance(r, dict)), default=0)
|
||
counters[kind] += 1
|
||
return counters[kind]
|
||
|
||
return next_id
|
||
|
||
|
||
def test_validate_equipment_batch():
|
||
world = seed_world()
|
||
rows = [
|
||
{"code": "EQ-X1", "name": "试验压接机", "capabilities": "OP-PRESS,OP-WELD",
|
||
"opStdTime": "OP-PRESS:2,OP-WELD:5", "movable": "是", "zone": "ZONE-A",
|
||
"availabilityRate": "0.95", "status": "RUNNING"},
|
||
]
|
||
batch = validate_batch("equipment", rows, world)
|
||
assert batch["okCount"] == 1
|
||
assert batch["errorCount"] == 0
|
||
assert "OP-PRESS" in batch["okRows"][0]["capabilities"]
|
||
|
||
|
||
def test_preview_csv_orders_and_commit():
|
||
world = seed_world()
|
||
product = {
|
||
"id": max(m.get("id", 0) for m in world["materials"]) + 1,
|
||
"code": "RUIYANG-IMPORT-PRODUCT", "name": "锐扬导入测试成品",
|
||
"type": "FINISHED_PRODUCT", "unit": "套", "stock": 0,
|
||
}
|
||
world["materials"].append(product)
|
||
before = len(world["salesOrders"])
|
||
csv_text = (
|
||
"订单号,客户,产品编码,数量,交期,优先级\n"
|
||
f"SO-IMP-1,导入客户,{product['code']},50,2026-08-01,3\n"
|
||
).encode("utf-8")
|
||
preview = preview_file("orders.csv", csv_text, world)
|
||
assert preview["canCommit"] is True
|
||
assert preview["totalOk"] >= 1
|
||
assert preview["batches"][0]["kind"] == "orders"
|
||
|
||
applied = apply_import_commit(world, _next_id_factory(world), preview["batches"])
|
||
assert applied["total"] >= 1
|
||
assert len(world["salesOrders"]) == before + 1
|
||
assert any(o.get("customerName") == "导入客户" for o in world["salesOrders"])
|
||
assert any(o.get("orderNo") == "SO-IMP-1" for o in world["salesOrders"])
|
||
|
||
|
||
def test_commit_flex_equipment():
|
||
world = seed_world()
|
||
batches = [{
|
||
"kind": "equipment",
|
||
"sheet": "设备",
|
||
"okRows": [{
|
||
"code": "EQ-NEW-99", "name": "新压接", "capabilities": ["OP-PRESS"],
|
||
"opStdTime": {"OP-PRESS": 1.5}, "movable": True, "moveTimeMin": 15,
|
||
"zone": "ZONE-A", "availabilityRate": 1.0, "status": "RUNNING",
|
||
"adaptableMolds": [],
|
||
}],
|
||
}]
|
||
applied = apply_import_commit(world, _next_id_factory(world), batches)
|
||
assert applied["summary"].get("flexEquipment", 0) == 1
|
||
assert any(e["code"] == "EQ-NEW-99" for e in world["flexEquipment"])
|
||
|
||
|
||
def test_xlsx_mom_template_multiblock_headers_soft_preview():
|
||
"""锐扬 MOM 收集表模板：多行标题/分块表头也能识别（round-55）。
|
||
|
||
- 「02-工艺模型」标题行 + 第 4 行真实表头（序号/ERP品号/产品名称/单套用量/单位）
|
||
- 「04-物料模型」物料组块（第 1 行）+ 物料主数据块（第 5 行）
|
||
- 「07-设备模型」第 1 行表头用「财务编号/设备名称」（无可执行工序 → soft GENERAL）
|
||
"""
|
||
import io
|
||
|
||
import openpyxl
|
||
from server.aps_domain.importers import preview_file
|
||
from server.state.seed import seed_world
|
||
|
||
wb = openpyxl.Workbook()
|
||
ws1 = wb.active
|
||
ws1.title = "02-工艺模型"
|
||
ws1.append(["湖南锐扬创科智能科技有限公司\n物料清单", "", "", "", "", "", ""])
|
||
ws1.append(["", "", "", "", "", "", ""])
|
||
ws1.append(["客户代码", "", "R001", "", "", "", ""])
|
||
ws1.append(["序号", "层次", "ERP品号", "产品名称", "单套用量", "单位",
|
||
"制造流程(根据工艺代码表填写对应的工艺代码）"])
|
||
ws1.append(["", "", "", "", "", "", "01", "02", "03"])
|
||
ws1.append([1, 1, "A0050101-00280", "外壳", 1, "PCS",
|
||
"ZP01组装", "LM01POP拉铆", "BZ01机柜包装"])
|
||
ws1.append([2, 1, "B0050121-22551", "柜体焊接组件", 1, "PCS",
|
||
"ZP01组装", "WELD焊接"])
|
||
|
||
ws4 = wb.create_sheet("04-物料模型")
|
||
ws4.append(["物料组", "物料组名称", "物料组编码", "所属工厂"])
|
||
ws4.append(["", "示例：成品组", "CP", "湖南锐扬"])
|
||
ws4.append(["物料主数据", "物料名称", "物料简称", "物料编码", "物料组名称", "物料组编码", "单位"])
|
||
ws4.append(["", "外壳", "外壳", "A0050101-00280", "成品", "CP", "PCS"])
|
||
ws4.append(["", "冷轧板", "冷轧板", "C0010102-00021", "原材料", "YCL", "KG"])
|
||
|
||
ws7 = wb.create_sheet("07-设备模型")
|
||
ws7.append(["设备台账管理", "序号", "财务编号", "设备名称", "设备型号"])
|
||
ws7.append(["", "1", "A00076", "光纤激光切割机", "D-FAST1530FCCBD300"])
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
world = seed_world()
|
||
preview = preview_file("湖南锐扬MOM主数据收集表.xlsx", buf.getvalue(), world, soft=True)
|
||
assert preview["canCommit"] is True
|
||
batches = {b["sheet"]: b for b in preview["batches"]}
|
||
assert batches["02-工艺模型"]["okCount"] == 2
|
||
row = batches["02-工艺模型"]["okRows"][0]
|
||
assert row["code"] == "A0050101-00280" and row["name"] == "外壳"
|
||
# 制造流程列 → routing 视图（round-57）
|
||
rts = batches["02-工艺模型#routing"]
|
||
assert rts["kind"] == "routing" and rts["okCount"] == 5
|
||
by_key = {(r["productCode"], r["seq"]): r for r in rts["okRows"]}
|
||
first = by_key[("A0050101-00280", 1)]
|
||
assert first["operationCode"] == "ZP01" and first["operationName"] == "组装"
|
||
assert by_key[("A0050101-00280", 2)]["operationName"] == "拉铆"
|
||
assert by_key[("B0050121-22551", 2)]["operationCode"] == "WELD"
|
||
assert batches["04-物料模型"]["okCount"] == 2
|
||
by_code = {r["code"]: r for r in batches["04-物料模型"]["okRows"]}
|
||
assert by_code["A0050101-00280"]["type"] == "FINISHED_PRODUCT"
|
||
assert by_code["C0010102-00021"]["type"] == "RAW_MATERIAL"
|
||
eq = batches["07-设备模型"]
|
||
assert eq["okCount"] == 1
|
||
assert eq["okRows"][0]["code"] == "A00076"
|
||
assert eq["okRows"][0]["capabilities"] == ["GENERAL"]
|
||
|
||
|
||
def test_xlsx_ruiyang_structured_diagnostics_preserve_source_rows_and_legacy_contract():
|
||
"""锐扬 53 行语义缩比：可恢复解析噪声、结构行和唯一真实阻塞应分离。"""
|
||
import io
|
||
from collections import Counter
|
||
|
||
import openpyxl
|
||
from server.aps_domain.importers import preview_file
|
||
from server.state.seed import seed_world
|
||
|
||
wb = openpyxl.Workbook()
|
||
|
||
# 02-工艺模型：第 5 行是制造流程二级序号表头，不是 materials 数据。
|
||
ws2 = wb.active
|
||
ws2.title = "02-工艺模型"
|
||
ws2.append(["湖南锐扬创科智能科技有限公司\n物料清单"] + [""] * 27)
|
||
ws2.append([""] * 28)
|
||
ws2.append(["客户代码", "", "R001"] + [""] * 25)
|
||
ws2.append([
|
||
"序号", "层次", "ERP品号", "客户图号", "产品名称", "客户版本",
|
||
"单套用量", "出数/底数", "单位", "品号属性", "仓库代码", "材质",
|
||
"厚度", "展开长*宽", "面积", "表面处理",
|
||
"制造流程(根据工艺代码表填写对应的工艺代码）",
|
||
] + [""] * 11)
|
||
for col in range(1, 17):
|
||
ws2.merge_cells(start_row=4, start_column=col, end_row=5, end_column=col)
|
||
ws2.merge_cells("Q4:AB4")
|
||
for offset in range(12):
|
||
ws2.cell(row=5, column=17 + offset, value=f"{offset + 1:02d}")
|
||
material_row = [None] * 28
|
||
material_row[0] = 1
|
||
material_row[1] = 1
|
||
material_row[2] = "MAT-001"
|
||
material_row[4] = "合成外壳"
|
||
material_row[6] = 1
|
||
material_row[8] = "PCS"
|
||
material_row[16] = "ZP01组装"
|
||
ws2.append(material_row)
|
||
|
||
# 07-设备模型：2 条重复 code 别名噪声、4 条嵌入 section、1 条真实缺码。
|
||
ws7 = wb.create_sheet("07-设备模型")
|
||
ws7.append([
|
||
"设备台账管理", "序号", "财务编号", "固定资产编号", "设备名称",
|
||
"设备型号", "厂内编号", "出厂编号", "位置/一级保养责任人", "供应商",
|
||
"购置年月", "设备类型", "设备状态", "备注", "", "设备状态",
|
||
])
|
||
ws7.append(["", 1, "FIN-VALID", "FA-VALID", "厂内编码设备", "MODEL-A",
|
||
"INTERNAL-VALID", "SERIAL-A", "A区", "供应商A", None,
|
||
"生产设备", "使用中"])
|
||
ws7.append(["", 2, "FIN-001", "FA-001", "财务编码设备一", "MODEL-B",
|
||
None, "SERIAL-B", "B区", "供应商B", None, "生产设备", "使用中"])
|
||
ws7.append(["", 3, "FIN-002", "FA-002", "财务编码设备二", "MODEL-C",
|
||
None, "SERIAL-C", "C区", "供应商C", None, "生产设备", "使用中"])
|
||
ws7.append(["", 4, None, None, "唯一缺码设备", "MODEL-D",
|
||
None, "SERIAL-D", "D区", "供应商D", None, "生产设备", "使用中"])
|
||
ws7.merge_cells("A1:A5")
|
||
|
||
ws7.append(["备品备件管理", "备件名称", "备件编码", "规格型号", "单位", "备注"])
|
||
ws7.append([None] * 16)
|
||
ws7.merge_cells("A6:A7")
|
||
ws7.append(["设备部件管理", "部件名称", "部件编码", "规格型号", "单位", "备注"])
|
||
ws7.append([None] * 16)
|
||
ws7.merge_cells("A8:A9")
|
||
ws7.append(["设备类型", "类型名称", "类型编码", "备注"])
|
||
ws7.append(["", "生产设备", "SBLX001"])
|
||
ws7.append([None] * 16)
|
||
ws7.merge_cells("A10:A12")
|
||
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
preview = preview_file(
|
||
"湖南锐扬MOM主数据收集表.xlsx",
|
||
buf.getvalue(),
|
||
seed_world(),
|
||
soft=True,
|
||
)
|
||
|
||
# 兼容性：原 preview/batch 字段必须继续存在。
|
||
assert {"filename", "batches", "totalOk", "totalErrors", "canCommit"} <= preview.keys()
|
||
batches = {b["sheet"]: b for b in preview["batches"]}
|
||
materials = batches["02-工艺模型"]
|
||
equipment = batches["07-设备模型"]
|
||
legacy_batch_fields = {
|
||
"sheet", "kind", "fieldMap", "headersRaw", "okRows", "errors",
|
||
"warnings", "okCount", "errorCount",
|
||
}
|
||
assert legacy_batch_fields <= materials.keys()
|
||
assert legacy_batch_fields <= equipment.keys()
|
||
assert isinstance(materials["diagnostics"], list)
|
||
assert isinstance(equipment["diagnostics"], list)
|
||
|
||
# 02 row 5：SECONDARY_SEQUENCE_HEADER 必须忽略且保留真实 Excel 行号。
|
||
secondary = [
|
||
d for d in materials["diagnostics"]
|
||
if d.get("code") == "SECONDARY_SEQUENCE_HEADER"
|
||
]
|
||
assert len(secondary) == 1
|
||
assert secondary[0]["classification"] == "ignorable_section"
|
||
assert secondary[0]["excelRow"] == 5
|
||
assert materials["okCount"] == 1
|
||
assert materials["errorCount"] == 0
|
||
|
||
# 重复 code 别名：最后一个非空编码优先；后续空列不能覆盖已有编码。
|
||
equipment_by_code = {row["code"]: row for row in equipment["okRows"]}
|
||
assert {"INTERNAL-VALID", "FA-001", "FA-002"} <= equipment_by_code.keys()
|
||
assert equipment["okCount"] == 3
|
||
assert equipment["errorCount"] == 1
|
||
assert preview["totalErrors"] == 1
|
||
|
||
# 真实 47/5/1 的小型缩比：2 条 parser_noise、5 条结构忽略、1 条 blocking。
|
||
all_diagnostics = materials["diagnostics"] + equipment["diagnostics"]
|
||
primary_classes = Counter(
|
||
d.get("classification") for d in all_diagnostics
|
||
if d.get("classification") in {"parser_noise", "ignorable_section", "blocking"}
|
||
)
|
||
assert primary_classes == Counter({
|
||
"parser_noise": 2,
|
||
"ignorable_section": 5,
|
||
"blocking": 1,
|
||
})
|
||
assert {
|
||
d["excelRow"] for d in equipment["diagnostics"]
|
||
if d.get("classification") == "parser_noise"
|
||
} == {3, 4}
|
||
assert {
|
||
d["excelRow"] for d in equipment["diagnostics"]
|
||
if d.get("classification") == "ignorable_section"
|
||
} == {6, 8, 10, 11}
|
||
blocking = [
|
||
d for d in equipment["diagnostics"]
|
||
if d.get("classification") == "blocking"
|
||
]
|
||
assert len(blocking) == 1 and blocking[0]["excelRow"] == 5
|
||
|
||
# soft 模式下 GENERAL warning 与成功行共存，不能把成功行重新算成 error。
|
||
assert all(equipment_by_code[code]["capabilities"] == ["GENERAL"]
|
||
for code in ("INTERNAL-VALID", "FA-001", "FA-002"))
|
||
assert any("GENERAL" in str(warning) for warning in equipment["warnings"])
|
||
general_diagnostics = [
|
||
d for d in equipment["diagnostics"]
|
||
if d.get("classification") == "warning" and "GENERAL" in str(d)
|
||
]
|
||
assert general_diagnostics
|
||
assert all(isinstance(d.get("excelRow"), int) for d in all_diagnostics)
|
||
|
||
# __source__ 只供内部追踪，不能泄漏到兼容的 okRows 领域行。
|
||
assert all(
|
||
"__source__" not in row
|
||
for batch in preview["batches"]
|
||
for row in batch["okRows"]
|
||
)
|
||
|
||
|
||
def test_xlsx_flat_header_row_0_behavior_unchanged():
|
||
"""平铺表第 1 行即表头：探测不得改变既有行为（round-55 回归）。"""
|
||
import io
|
||
|
||
import openpyxl
|
||
from server.aps_domain.importers import preview_file
|
||
from server.state.seed import seed_world
|
||
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "物料"
|
||
ws.append(["编码", "名称", "类型", "单位", "库存"])
|
||
ws.append(["M-FLAT-1", "平铺物料", "FINISHED_PRODUCT", "件", "10"])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
world = seed_world()
|
||
preview = preview_file("物料.xlsx", buf.getvalue(), world, soft=True)
|
||
assert preview["canCommit"] is True
|
||
assert preview["batches"][0]["okRows"][0]["code"] == "M-FLAT-1"
|
||
|
||
|
||
def test_order_csv_rush_column_maps_to_is_rush():
|
||
"""订单表「加急」列 → isRush（round-58，现场订单表常用列）。"""
|
||
from server.aps_domain.importers import preview_file
|
||
from server.state.seed import seed_world
|
||
|
||
world = seed_world()
|
||
product = {
|
||
"id": max(m.get("id", 0) for m in world["materials"]) + 1,
|
||
"code": "RUIYANG-IMPORT-PRODUCT", "name": "锐扬导入测试成品",
|
||
"type": "FINISHED_PRODUCT", "unit": "套", "stock": 0,
|
||
}
|
||
world["materials"].append(product)
|
||
csv_text = (
|
||
"订单号,客户,产品编码,数量,交期,优先级,加急\n"
|
||
f"SO-RUSH-1,现场客户,{product['code']},10,2026-08-10,3,是\n"
|
||
).encode("utf-8")
|
||
preview = preview_file("订单.csv", csv_text, world, soft=True)
|
||
assert preview["totalOk"] >= 1
|
||
row = preview["batches"][0]["okRows"][0]
|
||
assert row["isRush"] is True
|
||
assert row["orderNo"] == "SO-RUSH-1"
|
||
|
||
|
||
def test_material_create_ids_do_not_collide_with_order_projection():
|
||
"""订单先入库时 sync 会直接建成品；原材料 create 不得复用外部发号器的旧值。"""
|
||
world = empty_world()
|
||
counters: dict[str, int] = {}
|
||
|
||
def stale_next_id(kind: str) -> int:
|
||
counters[kind] = counters.get(kind, 0) + 1
|
||
return counters[kind]
|
||
|
||
orders = [{
|
||
"kind": "orders", "sheet": "销售订单",
|
||
"okRows": [{
|
||
"orderNo": "SO-ID-1", "productCode": "FG-ID", "productName": "成品",
|
||
"quantity": 2, "deliveryDate": "2026-08-12", "priority": 5,
|
||
}],
|
||
}]
|
||
apply_import_commit(world, stale_next_id, orders)
|
||
materials = [{
|
||
"kind": "materials", "sheet": "物料",
|
||
"okRows": [
|
||
{"code": "FG-ID", "name": "成品", "type": "FINISHED_PRODUCT", "unit": "件", "stock": 0},
|
||
{"code": "RM-ID", "name": "原料", "type": "RAW_MATERIAL", "unit": "PCS", "stock": 0},
|
||
],
|
||
}]
|
||
apply_import_commit(world, stale_next_id, materials)
|
||
|
||
rows = world["materials"]
|
||
assert len(rows) == len({row["id"] for row in rows})
|
||
assert len([row for row in rows if row.get("code") == "RM-ID"]) == 1
|
||
|
||
|
||
def test_generic_batches_sync_flex_bom_and_routing_for_mrp():
|
||
"""通用 Excel 批次必须同步经典 BOM/工艺，否则 MRP 只能生成根制造需求。"""
|
||
world = seed_world()
|
||
product_code = "RUIYANG-SYNC-PRODUCT"
|
||
raw_code = "RUIYANG-SYNC-RAW"
|
||
batches = [
|
||
{
|
||
"kind": "materials",
|
||
"sheet": "物料",
|
||
"okRows": [
|
||
{"code": product_code, "name": "锐扬同步成品", "type": "FINISHED_PRODUCT", "unit": "套", "stock": 0},
|
||
{"code": raw_code, "name": "锐扬同步原料", "type": "RAW_MATERIAL", "unit": "PCS", "stock": 0,
|
||
"inTransit": 0, "procurementLeadTime": 3},
|
||
],
|
||
},
|
||
{
|
||
"kind": "routing",
|
||
"sheet": "工艺路线",
|
||
"okRows": [
|
||
{"productCode": product_code, "seq": 1, "operationCode": "CUT",
|
||
"operationName": "激光下料", "stdTimePerUnit": 5.0, "sourcingType": "MAKE"},
|
||
],
|
||
},
|
||
{
|
||
"kind": "bom",
|
||
"sheet": "BOM",
|
||
"okRows": [
|
||
{"productCode": product_code, "materialCode": raw_code, "quantity": 2.0, "isKey": True},
|
||
],
|
||
},
|
||
{
|
||
"kind": "orders",
|
||
"sheet": "销售订单",
|
||
"okRows": [
|
||
{"orderNo": "SO-RUIYANG-SYNC", "productCode": product_code, "quantity": 3,
|
||
"deliveryDate": "2026-08-12", "priority": 2},
|
||
],
|
||
},
|
||
]
|
||
|
||
apply_import_commit(world, _next_id_factory(world), batches)
|
||
|
||
product = next(row for row in world["materials"] if row.get("code") == product_code)
|
||
assert any(row.get("productId") == product["id"] for row in world["boms"])
|
||
assert any(row.get("productId") == product["id"] for row in world["routings"])
|
||
assert any(row.get("orderNo") == "SO-RUIYANG-SYNC" for row in world["salesOrders"])
|
||
|
||
from server.aps_domain.mrp import decompose_orders
|
||
|
||
result = decompose_orders(world, _next_id_factory(world), "SO-RUIYANG-SYNC")
|
||
assert len(result["make"]) == 1
|
||
assert len(result["purchase"]) == 1
|
||
assert result["purchase"][0]["materialCode"] == raw_code
|
||
assert result["purchase"][0]["quantity"] == 6
|
||
|
||
|
||
def test_ruiyang_product_and_bom_header_aliases():
|
||
world = seed_world()
|
||
product = validate_batch(
|
||
"materials",
|
||
[{"productCode": "RY-P-001", "productName": "锐扬产品", "type": "成品", "unit": "套"}],
|
||
world,
|
||
soft=True,
|
||
)
|
||
assert product["okRows"][0]["code"] == "RY-P-001"
|
||
assert product["okRows"][0]["name"] == "锐扬产品"
|
||
|
||
raw = (
|
||
"父项产品编码,子项物料编码,单套用量,关键料\n"
|
||
"RY-P-001,RY-M-001,2,是\n"
|
||
).encode("utf-8-sig")
|
||
preview = preview_file("BOM.csv", raw, world, soft=True)
|
||
batch = preview["batches"][0]
|
||
assert batch["okCount"] == 1
|
||
assert batch["okRows"][0] == {
|
||
"productCode": "RY-P-001",
|
||
"materialCode": "RY-M-001",
|
||
"quantity": 2.0,
|
||
"consumeOp": "",
|
||
"isKey": True,
|
||
}
|
||
|
||
|
||
def test_calendar_csv_import_supports_continuous_demo_windows():
|
||
world = seed_world()
|
||
raw = (
|
||
"shiftCode,name,startTime,endTime,workdays,breaks,enabled\n"
|
||
"SIM08,连续仿真窗08,08:00,08:00,1/2/3/4/5/6/7,,是\n"
|
||
).encode("utf-8-sig")
|
||
preview = preview_file("06_calendar.csv", raw, world, soft=True)
|
||
batch = preview["batches"][0]
|
||
assert batch["kind"] == "calendar"
|
||
assert batch["okCount"] == 1
|
||
apply_import_commit(world, _next_id_factory(world), preview["batches"])
|
||
row = next(item for item in world["flexCalendar"] if item.get("shiftCode") == "SIM08")
|
||
assert row["startTime"] == "08:00"
|
||
assert row["endTime"] == "08:00"
|
||
assert row["workdays"] == [1, 2, 3, 4, 5, 6, 7]
|