feat(#63): 点位字典 CSV 导入+自动校验（复用内核 point_dict 校验器，增加 OPC 节点/模板级校验）
This commit is contained in:
@@ -0,0 +1,228 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""点位字典 CSV 导入 + 自动校验测试（issue #63）。
|
||||
|
||||
覆盖：
|
||||
1. 合法 CSV 导入通过（ti / resin 两套模板）；
|
||||
2. 表头校验（缺失列 / 列序错位）；
|
||||
3. 内核校验复用（量纲/数据类型/采样率/重复点号/协议）；
|
||||
4. OPC 节点格式校验（opcua/modbus/空）；
|
||||
5. 模板级量纲收窄（rpm 仅 resin 允许）；
|
||||
6. 报告 ok/汇总/字典化 + 粘贴框入口。
|
||||
"""
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
import _bootstrap # noqa: F401
|
||||
|
||||
from template_console.point_importer import ( # noqa: E402
|
||||
ImportReport,
|
||||
ImportRowIssue,
|
||||
Severity,
|
||||
TemplateKind,
|
||||
import_csv,
|
||||
import_csv_string,
|
||||
)
|
||||
|
||||
GOOD_TI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
|
||||
CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
|
||||
CLF-01,CLF-01.PRES,炉压,kPa,float,1000,true,ns=2;s=CLF.Pres,opcua
|
||||
"""
|
||||
|
||||
GOOD_RESIN = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
|
||||
R-801,R-801.TEMP,反应釜温度,℃,float,1000,true,ns=2;s=R801.Temp,opcua
|
||||
R-801,R-801.AGIT,搅拌转速,rpm,float,1000,true,ns=2;s=R801.Agit,opcua
|
||||
"""
|
||||
|
||||
BAD_MULTI = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol
|
||||
CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,badnode,opcua
|
||||
CLF-01,CLF-01.TEMP,炉压,kPa,badtype,0,true,ns=2;s=CLF.Pres,opcua
|
||||
CLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus
|
||||
"""
|
||||
|
||||
|
||||
class _TmpCsv:
|
||||
"""临时 CSV 文件助手。"""
|
||||
|
||||
def __init__(self, content):
|
||||
self._tmp = tempfile.mkdtemp()
|
||||
self.path = os.path.join(self._tmp, "points.csv")
|
||||
with open(self.path, "w", encoding="utf-8") as fh:
|
||||
fh.write(content)
|
||||
|
||||
def cleanup(self):
|
||||
import shutil
|
||||
shutil.rmtree(self._tmp, ignore_errors=True)
|
||||
|
||||
|
||||
class GoodImportTest(unittest.TestCase):
|
||||
"""合法 CSV 导入。"""
|
||||
|
||||
def test_good_ti_imports_ok(self):
|
||||
f = _TmpCsv(GOOD_TI)
|
||||
try:
|
||||
pd, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertTrue(rep.ok, rep.summary())
|
||||
self.assertEqual(rep.loaded_points, 2)
|
||||
self.assertEqual(rep.error_count, 0)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_good_resin_imports_ok_with_rpm(self):
|
||||
f = _TmpCsv(GOOD_RESIN)
|
||||
try:
|
||||
pd, rep = import_csv(f.path, template=TemplateKind.RESIN)
|
||||
self.assertTrue(rep.ok, rep.summary())
|
||||
# rpm 在 resin 模板合法
|
||||
self.assertEqual(rep.error_count, 0)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_report_summary_and_dict(self):
|
||||
f = _TmpCsv(GOOD_TI)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertIn("通过", rep.summary())
|
||||
d = rep.to_dict()
|
||||
self.assertTrue(d["ok"])
|
||||
self.assertEqual(d["template"], "ti")
|
||||
self.assertEqual(d["loaded_points"], 2)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
|
||||
class HeaderValidationTest(unittest.TestCase):
|
||||
"""表头校验。"""
|
||||
|
||||
def test_missing_column_is_error(self):
|
||||
bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp\n"
|
||||
f = _TmpCsv(bad)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertFalse(rep.ok)
|
||||
codes = [i.code for i in rep.issues if i.row == 1]
|
||||
self.assertIn("missing_column", codes)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_wrong_column_order_is_warn(self):
|
||||
# 列齐全但顺序错（name 提前）→ WARN，不阻断
|
||||
bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,protocol,opcNode\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,opcua,ns=2;s=CLF.Temp\n"
|
||||
f = _TmpCsv(bad)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertIn("bad_column_order", [i.code for i in rep.issues])
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
|
||||
class KernelValidationTest(unittest.TestCase):
|
||||
"""复用内核校验（量纲/数据类型/采样率/重复点号）。"""
|
||||
|
||||
def test_dup_point_detected(self):
|
||||
bad = GOOD_TI + "CLF-01,CLF-01.TEMP,炉温2,℃,float,1000,true,ns=2;s=CLF.Temp2,opcua\n"
|
||||
f = _TmpCsv(bad)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertFalse(rep.ok)
|
||||
self.assertIn("dup_point", [i.code for i in rep.issues])
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_bad_data_type_and_sample_rate(self):
|
||||
f = _TmpCsv(BAD_MULTI)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
codes = [i.code for i in rep.issues]
|
||||
self.assertIn("bad_data_type", codes)
|
||||
self.assertIn("bad_sample_rate", codes)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_bad_protocol_detected(self):
|
||||
bad = "device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,unknownproto\n"
|
||||
f = _TmpCsv(bad)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertIn("bad_protocol", [i.code for i in rep.issues])
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
|
||||
class OpcNodeValidationTest(unittest.TestCase):
|
||||
"""OPC 节点格式校验（配置台扩展维度）。"""
|
||||
|
||||
def test_bad_opcua_node_is_error(self):
|
||||
# BAD_MULTI 第1行 opcNode=badnode 协议 opcua → ERROR
|
||||
f = _TmpCsv(BAD_MULTI)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
opc_issues = [i for i in rep.issues if i.code == "bad_opc_node"]
|
||||
self.assertTrue(any(i.severity == Severity.ERROR for i in opc_issues))
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_valid_modbus_node_ok(self):
|
||||
# BAD_MULTI 第3行 holding:40010 modbus → 不报 bad_opc_node
|
||||
f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.PRES,炉压,kPa,float,500,true,holding:40010,modbus\n")
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
self.assertNotIn("bad_opc_node", [i.code for i in rep.issues
|
||||
if i.severity == Severity.ERROR])
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_empty_opc_node_is_warn(self):
|
||||
f = _TmpCsv("device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode,protocol\nCLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,,simulator\n")
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
empties = [i for i in rep.issues if i.code == "empty_opc_node"]
|
||||
self.assertEqual(len(empties), 1)
|
||||
self.assertEqual(empties[0].severity, Severity.WARN)
|
||||
# 警告不阻断
|
||||
self.assertTrue(rep.ok)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
|
||||
class TemplateUnitTest(unittest.TestCase):
|
||||
"""模板级量纲收窄。"""
|
||||
|
||||
def test_rpm_rejected_in_ti_template(self):
|
||||
f = _TmpCsv(GOOD_RESIN)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.TI)
|
||||
# rpm 是树脂专属，ti 模板应报 template_unit_mismatch
|
||||
self.assertIn("template_unit_mismatch", [i.code for i in rep.issues])
|
||||
self.assertFalse(rep.ok)
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
def test_rpm_allowed_in_resin_template(self):
|
||||
f = _TmpCsv(GOOD_RESIN)
|
||||
try:
|
||||
_, rep = import_csv(f.path, template=TemplateKind.RESIN)
|
||||
self.assertNotIn("template_unit_mismatch", [i.code for i in rep.issues])
|
||||
self.assertTrue(rep.ok, rep.summary())
|
||||
finally:
|
||||
f.cleanup()
|
||||
|
||||
|
||||
class ImportStringTest(unittest.TestCase):
|
||||
"""粘贴框入口（import_csv_string）。"""
|
||||
|
||||
def test_import_from_string(self):
|
||||
pd, rep = import_csv_string(GOOD_TI, template=TemplateKind.TI)
|
||||
self.assertTrue(rep.ok)
|
||||
self.assertEqual(len(pd), 2)
|
||||
|
||||
def test_import_string_bad_csv(self):
|
||||
bad = "device_id,point_id\nCLF-01,CLF-01.TEMP\n" # 缺列
|
||||
_, rep = import_csv_string(bad, template=TemplateKind.TI)
|
||||
self.assertFalse(rep.ok)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user