729 lines
28 KiB
Python
729 lines
28 KiB
Python
# ============================================================
|
|
# 现场完整生产路线导入黄金测试
|
|
# ============================================================
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from server.aps_domain.kangni_intake import (
|
|
DEFAULT_ROUTE_XLSX,
|
|
build_flex_bundle,
|
|
load_site_into_world,
|
|
parse_complete_route_xlsx,
|
|
)
|
|
from server.engines import PoolEngine
|
|
from server.state.seed import seed_world
|
|
|
|
ROUTE = Path(DEFAULT_ROUTE_XLSX)
|
|
|
|
|
|
@pytest.mark.skipif(not ROUTE.exists(), reason="现场 Excel 不在本机")
|
|
def test_parse_complete_route_order_102285668():
|
|
pkg = parse_complete_route_xlsx(ROUTE)
|
|
assert pkg["orderNo"] == "102285668"
|
|
assert pkg["productCode"] == "28200003654300"
|
|
assert len(pkg["operations"]) == 10
|
|
assert pkg["operations"][0]["operationCode"] == "0503727Z1M"
|
|
assert len(pkg["bom"]) >= 15
|
|
assert all(op["stdTimePerUnit"] > 0 for op in pkg["operations"])
|
|
assert any(op["stdTimeInferred"] for op in pkg["operations"])
|
|
|
|
|
|
@pytest.mark.skipif(not ROUTE.exists(), reason="现场 Excel 不在本机")
|
|
def test_load_site_writes_master_and_orders_and_schedules():
|
|
world = seed_world()
|
|
assert any(m["code"].startswith("CTRL-") for m in world["materials"])
|
|
assert any(o["orderNo"].startswith("SO") for o in world["salesOrders"])
|
|
|
|
meta = load_site_into_world(world, route_path=ROUTE, include_sibling_orders=False, clear_all=True)
|
|
assert meta["primaryOrder"] == "102285668"
|
|
assert meta.get("clearedDemo") is True
|
|
|
|
# 演示垃圾已清
|
|
assert not any(m["code"].startswith("CTRL-") for m in world["materials"])
|
|
assert not any(o["orderNo"].startswith("FO-") for o in world["flexOrders"])
|
|
assert not any(o["orderNo"].startswith("SO20") for o in world["salesOrders"])
|
|
|
|
# 主数据页可读
|
|
assert world["factories"][0]["name"] == "城轨机构装配工厂"
|
|
assert len(world["operations"]) == 10
|
|
assert any(m["code"] == "28200003654300" for m in world["materials"])
|
|
assert len(world["bomItems"]) >= 15
|
|
assert len(world["routingSteps"]) == 10
|
|
assert world["salesOrders"][0]["orderNo"] == "102285668"
|
|
assert world["salesOrders"][0]["status"] == "APPROVED"
|
|
|
|
# 柔性轨
|
|
assert world["flexOrders"][0]["orderNo"] == "102285668"
|
|
assert world["flexEquipment"]
|
|
assert world["flexParams"].get("fixedMasterSynced") is True
|
|
|
|
counters: dict[str, int] = {}
|
|
|
|
def nid(k: str) -> int:
|
|
counters[k] = counters.get(k, 0) + 1
|
|
return counters[k]
|
|
|
|
result = PoolEngine().solve(world, nid, sort_mode="BOTTLENECK", start_date="2026-06-26")
|
|
assert result["orderCount"] == 1
|
|
assert result["vlCount"] == 1
|
|
assert result["woCount"] == 10
|
|
|
|
|
|
@pytest.mark.skipif(not ROUTE.exists(), reason="现场 Excel 不在本机")
|
|
def test_bundle_meta_has_inferences():
|
|
bundle = build_flex_bundle(ROUTE, include_sibling_orders=False)
|
|
assert bundle["_meta"]["inferences"]
|
|
assert any("工时" in x or "工位" in x for x in bundle["_meta"]["inferences"])
|
|
|
|
|
|
def _write_std_time_fixtures(tmp_path: Path) -> tuple[Path, Path]:
|
|
from openpyxl import Workbook
|
|
|
|
route = tmp_path / "工艺路线.xlsx"
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "102285668"
|
|
ws.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时", "排序号"])
|
|
ws.append(["1", "0503727Z1M", "干线机构部装", "部装", None, "0010"])
|
|
ws.append(["2", "0503722Z1M", "干线机构总装1", "装配", None, "0020"])
|
|
wb.save(route)
|
|
|
|
std = tmp_path / "工时.xlsx"
|
|
wb2 = Workbook()
|
|
ws2 = wb2.active
|
|
ws2.title = "102285668"
|
|
ws2.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时/分钟", "排序号"])
|
|
ws2.append(["1", "0503727Z1M", "干线机构部装", "部装", 27.0, "0010"])
|
|
ws2.append(["2", "0503722Z1M", "干线机构总装1", "装配", 5.0, "0020"])
|
|
wb2.save(std)
|
|
return route, std
|
|
|
|
|
|
def test_routing_blank_std_prefers_std_time_workbook(tmp_path):
|
|
from server.aps_domain.kangni_intake import load_std_time_map, parse_routing_sheet
|
|
|
|
route, _ = _write_std_time_fixtures(tmp_path)
|
|
times = load_std_time_map(tmp_path)
|
|
assert times["102285668"]["0503727Z1M"] == 27.0
|
|
ops = parse_routing_sheet(route, "102285668", std_times=times["102285668"])
|
|
assert ops[0]["stdTimePerUnit"] == 27.0
|
|
assert ops[0]["stdTimeInferred"] is False
|
|
assert ops[0]["stdTimeSource"] == "工时表"
|
|
assert ops[1]["stdTimePerUnit"] == 5.0
|
|
|
|
|
|
def test_routing_blank_std_falls_back_to_inference(tmp_path):
|
|
from server.aps_domain.kangni_intake import parse_routing_sheet
|
|
|
|
route, _ = _write_std_time_fixtures(tmp_path)
|
|
ops = parse_routing_sheet(route, "102285668")
|
|
assert ops[0]["stdTimeInferred"] is True
|
|
assert ops[0]["stdTimeSource"] == "推断"
|
|
assert ops[0]["stdTimePerUnit"] > 0
|
|
|
|
|
|
def _write_mapping_fixtures(tmp_path: Path) -> tuple[Path, Path]:
|
|
from openpyxl import Workbook
|
|
|
|
equip = tmp_path / "设备能力映射模板.xlsx"
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "设备能力"
|
|
ws.append(["设备编号", "设备名称", "可执行工序编号", "单件工时(分钟,工序:分钟;多值分号)", "可动率",
|
|
"是否可移动", "移动耗时(分钟)", "区域编码", "适配模具编号", "状态", "备注"])
|
|
ws.append(["EQ-CG-01", "城轨装配台1", "0503727Z1M,0503722Z1M",
|
|
"0503727Z1M:27;0503722Z1M:5", 0.9, "否", 0,
|
|
"ZONE-CG", "MOLD-01", "RUNNING", ""])
|
|
wb.save(equip)
|
|
|
|
mold = tmp_path / "模具适配映射模板.xlsx"
|
|
wb2 = Workbook()
|
|
ws2 = wb2.active
|
|
ws2.title = "模具适配"
|
|
ws2.append(["模具编号", "模具名称", "适用工序编号", "适配设备编号(逗号分隔)",
|
|
"寿命上限", "已用寿命", "区域编码", "换型耗时(分钟)", "状态", "备注"])
|
|
ws2.append(["MOLD-01", "压接模具A", "0503722Z1M", "EQ-CG-01",
|
|
50000, 12000, "ZONE-CG", 20, "AVAILABLE", ""])
|
|
wb2.save(mold)
|
|
return equip, mold
|
|
|
|
|
|
def test_parse_equipment_and_mold_maps(tmp_path):
|
|
from server.aps_domain.kangni_intake import (
|
|
parse_equipment_capability_map,
|
|
parse_mold_adaptation_map,
|
|
)
|
|
|
|
equip, mold = _write_mapping_fixtures(tmp_path)
|
|
rows = parse_equipment_capability_map(equip)
|
|
assert rows[0]["code"] == "EQ-CG-01"
|
|
assert rows[0]["capabilities"] == ["0503727Z1M", "0503722Z1M"]
|
|
assert rows[0]["opStdTime"]["0503727Z1M"] == 27.0
|
|
assert rows[0]["movable"] is False
|
|
assert rows[0]["availabilityRate"] == 0.9
|
|
|
|
mrows = parse_mold_adaptation_map(mold)
|
|
assert mrows[0]["code"] == "MOLD-01"
|
|
assert mrows[0]["operationCode"] == "0503722Z1M"
|
|
assert mrows[0]["adaptableEquipment"] == ["EQ-CG-01"]
|
|
assert mrows[0]["lifeTotal"] == 50000
|
|
|
|
|
|
def test_build_flex_equipment_prefers_mapping_table(tmp_path):
|
|
from server.aps_domain.kangni_intake import build_flex_equipment_and_molds
|
|
|
|
_write_mapping_fixtures(tmp_path)
|
|
equipment, molds, mold_ops, infs = build_flex_equipment_and_molds(
|
|
tmp_path, [], {"0503727Z1M": 27.0, "0503722Z1M": 5.0},
|
|
["0503727Z1M", "0503722Z1M"], {"zone": {"code": "ZONE-CG"}},
|
|
)
|
|
assert len(equipment) == 1
|
|
assert equipment[0]["inferred"] is False
|
|
assert equipment[0]["availabilityRate"] == 0.9
|
|
assert len(molds) == 1
|
|
assert "0503722Z1M" in mold_ops
|
|
assert any("设备能力映射表" in x for x in infs)
|
|
assert any("模具适配映射表" in x for x in infs)
|
|
|
|
|
|
def test_build_flex_equipment_falls_back_to_synthetic(tmp_path):
|
|
from server.aps_domain.kangni_intake import build_flex_equipment_and_molds
|
|
|
|
equipment, molds, mold_ops, infs = build_flex_equipment_and_molds(
|
|
tmp_path, [], {"OP-1": 10.0}, ["OP-1"],
|
|
{"stationCount": 2, "zone": {"code": "ZONE-CG"}, "availabilityRate": 0.95},
|
|
)
|
|
assert len(equipment) == 2
|
|
assert all(e["inferred"] for e in equipment)
|
|
assert molds == []
|
|
assert mold_ops == set()
|
|
assert any("合成 2 台" in x for x in infs)
|
|
|
|
|
|
def test_fixed_master_preserves_measured_table_and_inferred_time_sources():
|
|
from server.aps_domain.kangni_intake import _apply_std_time_map, build_fixed_master
|
|
|
|
package = {
|
|
"orderNo": "TIME-SOURCE-001",
|
|
"productCode": "TIME-SOURCE-PRODUCT",
|
|
"productName": "工时来源验证产品",
|
|
"drawingNo": "DRAWING-001",
|
|
"quantity": 1.0,
|
|
"planStart": "2026-08-17",
|
|
"dueDate": "2026-08-20",
|
|
"wbs": "WBS-TIME-SOURCE",
|
|
"project": "康尼工时来源验证",
|
|
"status": "RELEASED",
|
|
"kitStatus": "未知",
|
|
"bom": [],
|
|
"operations": [
|
|
{
|
|
"seq": 10,
|
|
"operationCode": "OP-MEASURED",
|
|
"operationName": "实测工序",
|
|
"opType": "装配",
|
|
"stdTimePerUnit": 11.0,
|
|
"stdTimeInferred": False,
|
|
"stdTimeSource": "实测",
|
|
},
|
|
{
|
|
"seq": 20,
|
|
"operationCode": "OP-TABLE",
|
|
"operationName": "工时表工序",
|
|
"opType": "装配",
|
|
"stdTimePerUnit": 30.0,
|
|
"stdTimeInferred": True,
|
|
"stdTimeSource": "推断",
|
|
},
|
|
{
|
|
"seq": 30,
|
|
"operationCode": "OP-INFERRED",
|
|
"operationName": "推断工序",
|
|
"opType": "装配",
|
|
"stdTimePerUnit": 33.0,
|
|
"stdTimeInferred": True,
|
|
"stdTimeSource": "推断",
|
|
},
|
|
],
|
|
}
|
|
_apply_std_time_map(
|
|
package,
|
|
{"TIME-SOURCE-001": {"OP-MEASURED": 99.0, "OP-TABLE": 22.0}},
|
|
)
|
|
|
|
assert package["operations"][0]["stdTimePerUnit"] == 11.0
|
|
assert package["operations"][0]["stdTimeSource"] == "实测"
|
|
assert package["operations"][1]["stdTimePerUnit"] == 22.0
|
|
assert package["operations"][1]["stdTimeSource"] == "工时表"
|
|
assert package["operations"][2]["stdTimePerUnit"] == 33.0
|
|
assert package["operations"][2]["stdTimeSource"] == "推断"
|
|
|
|
fixed = build_fixed_master([package], station_count=1)
|
|
steps = fixed["routingSteps"]
|
|
assert [row["runTimePerUnit"] for row in steps] == [11.0, 22.0, 33.0]
|
|
assert [row["stdTimeSource"] for row in steps] == ["实测", "工时表", "推断"]
|
|
|
|
|
|
def _write_route_book(path: Path, sheets: dict[str, list[tuple[str, str, str, str]]]) -> None:
|
|
from openpyxl import Workbook
|
|
|
|
wb = Workbook()
|
|
wb.remove(wb.active)
|
|
for sheet_name, operations in sheets.items():
|
|
ws = wb.create_sheet(sheet_name)
|
|
ws.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时", "排序号"])
|
|
for index, (code, seq, name, op_type) in enumerate(operations, start=1):
|
|
ws.append([index, code, name, op_type, None, seq])
|
|
wb.save(path)
|
|
|
|
|
|
def _write_std_book(path: Path, orders: dict[str, list[tuple[str, str, str, str, float]]]) -> None:
|
|
from openpyxl import Workbook
|
|
|
|
wb = Workbook()
|
|
wb.remove(wb.active)
|
|
for order_no, operations in orders.items():
|
|
ws = wb.create_sheet(order_no)
|
|
ws.append(["序号", "工序编号", "工序名称", "工序类型", "标准工时/分钟", "排序号"])
|
|
for index, (code, seq, name, op_type, minutes) in enumerate(operations, start=1):
|
|
ws.append([index, code, name, op_type, minutes, seq])
|
|
wb.save(path)
|
|
|
|
|
|
def _write_bom_book(path: Path, sheets: dict[str, list[tuple[str, str, str, float, str]]]) -> None:
|
|
from openpyxl import Workbook
|
|
|
|
wb = Workbook()
|
|
wb.remove(wb.active)
|
|
for sheet_name, rows in sheets.items():
|
|
ws = wb.create_sheet(sheet_name)
|
|
ws.append(["工单编号", "MES工序编号", "物料编号", "物料名称", "单位用量"])
|
|
for order_no, operation, material, quantity, name in rows:
|
|
ws.append([order_no, operation, material, name, quantity])
|
|
wb.save(path)
|
|
|
|
|
|
def _write_orders_book(path: Path, order_numbers: list[str]) -> None:
|
|
from openpyxl import Workbook
|
|
|
|
wb = Workbook()
|
|
ws = wb.active
|
|
ws.title = "订单"
|
|
ws.append([
|
|
"生产订单", "WBS号", "料号", "物料名称",
|
|
"计划开始时间", "计划结束时间", "订单数量",
|
|
])
|
|
for index, order_no in enumerate(order_numbers, start=1):
|
|
ws.append([
|
|
order_no,
|
|
f"WBS-{index:03}",
|
|
f"PRODUCT-{index:03}",
|
|
f"产品{index:03}",
|
|
"2026-08-01",
|
|
"2026-08-31",
|
|
1,
|
|
])
|
|
wb.save(path)
|
|
|
|
|
|
def _write_data_dir_fixture(root: Path, order_count: int = 10) -> list[str]:
|
|
root.mkdir(parents=True, exist_ok=True)
|
|
order_numbers = [f"ORDER-{index:03}" for index in range(1, order_count + 1)]
|
|
_write_orders_book(root / "订单.xlsx", order_numbers)
|
|
_write_route_book(root / "工艺路线.xlsx", {
|
|
order_no: [(f"OP-{index:03}", "0010", f"工序{index:03}", "装配")]
|
|
for index, order_no in enumerate(order_numbers, start=1)
|
|
})
|
|
_write_std_book(root / "工时.xlsx", {
|
|
order_no: [(f"OP-{index:03}", "0010", f"工序{index:03}", "装配", float(index))]
|
|
for index, order_no in enumerate(order_numbers, start=1)
|
|
})
|
|
_write_bom_book(root / "BOM.xlsx", {
|
|
order_no: [(order_no, f"OP-{index:03}", f"MAT-{index:03}", 1.0, f"物料{index:03}")]
|
|
for index, order_no in enumerate(order_numbers, start=1)
|
|
})
|
|
return order_numbers
|
|
|
|
|
|
def test_data_dir_only_payload_builds_10_orders_and_apply_is_pure(tmp_path: Path):
|
|
import json
|
|
import shutil
|
|
from copy import deepcopy
|
|
|
|
from server.aps_domain.kangni_intake import (
|
|
apply_site_payload_to_world,
|
|
build_site_payload_from_data_dir,
|
|
)
|
|
|
|
data_dir = tmp_path / "upload"
|
|
_write_data_dir_fixture(data_dir)
|
|
assert not list(data_dir.glob("*完整生产路线*.xlsx"))
|
|
|
|
payload = build_site_payload_from_data_dir(data_dir, station_count=2)
|
|
serialized = json.dumps(payload, ensure_ascii=False, sort_keys=True, allow_nan=False)
|
|
assert json.loads(serialized)["meta"]["sourceMode"] == "data-dir-only"
|
|
assert str(data_dir) not in serialized
|
|
assert payload["meta"]["sourceFiles"] == {
|
|
"orders": "订单.xlsx",
|
|
"routing": "工艺路线.xlsx",
|
|
"stdtime": "工时.xlsx",
|
|
"bom": "BOM.xlsx",
|
|
"workbooks": ["BOM.xlsx", "工时.xlsx", "工艺路线.xlsx", "订单.xlsx"],
|
|
}
|
|
assert len(payload["fixed"]["salesOrders"]) == 10
|
|
assert len(payload["fixed"]["routingSteps"]) == 10
|
|
assert len(payload["flex"]["flexOrders"]) == 10
|
|
assert len(payload["flex"]["flexRoutings"]) == 10
|
|
assert len(payload["flex"]["flexOperations"]) == 10
|
|
assert len(payload["flex"]["flexBom"]) == 10
|
|
assert payload["meta"]["bomSourceRowCount"] == 10
|
|
assert payload["meta"]["stdTimeSourceCounts"] == {"工时表": 10}
|
|
assert all(
|
|
row["stdTimeSource"] == "工时表"
|
|
for row in payload["flex"]["flexRoutings"]
|
|
)
|
|
|
|
mirror_dir = tmp_path / "different-upload-root"
|
|
mirror_dir.mkdir()
|
|
for source in data_dir.glob("*.xlsx"):
|
|
shutil.copy2(source, mirror_dir / source.name)
|
|
mirror_payload = build_site_payload_from_data_dir(mirror_dir, station_count=2)
|
|
mirror_serialized = json.dumps(
|
|
mirror_payload,
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
allow_nan=False,
|
|
)
|
|
assert str(mirror_dir) not in mirror_serialized
|
|
assert mirror_serialized == serialized
|
|
|
|
payload_before = deepcopy(payload)
|
|
world: dict = {"unrelated": {"keep": True}}
|
|
first_meta = apply_site_payload_to_world(world, payload)
|
|
first_world = deepcopy(world)
|
|
second_meta = apply_site_payload_to_world(world, payload)
|
|
assert payload == payload_before
|
|
assert world == first_world
|
|
assert first_meta == second_meta
|
|
assert world["unrelated"] == {"keep": True}
|
|
assert len(world["salesOrders"]) == 10
|
|
assert len(world["flexOrders"]) == 10
|
|
|
|
|
|
def test_data_dir_only_payload_rejects_routing_ambiguity_and_aggregates_bom_usage(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import (
|
|
KangniDataMappingError,
|
|
build_site_payload_from_data_dir,
|
|
)
|
|
|
|
routing_dir = tmp_path / "routing-ambiguous"
|
|
_write_data_dir_fixture(routing_dir, order_count=1)
|
|
_write_route_book(routing_dir / "工艺路线.xlsx", {
|
|
"Sheet2": [("OP-001", "0010", "名称A", "装配")],
|
|
"Sheet4": [("OP-001", "0010", "名称B", "装配")],
|
|
})
|
|
with pytest.raises(KangniDataMappingError, match="内容不一致"):
|
|
build_site_payload_from_data_dir(routing_dir)
|
|
|
|
bom_dir = tmp_path / "bom-ambiguous"
|
|
_write_data_dir_fixture(bom_dir, order_count=1)
|
|
row = ("ORDER-001", "OP-001", "MAT-001", 1.0, "物料001")
|
|
_write_bom_book(bom_dir / "BOM.xlsx", {
|
|
"Sheet2": [row],
|
|
"Sheet3": [("ORDER-001", "OP-001", "MAT-001", 2.0, "物料001")],
|
|
})
|
|
payload = build_site_payload_from_data_dir(bom_dir)
|
|
assert payload["flex"]["flexBom"][0]["quantity"] == 3.0
|
|
bom_mapping = payload["meta"]["bomMappings"]["ORDER-001"]
|
|
assert bom_mapping["aggregatedRows"] == 1
|
|
assert bom_mapping["aggregationGroups"][0]["quantity"] == 3.0
|
|
|
|
|
|
def test_routing_exact_sheet_has_priority_over_generic(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import parse_routing_sheet
|
|
|
|
route = tmp_path / "工艺路线.xlsx"
|
|
_write_route_book(route, {
|
|
"ORDER-EXACT": [("OP-EXACT", "0010", "精确工序", "装配")],
|
|
"Sheet2": [("OP-GENERIC", "0010", "通用工序", "装配")],
|
|
})
|
|
diagnostics = {}
|
|
operations = parse_routing_sheet(
|
|
route,
|
|
"ORDER-EXACT",
|
|
std_times={"OP-EXACT": 12.0},
|
|
canonical_operations=[{"operationCode": "OP-GENERIC", "seq": 10}],
|
|
diagnostics=diagnostics,
|
|
)
|
|
assert [row["operationCode"] for row in operations] == ["OP-EXACT"]
|
|
assert operations[0]["stdTimeSource"] == "工时表"
|
|
assert diagnostics["matchMode"] == "exact-sheet"
|
|
assert diagnostics["matchedSheetCandidates"] == ["ORDER-EXACT"]
|
|
|
|
|
|
def test_generic_routing_accepts_identical_candidates_and_records_evidence(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import (
|
|
load_std_route_map,
|
|
load_std_time_map,
|
|
parse_routing_sheet,
|
|
)
|
|
|
|
route = tmp_path / "工艺路线.xlsx"
|
|
shared = [
|
|
("OP-1", "0070", "部装", "部装"),
|
|
("OP-2", "0080", "总装", "装配"),
|
|
]
|
|
_write_route_book(route, {
|
|
"Sheet2": shared,
|
|
"Sheet3": [("OP-X", "0010", "其它", "装配")],
|
|
"Sheet4": shared,
|
|
})
|
|
_write_std_book(tmp_path / "工时.xlsx", {
|
|
"ORDER-A": [
|
|
("OP-1", "0070", "部装", "部装", 27.0),
|
|
("OP-2", "0080", "总装", "装配", 5.0),
|
|
],
|
|
})
|
|
diagnostics = {}
|
|
operations = parse_routing_sheet(
|
|
route,
|
|
"ORDER-A",
|
|
std_times=load_std_time_map(tmp_path)["ORDER-A"],
|
|
canonical_operations=load_std_route_map(tmp_path)["ORDER-A"],
|
|
diagnostics=diagnostics,
|
|
)
|
|
assert [row["operationCode"] for row in operations] == ["OP-1", "OP-2"]
|
|
assert [row["stdTimePerUnit"] for row in operations] == [27.0, 5.0]
|
|
assert diagnostics["matchedSheet"] == "Sheet2"
|
|
assert diagnostics["matchedSheetCandidates"] == ["Sheet2", "Sheet4"]
|
|
|
|
|
|
def test_generic_routing_rejects_missing_or_ambiguous_content(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import KangniDataMappingError, parse_routing_sheet
|
|
|
|
route = tmp_path / "工艺路线.xlsx"
|
|
_write_route_book(route, {
|
|
"Sheet2": [("OP-1", "0010", "名称A", "装配")],
|
|
"Sheet4": [("OP-1", "0010", "名称B", "装配")],
|
|
})
|
|
canonical = [{"operationCode": "OP-1", "seq": 10, "operationName": "名称A", "opType": "装配"}]
|
|
with pytest.raises(KangniDataMappingError, match="内容不一致"):
|
|
parse_routing_sheet(route, "ORDER-A", canonical_operations=canonical)
|
|
with pytest.raises(KangniDataMappingError, match="没有.*一致"):
|
|
parse_routing_sheet(
|
|
route,
|
|
"ORDER-MISSING",
|
|
canonical_operations=[{"operationCode": "OP-NONE", "seq": 10}],
|
|
)
|
|
with pytest.raises(KangniDataMappingError, match="无规范工序序列"):
|
|
parse_routing_sheet(route, "ORDER-NO-CANONICAL")
|
|
|
|
|
|
def test_bom_exact_and_generic_sheets_filter_orders_without_cross_talk(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import parse_bom_sheet
|
|
|
|
bom = tmp_path / "BOM.xlsx"
|
|
_write_bom_book(bom, {
|
|
"ORDER-A": [("ORDER-A", "OP-1", "MAT-A1", 1.0, "物料A1")],
|
|
"Sheet2": [
|
|
("ORDER-B", "OP-B", "MAT-B", 1.0, "物料B"),
|
|
("ORDER-A", "OP-2", "MAT-A2", 2.0, "物料A2"),
|
|
],
|
|
})
|
|
diagnostics = {}
|
|
rows = parse_bom_sheet(bom, "ORDER-A", diagnostics=diagnostics)
|
|
assert [row["materialCode"] for row in rows] == ["MAT-A1", "MAT-A2"]
|
|
assert diagnostics["matchedSheetCandidates"] == ["ORDER-A", "Sheet2"]
|
|
missing = {}
|
|
assert parse_bom_sheet(bom, "ORDER-C", diagnostics=missing) == []
|
|
assert missing["status"] == "missing"
|
|
|
|
|
|
def test_bom_cross_sheet_duplicate_is_evidenced_and_distinct_usage_is_aggregated(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import parse_bom_sheet
|
|
|
|
duplicate = tmp_path / "BOM-duplicate.xlsx"
|
|
row = ("ORDER-A", "OP-1", "MAT-A", 1.0, "物料A")
|
|
_write_bom_book(duplicate, {"Sheet2": [row], "Sheet3": [row]})
|
|
diagnostics = {}
|
|
rows = parse_bom_sheet(duplicate, "ORDER-A", diagnostics=diagnostics)
|
|
assert len(rows) == 1
|
|
assert diagnostics["sourceRowCount"] == 2
|
|
assert diagnostics["deduplicatedRows"] == 1
|
|
|
|
conflict = tmp_path / "BOM-conflict.xlsx"
|
|
_write_bom_book(conflict, {
|
|
"Sheet2": [row],
|
|
"Sheet3": [("ORDER-A", "OP-1", "MAT-A", 2.0, "物料A")],
|
|
})
|
|
aggregated = {}
|
|
rows = parse_bom_sheet(conflict, "ORDER-A", diagnostics=aggregated)
|
|
assert rows[0]["quantity"] == 3.0
|
|
assert aggregated["aggregatedRows"] == 1
|
|
assert aggregated["aggregationGroups"][0]["sourceRows"] == 2
|
|
|
|
|
|
def test_bom_same_sheet_exact_signature_is_deduplicated_and_distinct_usage_is_summed(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import parse_bom_sheet
|
|
|
|
duplicate = tmp_path / "BOM-same-sheet-duplicate.xlsx"
|
|
row = ("ORDER-A", "OP-1", "MAT-A", 1.0, "物料A")
|
|
_write_bom_book(duplicate, {"Sheet2": [row, row]})
|
|
diagnostics = {}
|
|
rows = parse_bom_sheet(duplicate, "ORDER-A", diagnostics=diagnostics)
|
|
assert len(rows) == 1
|
|
assert diagnostics["sourceRowCount"] == 2
|
|
assert diagnostics["outputRowCount"] == 1
|
|
assert diagnostics["deduplicatedRows"] == 1
|
|
|
|
quantity_conflict = tmp_path / "BOM-same-sheet-quantity-conflict.xlsx"
|
|
_write_bom_book(quantity_conflict, {
|
|
"Sheet2": [
|
|
row,
|
|
("ORDER-A", "OP-1", "MAT-A", 2.0, "物料A"),
|
|
],
|
|
})
|
|
aggregation = {}
|
|
rows = parse_bom_sheet(quantity_conflict, "ORDER-A", diagnostics=aggregation)
|
|
assert rows[0]["quantity"] == 3.0
|
|
assert rows[0]["isKey"] is True
|
|
assert aggregation["aggregatedRows"] == 1
|
|
|
|
|
|
def test_flex_bom_projection_preserves_same_material_at_different_consume_ops(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import (
|
|
build_site_payload_from_data_dir,
|
|
)
|
|
|
|
data_dir = tmp_path / "consume-op-conflict"
|
|
_write_data_dir_fixture(data_dir, order_count=1)
|
|
_write_bom_book(data_dir / "BOM.xlsx", {
|
|
"ORDER-001": [
|
|
("ORDER-001", "OP-001", "MAT-SHARED", 1.0, "共享物料"),
|
|
("ORDER-001", "OP-ALT", "MAT-SHARED", 1.0, "共享物料"),
|
|
],
|
|
})
|
|
payload = build_site_payload_from_data_dir(data_dir)
|
|
projected = payload["flex"]["flexBom"]
|
|
assert {(row["materialCode"], row["consumeOp"]) for row in projected} == {
|
|
("MAT-SHARED", "OP-001"),
|
|
("MAT-SHARED", "OP-ALT"),
|
|
}
|
|
|
|
|
|
def test_flex_bom_projection_cross_order_conflict_and_exact_dedup(tmp_path: Path):
|
|
from openpyxl import load_workbook
|
|
|
|
from server.aps_domain.kangni_intake import (
|
|
KangniDataMappingError,
|
|
build_site_payload_from_data_dir,
|
|
)
|
|
|
|
conflict_dir = tmp_path / "cross-order-conflict"
|
|
order_numbers = _write_data_dir_fixture(conflict_dir, order_count=2)
|
|
wb = load_workbook(conflict_dir / "订单.xlsx")
|
|
ws = wb.active
|
|
ws.cell(row=2, column=3, value="PRODUCT-SHARED")
|
|
ws.cell(row=3, column=3, value="PRODUCT-SHARED")
|
|
ws.cell(row=2, column=4, value="共享产品")
|
|
ws.cell(row=3, column=4, value="共享产品")
|
|
wb.save(conflict_dir / "订单.xlsx")
|
|
shared_route = [("OP-SHARED", "0010", "共享工序", "装配")]
|
|
_write_route_book(conflict_dir / "工艺路线.xlsx", {
|
|
order_no: shared_route for order_no in order_numbers
|
|
})
|
|
_write_std_book(conflict_dir / "工时.xlsx", {
|
|
order_no: [("OP-SHARED", "0010", "共享工序", "装配", 5.0)]
|
|
for order_no in order_numbers
|
|
})
|
|
_write_bom_book(conflict_dir / "BOM.xlsx", {
|
|
order_numbers[0]: [(order_numbers[0], "OP-SHARED", "MAT-SHARED", 1.0, "共享物料")],
|
|
order_numbers[1]: [(order_numbers[1], "OP-SHARED", "MAT-SHARED", 2.0, "共享物料")],
|
|
})
|
|
with pytest.raises(KangniDataMappingError, match="flex BOM 投影冲突"):
|
|
build_site_payload_from_data_dir(conflict_dir)
|
|
|
|
dedup_dir = tmp_path / "cross-order-exact-dedup"
|
|
order_numbers = _write_data_dir_fixture(dedup_dir, order_count=2)
|
|
wb = load_workbook(dedup_dir / "订单.xlsx")
|
|
ws = wb.active
|
|
ws.cell(row=2, column=3, value="PRODUCT-SHARED")
|
|
ws.cell(row=3, column=3, value="PRODUCT-SHARED")
|
|
ws.cell(row=2, column=4, value="共享产品")
|
|
ws.cell(row=3, column=4, value="共享产品")
|
|
wb.save(dedup_dir / "订单.xlsx")
|
|
shared_route = [("OP-SHARED", "0010", "共享工序", "装配")]
|
|
_write_route_book(dedup_dir / "工艺路线.xlsx", {
|
|
order_no: shared_route for order_no in order_numbers
|
|
})
|
|
_write_std_book(dedup_dir / "工时.xlsx", {
|
|
order_no: [("OP-SHARED", "0010", "共享工序", "装配", 5.0)]
|
|
for order_no in order_numbers
|
|
})
|
|
_write_bom_book(dedup_dir / "BOM.xlsx", {
|
|
order_no: [(order_no, "OP-SHARED", "MAT-SHARED", 1.0, "共享物料")]
|
|
for order_no in order_numbers
|
|
})
|
|
payload = build_site_payload_from_data_dir(dedup_dir)
|
|
assert payload["meta"]["bomSourceRowCount"] == 2
|
|
assert payload["meta"]["projectedBomCount"] == 1
|
|
assert payload["flex"]["flexBom"] == [{
|
|
"productCode": "PRODUCT-SHARED",
|
|
"materialCode": "MAT-SHARED",
|
|
"quantity": 1.0,
|
|
"consumeOp": "OP-SHARED",
|
|
"isKey": True,
|
|
}]
|
|
|
|
|
|
def test_flex_material_name_conflict_across_orders_fails_closed(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import (
|
|
KangniDataMappingError,
|
|
build_site_payload_from_data_dir,
|
|
)
|
|
|
|
data_dir = tmp_path / "material-name-conflict"
|
|
order_numbers = _write_data_dir_fixture(data_dir, order_count=2)
|
|
_write_bom_book(data_dir / "BOM.xlsx", {
|
|
order_numbers[0]: [(order_numbers[0], "OP-001", "MAT-SHARED", 1.0, "共享物料A")],
|
|
order_numbers[1]: [(order_numbers[1], "OP-002", "MAT-SHARED", 1.0, "共享物料B")],
|
|
})
|
|
|
|
with pytest.raises(KangniDataMappingError, match="物料编码 MAT-SHARED 名称冲突"):
|
|
build_site_payload_from_data_dir(data_dir)
|
|
|
|
|
|
def test_bom_and_routing_same_generic_name_are_not_bound(tmp_path: Path):
|
|
from server.aps_domain.kangni_intake import parse_bom_sheet, parse_routing_sheet
|
|
|
|
route = tmp_path / "工艺路线.xlsx"
|
|
_write_route_book(route, {
|
|
"Sheet3": [("OP-B", "0010", "B工序", "装配")],
|
|
"Sheet4": [("OP-A", "0010", "A工序", "装配")],
|
|
})
|
|
bom = tmp_path / "BOM.xlsx"
|
|
_write_bom_book(bom, {
|
|
"Sheet3": [("ORDER-A", "OP-A", "MAT-A", 1.0, "物料A")],
|
|
"Sheet4": [("ORDER-B", "OP-B", "MAT-B", 1.0, "物料B")],
|
|
})
|
|
bom_diagnostics = {}
|
|
routing_diagnostics = {}
|
|
assert parse_bom_sheet(bom, "ORDER-A", diagnostics=bom_diagnostics)
|
|
operations = parse_routing_sheet(
|
|
route,
|
|
"ORDER-A",
|
|
canonical_operations=[{
|
|
"operationCode": "OP-A",
|
|
"seq": 10,
|
|
"operationName": "A工序",
|
|
"opType": "装配",
|
|
}],
|
|
diagnostics=routing_diagnostics,
|
|
)
|
|
assert [row["operationCode"] for row in operations] == ["OP-A"]
|
|
assert bom_diagnostics["matchedSheet"] == "Sheet3"
|
|
assert routing_diagnostics["matchedSheet"] == "Sheet4"
|