implement adding and deleting of Node references

opcua/common/node.py

@@ -570,6 +570,52 @@ class Node(object):
         results = opcua.common.manage_nodes.delete_nodes(self.server, [self], recursive, delete_references)
         _check_results(results)
 
+    def delete_reference(self, target, reftype, forward=True, bidirectional=True):
+        """
+        Delete given node's references from address space
+        """
+        known_refs = self.get_references(reftype, includesubtypes=False)
+        targetid = _to_nodeid(target)
+
+        for r in known_refs:
+            if r.NodeId == targetid and r.IsForward == forward:
+                rdesc = r
+                break
+        else:
+            raise ua.UaStatusCodeError(ua.StatusCodes.BadNotFound)
+
+        ditem = ua.DeleteReferencesItem()
+        ditem.SourceNodeId = self.nodeid
+        ditem.TargetNodeId = rdesc.NodeId
+        ditem.ReferenceTypeId = rdesc.ReferenceTypeId
+        ditem.IsForward = rdesc.IsForward
+        ditem.DeleteBidirectional = bidirectional
+
+        self.server.delete_references([ditem])[0].check()
+
+    def add_reference(self, target, reftype, forward=True, bidirectional=True):
+        """
+        Add reference to node
+        """
+
+        aitem = ua.AddReferencesItem()
+        aitem.SourceNodeId = self.nodeid
+        aitem.TargetNodeId = _to_nodeid(target)
+        aitem.ReferenceTypeId = _to_nodeid(reftype)
+        aitem.IsForward = forward
+
+        params = [aitem]
+
+        if bidirectional:
+            aitem2 = ua.AddReferencesItem()
+            aitem2.SourceNodeId = aitem.TargetNodeId
+            aitem2.TargetNodeId = aitem.SourceNodeId
+            aitem2.ReferenceTypeId = aitem.ReferenceTypeId
+            aitem2.IsForward = not forward
+            params.append(aitem2)
+
+        results =  self.server.add_references(params)
+        _check_results(results, len(params))
 
     def add_folder(self, nodeid, bname):
         return  opcua.common.manage_nodes.create_folder(self, nodeid, bname)

tests/tests_common.py

@@ -122,6 +122,59 @@ class CommonTests(object):
             with self.assertRaises(ua.UaStatusCodeError):
                 node.get_browse_name()
 
+    def test_delete_references(self):
+        newtype = self.opc.get_node(ua.ObjectIds.HierarchicalReferences).add_reference_type(0, "HasSuperSecretVariable")
+
+        obj = self.opc.get_objects_node()
+        fold = obj.add_folder(2, "FolderToRef")
+        var = fold.add_variable(2, "VarToRef", 42)
+
+        fold.add_reference(var, newtype)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [fold])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [var])
+
+        fold.delete_reference(var, newtype)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [])
+
+        fold.add_reference(var, newtype, bidirectional=False)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [var])
+
+        fold.delete_reference(var, newtype)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [])
+
+
+        var.add_reference(fold, newtype, forward=False, bidirectional=False)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [fold])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [])
+
+        with self.assertRaises(ua.UaStatusCodeError):
+            fold.delete_reference(var, newtype)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [fold])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [])
+
+        with self.assertRaises(ua.UaStatusCodeError):
+            var.delete_reference(fold, newtype)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [fold])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [])
+
+        var.delete_reference(fold, newtype, forward=False)
+
+        self.assertEqual(var.get_referenced_nodes(newtype), [])
+        self.assertEqual(fold.get_referenced_nodes(newtype), [])
+
+        # clean-up
+        self.opc.delete_nodes([fold, newtype], recursive=True)
+
     def test_server_node(self):
         node = self.opc.get_server_node()
         self.assertEqual(ua.QualifiedName('Server', 0), node.get_browse_name())