feat: 完成 issue #25 ① 和利时 DCS OPC 接口参数化适配

2026-08-04 23:34:07 +08:00
parent 8dc379c479
commit 50b51d4a27
5 changed files with 283 additions and 3 deletions
@@ -0,0 +1,139 @@
+# -*- coding: utf-8 -*-
+"""和利时 DCS OPC 驱动参数化适配测试（issue #25）。
+
+覆盖：
+1. 驱动注册表包含 hollysys（from_template 可实例化）；
+2. 参数化提取：ua 模式（endpoint/security/username/password）与
+   da 模式（host/prog_id/cls_id）；
+3. 只读约束：驱动无任何写/控制方法（PRD 9 章零控制指令下发）；
+4. 依赖缺失路径：UA/DA 底层库未安装时 connect 抛出明确 ConnectionError；
+5. 委托路由：connect 后 read_points/close 委托到底层实现。
+"""
+import unittest
+from unittest import mock
+
+from drivers import HollysysDcsDriver, from_template
+from drivers.base import Driver
+from drivers.hollysys_driver import _HollysysDaDriver
+from drivers.opcua_driver import OpcUaDriver
+from point_dict.loader import Point
+
+UA_CONFIG = {
+    "mode": "ua",
+    "endpoint": "opc.tcp://10.20.1.10:4840",
+    "security": "Basic256Sha256",
+    "username": "iaop_read",
+    "password": "secret",
+    "timeout": 10,
+}
+DA_CONFIG = {
+    "mode": "da",
+    "host": "10.20.1.10",
+    "prog_id": "Hollysys.OPCServer",
+    "timeout": 10,
+}
+
+
+def make_point(point_id="CLF-01.TEMP", opc_node="ns=2;s=CLF.Temp"):
+    return Point(
+        device_id="CLF-01", point_id=point_id, name="炉温", unit="℃",
+        data_type="float", sample_rate=1000, quality_code=True,
+        opc_node=opc_node, protocol="hollysys",
+    )
+
+
+class TestRegistry(unittest.TestCase):
+    """注册表与工厂。"""
+
+    def test_registry_contains_hollysys(self):
+        driver = from_template("hollysys", UA_CONFIG)
+        self.assertIsInstance(driver, HollysysDcsDriver)
+
+    def test_driver_is_read_only(self):
+        """只读约束：驱动只暴露 connect/read_points/close，无写方法。"""
+        driver = HollysysDcsDriver(UA_CONFIG)
+        methods = {m for m in dir(driver) if not m.startswith("_")}
+        self.assertIn("read_points", methods)
+        self.assertNotIn("write", methods)
+        self.assertNotIn("write_points", methods)
+        self.assertNotIn("set_value", methods)
+
+
+class TestParameterization(unittest.TestCase):
+    """参数化提取。"""
+
+    def test_ua_params(self):
+        driver = HollysysDcsDriver(UA_CONFIG)
+        self.assertEqual(driver.mode, "ua")
+        ua = OpcUaDriver(UA_CONFIG)
+        self.assertEqual(ua.endpoint, "opc.tcp://10.20.1.10:4840")
+        self.assertEqual(ua.security, "Basic256Sha256")
+        self.assertEqual(ua.username, "iaop_read")
+        self.assertEqual(ua.password, "secret")
+
+    def test_da_params(self):
+        driver = HollysysDcsDriver(DA_CONFIG)
+        self.assertEqual(driver.mode, "da")
+        da = _HollysysDaDriver(DA_CONFIG)
+        self.assertEqual(da.host, "10.20.1.10")
+        self.assertEqual(da.prog_id, "Hollysys.OPCServer")
+        self.assertEqual(_HollysysDaDriver({}).prog_id,
+                         "Hollysys.OPCServer")  # 默认 ProgID
+
+    def test_default_mode_is_ua(self):
+        driver = HollysysDcsDriver({})
+        self.assertEqual(driver.mode, "ua")
+
+
+class TestConnectDependencyErrors(unittest.TestCase):
+    """底层库未安装时的明确报错（不静默失败）。"""
+
+    def test_ua_missing_dependency(self):
+        driver = HollysysDcsDriver(UA_CONFIG)
+        with mock.patch.dict("sys.modules", {"opcua": None}):
+            with self.assertRaises(ConnectionError) as ctx:
+                driver.connect()
+        self.assertIn("opcua", str(ctx.exception))
+
+    def test_da_missing_dependency(self):
+        driver = HollysysDcsDriver(DA_CONFIG)
+        with mock.patch.dict("sys.modules", {"OpenOPC": None}):
+            with self.assertRaises(ConnectionError) as ctx:
+                driver.connect()
+        self.assertIn("OpenOPC", str(ctx.exception))
+
+
+class TestDelegation(unittest.TestCase):
+    """connect 后 read_points/close 委托到底层实现。"""
+
+    def test_ua_delegates_to_opcua_driver(self):
+        driver = HollysysDcsDriver(UA_CONFIG)
+        with mock.patch("drivers.opcua_driver.OpcUaDriver.connect"):
+            driver.connect()
+        self.assertIsInstance(driver._delegate, OpcUaDriver)
+        point = make_point()
+        with mock.patch.object(driver._delegate, "read_points",
+                               return_value={"CLF-01.TEMP": 850.5}) as rd:
+            values = driver.read_points([point])
+        rd.assert_called_once_with([point])
+        self.assertEqual(values["CLF-01.TEMP"], 850.5)
+        with mock.patch.object(driver._delegate, "close"):
+            driver.close()
+        self.assertIsNone(driver._delegate)
+
+    def test_da_delegates_to_da_driver(self):
+        driver = HollysysDcsDriver(DA_CONFIG)
+        fake_openopc = mock.MagicMock()
+        with mock.patch.dict("sys.modules", {"OpenOPC": fake_openopc}):
+            with mock.patch.object(_HollysysDaDriver, "connect"):
+                driver.connect()
+        self.assertIsInstance(driver._delegate, _HollysysDaDriver)
+
+    def test_read_before_connect_raises(self):
+        driver = HollysysDcsDriver(UA_CONFIG)
+        with self.assertRaises(ConnectionError):
+            driver.read_points([make_point()])
+
+
+if __name__ == "__main__":
+    unittest.main()