opcua: Fixes inconsistencies in _check_path implementation

tests/connectors/opcua/test_opcua_connector.py

@@ -0,0 +1,34 @@
+import logging
+import unittest
+from unittest import mock
+
+from opcua import Node
+from opcua.ua import FourByteNodeId
+
+from thingsboard_gateway.connectors.opcua.opcua_connector import OpcUaConnector
+
+logging.basicConfig(level=logging.ERROR,
+                    format='%(asctime)s - %(levelname)s - %(module)s - %(lineno)d - %(message)s',
+                    datefmt='%Y-%m-%d %H:%M:%S')
+log = logging.getLogger("root")
+
+EMPTY_CONFIG = {'server': {'mapping': [], 'url': 'opc.tcp://localhost:4840'}}
+
+class TestOpcUaConnector(unittest.TestCase):
+    def test_check_path_browsepath_partial(self):
+        connector = OpcUaConnector(None, EMPTY_CONFIG, None)
+        with mock.patch.object(OpcUaConnector, 'get_node_path', return_value='Root\\.Objects\\.Machine'):
+            result = connector._check_path('Cycle\\.Cycle\\Status', Node(nodeid=FourByteNodeId("ns=2;i=1"), server=None))
+            self.assertEqual(result, 'Root\\\\.Objects\\\\.Machine\\\\.Cycle\\\\.Cycle\\\\Status')
+
+    def test_check_path_browsepath_partial_with_path_seperator(self):
+        connector = OpcUaConnector(None, EMPTY_CONFIG, None)
+        with mock.patch.object(OpcUaConnector, 'get_node_path', return_value='Root\\.Objects\\.Machine'):
+            result = connector._check_path('\\.Cycle\\.Cycle\\Status', Node(nodeid=FourByteNodeId("ns=2;i=1"), server=None))
+            self.assertEqual(result, 'Root\\\\.Objects\\\\.Machine\\\\.Cycle\\\\.Cycle\\\\Status')
+
+    def test_check_path_browsepath_full(self):
+        connector = OpcUaConnector(None, EMPTY_CONFIG, None)
+        with mock.patch.object(OpcUaConnector, 'get_node_path', return_value='Root\\.Objects\\.Machine'):
+            result = connector._check_path('Root\\.Objects\\.Machine\\.Cycle\\.Cycle\\Status', Node(nodeid=FourByteNodeId("ns=2;i=1"), server=None))
+            self.assertEqual(result, 'Root\\\\.Objects\\\\.Machine\\\\.Cycle\\\\.Cycle\\\\Status')

thingsboard_gateway/connectors/opcua/opcua_connector.py

@@ -685,13 +685,13 @@ def _check_path(self, config_path, node):
         if re.search(r"^root", config_path.lower()) is None:
             node_path = self.get_node_path(node)
             # node_path = '\\\\.'.join(char.split(":")[1] for char in node.get_path(200000, True))
-            if config_path[-3:] != '\\.':
-                information_path = (node_path + '\\.' + config_path).replace('\\', '\\\\')
+            if config_path[:2] != '\\.':
+                information_path = node_path + '\\.' + config_path
             else:
-                information_path = node_path + config_path.replace('\\', '\\\\')
+                information_path = node_path + config_path
         else:
             information_path = config_path
-        result = information_path[:]
+        result = information_path[:].replace('\\', '\\\\')
         return result
 
     @property