yunmei
f49c0920d4
- 点位字典 CSV schema/加载/自动校验（缺失字段/量纲/重复点号，PRD 5.1） - 协议可插拔只读驱动：OPC UA(S7-1200 适配)/S7/Modbus/称重/能源/模拟 - 周期采集引擎：只读+背压保护+健康度指标（丢失率/P99/可用性） - Kafka 流式上行 + 本地 spool 断点续传（丢失率≤0.02% 保障） - 模板配置外置（gateway.yaml + 点位字典 CSV），换行业零改码 - 18 个单元测试全绿；端到端运行 SLA 达标
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""点位字典加载器测试。"""
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
|
|
from point_dict.loader import load_point_dict_csv
|
|
|
|
GOOD_CSV = """device_id,point_id,name,unit,dataType,sampleRate,qualityCode,opcNode
|
|
CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp
|
|
CLF-01,CLF-01.PRES,炉压,kPa,float,1000,,ns=2;s=CLF.Pres
|
|
"""
|
|
|
|
|
|
class LoaderTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self._tmp = tempfile.mkdtemp()
|
|
self.path = os.path.join(self._tmp, "points.csv")
|
|
|
|
def _write(self, content):
|
|
with open(self.path, "w", encoding="utf-8") as fh:
|
|
fh.write(content)
|
|
return self.path
|
|
|
|
def test_load_valid_csv(self):
|
|
path = self._write(GOOD_CSV)
|
|
pd = load_point_dict_csv(path)
|
|
self.assertEqual(len(pd), 2)
|
|
p0 = pd.points[0]
|
|
self.assertEqual(p0.device_id, "CLF-01")
|
|
self.assertEqual(p0.point_id, "CLF-01.TEMP")
|
|
self.assertEqual(p0.unit, "℃")
|
|
self.assertEqual(p0.data_type, "float")
|
|
self.assertEqual(p0.sample_rate, 1000)
|
|
self.assertTrue(p0.quality_code)
|
|
self.assertEqual(p0.opc_node, "ns=2;s=CLF.Temp")
|
|
self.assertEqual(p0.row_number, 2)
|
|
|
|
def test_quality_code_default_true(self):
|
|
"""qualityCode 列留空时默认 true。"""
|
|
path = self._write(GOOD_CSV)
|
|
pd = load_point_dict_csv(path)
|
|
self.assertTrue(pd.points[1].quality_code)
|
|
|
|
def test_utf8_bom_tolerated(self):
|
|
path = self._write("\ufeff" + GOOD_CSV)
|
|
pd = load_point_dict_csv(path)
|
|
self.assertEqual(len(pd), 2)
|
|
|
|
def test_grouping_by_device(self):
|
|
path = self._write(GOOD_CSV)
|
|
pd = load_point_dict_csv(path)
|
|
grouped = pd.by_device_id()
|
|
self.assertEqual(set(grouped.keys()), {"CLF-01"})
|
|
self.assertEqual(len(grouped["CLF-01"]), 2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|