Files

67 lines
2.2 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,protocol
CLF-01,CLF-01.TEMP,炉温,℃,float,1000,true,ns=2;s=CLF.Temp,opcua
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.protocol, "opcua")
self.assertEqual(p0.row_number, 2)
def test_protocol_blank_defaults_none(self):
"""protocol 列留空时默认 None(走模板 YAML 前缀路由)。"""
path = self._write(GOOD_CSV)
pd = load_point_dict_csv(path)
self.assertIsNone(pd.points[1].protocol)
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()