# encoding: utf-8 from concurrent.futures import Future, TimeoutError import time from datetime import datetime from datetime import timedelta import math from contextlib import contextmanager from opcua import ua from opcua import Node from opcua import uamethod from opcua import instantiate from opcua import copy_node from opcua.common import ua_utils from opcua.common.methods import call_method_full def add_server_methods(srv): @uamethod def func(parent, value): return value * 2 o = srv.get_objects_node() v = o.add_method(ua.NodeId("ServerMethod", 2), ua.QualifiedName('ServerMethod', 2), func, [ua.VariantType.Int64], [ua.VariantType.Int64]) @uamethod def func2(parent, methodname, value): if methodname == "panic": return ua.StatusCode(ua.StatusCodes.BadOutOfMemory) if methodname != "sin": res = ua.CallMethodResult() res.StatusCode = ua.StatusCode(ua.StatusCodes.BadInvalidArgument) res.InputArgumentResults = [ua.StatusCode(ua.StatusCodes.BadNotSupported), ua.StatusCode()] return res return math.sin(value) o = srv.get_objects_node() v = o.add_method(ua.NodeId("ServerMethodArray", 2), ua.QualifiedName('ServerMethodArray', 2), func2, [ua.VariantType.String, ua.VariantType.Int64], [ua.VariantType.Int64]) @uamethod def func3(parent, mylist): return [i * 2 for i in mylist] o = srv.get_objects_node() v = o.add_method(ua.NodeId("ServerMethodArray2", 2), ua.QualifiedName('ServerMethodArray2', 2), func3, [ua.VariantType.Int64], [ua.VariantType.Int64]) @uamethod def func4(parent): return None base_otype= srv.get_node(ua.ObjectIds.BaseObjectType) custom_otype = base_otype.add_object_type(2, 'ObjectWithMethodsType') custom_otype.add_method(2, 'ServerMethodDefault', func4) custom_otype.add_method(2, 'ServerMethodMandatory', func4).set_modelling_rule(True) custom_otype.add_method(2, 'ServerMethodOptional', func4).set_modelling_rule(False) custom_otype.add_method(2, 'ServerMethodNone', func4).set_modelling_rule(None) o.add_object(2, 'ObjectWithMethods', custom_otype) @uamethod def func5(parent): return 1,2,3 o = srv.get_objects_node() v = o.add_method(ua.NodeId("ServerMethodTuple", 2), ua.QualifiedName('ServerMethodTuple', 2), func5, [], [ua.VariantType.Int64, ua.VariantType.Int64, ua.VariantType.Int64]) class CommonTests(object): ''' Tests that will be run twice. Once on server side and once on client side since we have been carefull to have the exact same api on server and client side ''' # jyst to avoid editor warnings opc = None assertEqual = lambda x, y: True assertIn = lambda x, y: True @contextmanager def assertNotRaises(self, exc_type): try: yield None except exc_type: raise self.failureException('{} raised'.format(exc_type.__name__)) def test_find_servers(self): servers = self.opc.find_servers() # FIXME : finish def test_add_node_bad_args(self): obj = self.opc.get_objects_node() with self.assertRaises(TypeError): fold = obj.add_folder(1.2, "kk") with self.assertRaises(TypeError): fold = obj.add_folder(ua.UaError, "khjh") with self.assertRaises(ua.UaError): fold = obj.add_folder("kjk", 1.2) with self.assertRaises(TypeError): fold = obj.add_folder("i=0;s='oooo'", 1.2) with self.assertRaises(ua.UaError): fold = obj.add_folder("i=0;s='oooo'", "tt:oioi") def test_delete_nodes(self): obj = self.opc.get_objects_node() fold = obj.add_folder(2, "FolderToDelete") var = fold.add_variable(2, "VarToDelete", 9.1) childs = fold.get_children() self.assertIn(var, childs) self.opc.delete_nodes([var]) with self.assertRaises(ua.UaStatusCodeError): var.set_value(7.8) with self.assertRaises(ua.UaStatusCodeError): obj.get_child(["2:FolderToDelete", "2:VarToDelete"]) childs = fold.get_children() self.assertNotIn(var, childs) def test_delete_nodes_recursive(self): obj = self.opc.get_objects_node() fold = obj.add_folder(2, "FolderToDeleteR") var = fold.add_variable(2, "VarToDeleteR", 9.1) self.opc.delete_nodes([fold, var]) with self.assertRaises(ua.UaStatusCodeError): var.set_value(7.8) with self.assertRaises(ua.UaStatusCodeError): obj.get_child(["2:FolderToDelete", "2:VarToDelete"]) def test_delete_nodes_recursive2(self): obj = self.opc.get_objects_node() fold = obj.add_folder(2, "FolderToDeleteRoot") nfold = fold mynodes = [] for i in range(7): nfold = fold.add_folder(2, "FolderToDeleteRoot") var = fold.add_variable(2, "VarToDeleteR", 9.1) var = fold.add_property(2, "ProToDeleteR", 9.1) prop = fold.add_property(2, "ProToDeleteR", 9.1) o = fold.add_object(3, "ObjToDeleteR") o_var = o.add_variable(3, "VarToDeleteRR", 9.2) o_prop = o.add_property(3, "PropToDeleteRR", 9.2) mynodes.append(nfold) mynodes.append(var) mynodes.append(prop) mynodes.append(o) mynodes.append(o_var) mynodes.append(o_prop) self.opc.delete_nodes([fold], recursive=True) for node in mynodes: 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()) def test_root(self): root = self.opc.get_root_node() self.assertEqual(ua.QualifiedName('Root', 0), root.get_browse_name()) self.assertEqual(ua.LocalizedText('Root'), root.get_display_name()) nid = ua.NodeId(84, 0) self.assertEqual(nid, root.nodeid) def test_objects(self): objects = self.opc.get_objects_node() self.assertEqual(ua.QualifiedName('Objects', 0), objects.get_browse_name()) nid = ua.NodeId(85, 0) self.assertEqual(nid, objects.nodeid) def test_browse(self): objects = self.opc.get_objects_node() obj = objects.add_object(4, "browsetest") folder = obj.add_folder(4, "folder") prop = obj.add_property(4, "property", 1) prop2 = obj.add_property(4, "property2", 2) var = obj.add_variable(4, "variable", 3) obj2 = obj.add_object(4, "obj") alle = obj.get_children() self.assertTrue(prop in alle) self.assertTrue(prop2 in alle) self.assertTrue(var in alle) self.assertTrue(folder in alle) self.assertFalse(obj in alle) props = obj.get_children(refs=ua.ObjectIds.HasProperty) self.assertTrue(prop in props) self.assertTrue(prop2 in props) self.assertFalse(var in props) self.assertFalse(folder in props) self.assertFalse(obj2 in props) all_vars = obj.get_children(nodeclassmask=ua.NodeClass.Variable) self.assertTrue(prop in all_vars) self.assertTrue(var in all_vars) self.assertFalse(folder in props) self.assertFalse(obj2 in props) all_objs = obj.get_children(nodeclassmask=ua.NodeClass.Object) self.assertTrue(folder in all_objs) self.assertTrue(obj2 in all_objs) self.assertFalse(var in all_objs) def test_browse_references(self): objects = self.opc.get_objects_node() folder = objects.add_folder(4, "folder") childs = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False) self.assertTrue(folder in childs) childs = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Both, includesubtypes=False) self.assertTrue(folder in childs) childs = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertFalse(folder in childs) parents = folder.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertTrue(objects in parents) parents = folder.get_referenced_nodes(refs=ua.ObjectIds.HierarchicalReferences, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertTrue(objects in parents) parent = folder.get_parent() self.assertEqual(parent, objects) def test_browsename_with_spaces(self): o = self.opc.get_objects_node() v = o.add_variable(3, 'BNVariable with spaces and %&+?/', 1.3) v2 = o.get_child("3:BNVariable with spaces and %&+?/") self.assertEqual(v, v2) def test_non_existing_path(self): root = self.opc.get_root_node() with self.assertRaises(ua.UaStatusCodeError): server_time_node = root.get_child(['0:Objects', '0:Server', '0:nonexistingnode']) def test_bad_attribute(self): root = self.opc.get_root_node() with self.assertRaises(ua.UaStatusCodeError): root.set_value(99) def test_get_node_by_nodeid(self): root = self.opc.get_root_node() server_time_node = root.get_child(['0:Objects', '0:Server', '0:ServerStatus', '0:CurrentTime']) correct = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime)) self.assertEqual(server_time_node, correct) def test_datetime_read(self): time_node = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime)) dt = time_node.get_value() utcnow = datetime.utcnow() delta = utcnow - dt self.assertTrue(delta < timedelta(seconds=1)) def test_datetime_write(self): time_node = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_CurrentTime)) now = datetime.utcnow() objects = self.opc.get_objects_node() v1 = objects.add_variable(4, "test_datetime", now) tid = v1.get_value() self.assertEqual(now, tid) def test_variant_array_dim(self): objects = self.opc.get_objects_node() l = [[[1.0, 1.0, 1.0, 1.0], [2.0, 2.0, 2.0, 2.0], [3.0, 3.0, 3.0, 3.0]],[[5.0, 5.0, 5.0, 5.0], [7.0, 8.0, 9.0, 01.0], [1.0, 1.0, 1.0, 1.0]]] v = objects.add_variable(3, 'variableWithDims', l) v.set_array_dimensions([0, 0, 0]) dim = v.get_array_dimensions() self.assertEqual(dim, [0, 0, 0]) v.set_value_rank(0) rank = v.get_value_rank() self.assertEqual(rank, 0) v2 = v.get_value() self.assertEqual(v2, l) dv = v.get_data_value() self.assertEqual(dv.Value.Dimensions, [2,3,4]) l = [[[], [], []], [[], [], []]] variant = ua.Variant(l, ua.VariantType.UInt32) v = objects.add_variable(3, 'variableWithDimsEmpty', variant) v2 = v.get_value() self.assertEqual(v2, l) dv = v.get_data_value() self.assertEqual(dv.Value.Dimensions, [2,3,0]) def test_add_numeric_variable(self): objects = self.opc.get_objects_node() v = objects.add_variable('ns=3;i=888;', '3:numericnodefromstring', 99) nid = ua.NodeId(888, 3) qn = ua.QualifiedName('numericnodefromstring', 3) self.assertEqual(nid, v.nodeid) self.assertEqual(qn, v.get_browse_name()) def test_add_string_variable(self): objects = self.opc.get_objects_node() v = objects.add_variable('ns=3;s=stringid;', '3:stringnodefromstring', [68]) nid = ua.NodeId('stringid', 3) qn = ua.QualifiedName('stringnodefromstring', 3) self.assertEqual(nid, v.nodeid) self.assertEqual(qn, v.get_browse_name()) def test_utf8(self): objects = self.opc.get_objects_node() utf_string = "æøå@%&" bn = ua.QualifiedName(utf_string, 3) nid = ua.NodeId("æølå", 3) val = "æøå" v = objects.add_variable(nid, bn, val) self.assertEqual(nid, v.nodeid) val2 = v.get_value() self.assertEqual(val, val2) bn2 = v.get_browse_name() self.assertEqual(bn, bn2) def test_null_variable(self): objects = self.opc.get_objects_node() var = objects.add_variable(3, 'nullstring', "a string") var.set_value(None) val = var.get_value() self.assertEqual(val, None) var.set_value("") val = var.get_value() self.assertNotEqual(val, None) self.assertEqual(val, "") def test_variable_data_type(self): objects = self.opc.get_objects_node() var = objects.add_variable(3, 'stringfordatatype', "a string") val = var.get_data_type_as_variant_type() self.assertEqual(val, ua.VariantType.String) var = objects.add_variable(3, 'stringarrayfordatatype', ["a", "b"]) val = var.get_data_type_as_variant_type() self.assertEqual(val, ua.VariantType.String) def test_add_string_array_variable(self): objects = self.opc.get_objects_node() v = objects.add_variable('ns=3;s=stringarrayid;', '9:stringarray', ['l', 'b']) nid = ua.NodeId('stringarrayid', 3) qn = ua.QualifiedName('stringarray', 9) self.assertEqual(nid, v.nodeid) self.assertEqual(qn, v.get_browse_name()) val = v.get_value() self.assertEqual(['l', 'b'], val) def test_add_numeric_node(self): objects = self.opc.get_objects_node() nid = ua.NodeId(9999, 3) qn = ua.QualifiedName('AddNodeVar1', 3) v1 = objects.add_variable(nid, qn, 0) self.assertEqual(nid, v1.nodeid) self.assertEqual(qn, v1.get_browse_name()) def test_add_string_node(self): objects = self.opc.get_objects_node() qn = ua.QualifiedName('AddNodeVar2', 3) nid = ua.NodeId('AddNodeVar2Id', 3) v2 = objects.add_variable(nid, qn, 0) self.assertEqual(nid, v2.nodeid) self.assertEqual(qn, v2.get_browse_name()) def test_add_find_node_(self): objects = self.opc.get_objects_node() o = objects.add_object('ns=2;i=101;', '2:AddFindObject') o2 = objects.get_child('2:AddFindObject') self.assertEqual(o, o2) def test_node_path(self): objects = self.opc.get_objects_node() o = objects.add_object('ns=2;i=105;', '2:NodePathObject') root = self.opc.get_root_node() o2 = root.get_child(['0:Objects', '2:NodePathObject']) self.assertEqual(o, o2) def test_add_read_node(self): objects = self.opc.get_objects_node() o = objects.add_object('ns=2;i=102;', '2:AddReadObject') nid = ua.NodeId(102, 2) self.assertEqual(o.nodeid, nid) qn = ua.QualifiedName('AddReadObject', 2) self.assertEqual(o.get_browse_name(), qn) def test_simple_value(self): o = self.opc.get_objects_node() v = o.add_variable(3, 'VariableTestValue', 4.32) val = v.get_value() self.assertEqual(4.32, val) def test_add_exception(self): objects = self.opc.get_objects_node() o = objects.add_object('ns=2;i=103;', '2:AddReadObject') with self.assertRaises(ua.UaStatusCodeError): o2 = objects.add_object('ns=2;i=103;', '2:AddReadObject') def test_negative_value(self): o = self.opc.get_objects_node() v = o.add_variable(3, 'VariableNegativeValue', 4) v.set_value(-4.54) val = v.get_value() self.assertEqual(-4.54, val) def test_read_server_state(self): statenode = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_ServerStatus_State)) state = statenode.get_value() self.assertEqual(state, 0) def test_bad_node(self): bad = self.opc.get_node(ua.NodeId(999, 999)) with self.assertRaises(ua.UaStatusCodeError): bad.get_browse_name() with self.assertRaises(ua.UaStatusCodeError): bad.set_value(89) with self.assertRaises(ua.UaStatusCodeError): bad.add_object(0, "0:myobj") with self.assertRaises(ua.UaStatusCodeError): bad.get_child("0:myobj") def test_value(self): o = self.opc.get_objects_node() var = ua.Variant(1.98, ua.VariantType.Double) v = o.add_variable(3, 'VariableValue', var) val = v.get_value() self.assertEqual(1.98, val) dvar = ua.DataValue(var) dv = v.get_data_value() self.assertEqual(ua.DataValue, type(dv)) self.assertEqual(dvar.Value, dv.Value) self.assertEqual(dvar.Value, var) def test_set_value(self): o = self.opc.get_objects_node() var = ua.Variant(1.98, ua.VariantType.Double) dvar = ua.DataValue(var) v = o.add_variable(3, 'VariableValue', var) v.set_value(var.Value) v1 = v.get_value() self.assertEqual(v1, var.Value) v.set_value(var) v2 = v.get_value() self.assertEqual(v2, var.Value) v.set_data_value(dvar) v3 = v.get_data_value() self.assertEqual(v3.Value, dvar.Value) def test_array_value(self): o = self.opc.get_objects_node() v = o.add_variable(3, 'VariableArrayValue', [1, 2, 3]) val = v.get_value() self.assertEqual([1, 2, 3], val) def test_bool_variable(self): o = self.opc.get_objects_node() v = o.add_variable(3, 'BoolVariable', True) dt = v.get_data_type_as_variant_type() self.assertEqual(dt, ua.VariantType.Boolean) val = v.get_value() self.assertEqual(True, val) v.set_value(False) val = v.get_value() self.assertEqual(False, val) def test_array_size_one_value(self): o = self.opc.get_objects_node() v = o.add_variable(3, 'VariableArrayValue', [1, 2, 3]) v.set_value([1]) val = v.get_value() self.assertEqual([1], val) def test_use_namespace(self): idx_orig = 1 uri = self.opc.get_namespace_array()[idx_orig] idx = self.opc.get_namespace_index(uri) self.assertEqual(idx_orig, idx) root = self.opc.get_root_node() myvar = root.add_variable(idx, 'var_in_custom_namespace', [5]) myid = myvar.nodeid self.assertEqual(idx, myid.NamespaceIndex) def test_method(self): o = self.opc.get_objects_node() m = o.get_child("2:ServerMethod") result = o.call_method("2:ServerMethod", 2.1) self.assertEqual(result, 4.2) with self.assertRaises(ua.UaStatusCodeError): # FIXME: we should raise a more precise exception result = o.call_method("2:ServerMethod", 2.1, 89, 9) with self.assertRaises(ua.UaStatusCodeError): result = o.call_method(ua.NodeId(999), 2.1) # non existing method def test_method_array(self): o = self.opc.get_objects_node() m = o.get_child("2:ServerMethodArray") result = o.call_method(m, "sin", ua.Variant(math.pi)) self.assertTrue(result < 0.01) with self.assertRaises(ua.UaStatusCodeError) as cm: result = o.call_method(m, "cos", ua.Variant(math.pi)) self.assertEqual(cm.exception.code, ua.StatusCodes.BadInvalidArgument) with self.assertRaises(ua.UaStatusCodeError) as cm: result = o.call_method(m, "panic", ua.Variant(math.pi)) self.assertEqual(cm.exception.code, ua.StatusCodes.BadOutOfMemory) def test_method_array2(self): o = self.opc.get_objects_node() m = o.get_child("2:ServerMethodArray2") result = o.call_method(m, [1.1, 3.4, 9]) self.assertEqual(result, [2.2, 6.8, 18]) result = call_method_full(o, m, [1.1, 3.4, 9]) self.assertEqual(result.OutputArguments, [[2.2, 6.8, 18]]) def test_method_tuple(self): o = self.opc.get_objects_node() m = o.get_child("2:ServerMethodTuple") result = o.call_method(m) self.assertEqual(result, [1, 2, 3]) result = call_method_full(o, m) self.assertEqual(result.OutputArguments, [1, 2, 3]) def test_method_none(self): # this test calls the function linked to the type's method.. o = self.opc.get_node(ua.ObjectIds.BaseObjectType).get_child("2:ObjectWithMethodsType") m = o.get_child("2:ServerMethodDefault") result = o.call_method(m) self.assertEqual(result, None) result = call_method_full(o, m) self.assertEqual(result.OutputArguments, []) def test_add_nodes(self): objects = self.opc.get_objects_node() f = objects.add_folder(3, 'MyFolder') child = objects.get_child("3:MyFolder") self.assertEqual(child, f) o = f.add_object(3, 'MyObject') child = f.get_child("3:MyObject") self.assertEqual(child, o) v = f.add_variable(3, 'MyVariable', 6) child = f.get_child("3:MyVariable") self.assertEqual(child, v) p = f.add_property(3, 'MyProperty', 10) child = f.get_child("3:MyProperty") self.assertEqual(child, p) childs = f.get_children() self.assertTrue(o in childs) self.assertTrue(v in childs) self.assertTrue(p in childs) def test_modelling_rules(self): obj = self.opc.nodes.base_object_type.add_object_type(2, 'MyFooObjectType') v = obj.add_variable(2, "myvar", 1.1) v.set_modelling_rule(True) p = obj.add_property(2, "myvar", 1.1) p.set_modelling_rule(False) refs = obj.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) self.assertEqual(len(refs), 0) refs = v.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) self.assertEqual(refs[0], self.opc.get_node(ua.ObjectIds.ModellingRule_Mandatory)) refs = p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) self.assertEqual(refs[0], self.opc.get_node(ua.ObjectIds.ModellingRule_Optional)) p.set_modelling_rule(None) refs = p.get_referenced_nodes(refs=ua.ObjectIds.HasModellingRule) self.assertEqual(len(refs), 0) def test_incl_subtypes(self): base_type = self.opc.get_root_node().get_child(["0:Types", "0:ObjectTypes", "0:BaseObjectType"]) descs = base_type.get_children_descriptions(includesubtypes=True) self.assertTrue(len(descs) > 10) descs = base_type.get_children_descriptions(includesubtypes=False) self.assertEqual(len(descs), 0) def test_add_node_with_type(self): objects = self.opc.get_objects_node() f = objects.add_folder(3, 'MyFolder_TypeTest') o = f.add_object(3, 'MyObject1', ua.ObjectIds.BaseObjectType) self.assertEqual(o.get_type_definition().Identifier, ua.ObjectIds.BaseObjectType) o = f.add_object(3, 'MyObject2', ua.NodeId(ua.ObjectIds.BaseObjectType, 0)) self.assertEqual(o.get_type_definition().Identifier, ua.ObjectIds.BaseObjectType) base_otype= self.opc.get_node(ua.ObjectIds.BaseObjectType) custom_otype = base_otype.add_object_type(2, 'MyFooObjectType') o = f.add_object(3, 'MyObject3', custom_otype.nodeid) self.assertEqual(o.get_type_definition().Identifier, custom_otype.nodeid.Identifier) references = o.get_references(refs=ua.ObjectIds.HasTypeDefinition, direction=ua.BrowseDirection.Forward) self.assertEqual(len(references), 1) self.assertEqual(references[0].NodeId, custom_otype.nodeid) def test_references_for_added_nodes(self): objects = self.opc.get_objects_node() o = objects.add_object(3, 'MyObject') nodes = objects.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Forward, includesubtypes=False) self.assertTrue(o in nodes) nodes = o.get_referenced_nodes(refs=ua.ObjectIds.Organizes, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertTrue(objects in nodes) self.assertEqual(o.get_parent(), objects) self.assertEqual(o.get_type_definition().Identifier, ua.ObjectIds.BaseObjectType) self.assertEqual(o.get_references(ua.ObjectIds.HasModellingRule), []) o2 = o.add_object(3, 'MySecondObject') nodes = o.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Forward, includesubtypes=False) self.assertTrue(o2 in nodes) nodes = o2.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertTrue(o in nodes) self.assertEqual(o2.get_parent(), o) self.assertEqual(o2.get_type_definition().Identifier, ua.ObjectIds.BaseObjectType) self.assertEqual(o2.get_references(ua.ObjectIds.HasModellingRule), []) v = o.add_variable(3, 'MyVariable', 6) nodes = o.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Forward, includesubtypes=False) self.assertTrue(v in nodes) nodes = v.get_referenced_nodes(refs=ua.ObjectIds.HasComponent, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertTrue(o in nodes) self.assertEqual(v.get_parent(), o) self.assertEqual(v.get_type_definition().Identifier, ua.ObjectIds.BaseDataVariableType) self.assertEqual(v.get_references(ua.ObjectIds.HasModellingRule), []) p = o.add_property(3, 'MyProperty', 2) nodes = o.get_referenced_nodes(refs=ua.ObjectIds.HasProperty, direction=ua.BrowseDirection.Forward, includesubtypes=False) self.assertTrue(p in nodes) nodes = p.get_referenced_nodes(refs=ua.ObjectIds.HasProperty, direction=ua.BrowseDirection.Inverse, includesubtypes=False) self.assertTrue(o in nodes) self.assertEqual(p.get_parent(), o) self.assertEqual(p.get_type_definition().Identifier, ua.ObjectIds.PropertyType) self.assertEqual(p.get_references(ua.ObjectIds.HasModellingRule), []) m = objects.get_child("2:ServerMethod") self.assertEqual(m.get_references(ua.ObjectIds.HasModellingRule), []) def test_path_string(self): o = self.opc.nodes.objects.add_folder(1, "titif").add_object(3, "opath") path = o.get_path(as_string=True) self.assertEqual(["0:Root", "0:Objects", "1:titif", "3:opath"], path) path = o.get_path(2, as_string=True) self.assertEqual(["1:titif", "3:opath"], path) def test_path(self): of = self.opc.nodes.objects.add_folder(1, "titif") op = of.add_object(3, "opath") path = op.get_path() self.assertEqual([self.opc.nodes.root, self.opc.nodes.objects, of, op], path) path = op.get_path(2) self.assertEqual([of, op], path) target = self.opc.get_node("i=13387") path = target.get_path() self.assertEqual([self.opc.nodes.root, self.opc.nodes.types, self.opc.nodes.object_types, self.opc.nodes.base_object_type, self.opc.nodes.folder_type, self.opc.get_node(ua.ObjectIds.FileDirectoryType), target], path) def test_get_endpoints(self): endpoints = self.opc.get_endpoints() self.assertTrue(len(endpoints) > 0) self.assertTrue(endpoints[0].EndpointUrl.startswith("opc.tcp://")) def test_copy_node(self): dev_t = self.opc.nodes.base_data_type.add_object_type(0, "MyDevice") v_t = dev_t.add_variable(0, "sensor", 1.0) p_t = dev_t.add_property(0, "sensor_id", "0340") ctrl_t = dev_t.add_object(0, "controller") prop_t = ctrl_t.add_property(0, "state", "Running") # Create device sutype devd_t = dev_t.add_object_type(0, "MyDeviceDervived") v_t = devd_t.add_variable(0, "childparam", 1.0) p_t = devd_t.add_property(0, "sensorx_id", "0340") nodes = copy_node(self.opc.nodes.objects, dev_t) mydevice = nodes[0] self.assertEqual(mydevice.get_node_class(), ua.NodeClass.ObjectType) self.assertEqual(len(mydevice.get_children()), 4) obj = mydevice.get_child(["0:controller"]) prop = mydevice.get_child(["0:controller", "0:state"]) self.assertEqual(prop.get_type_definition().Identifier, ua.ObjectIds.PropertyType) self.assertEqual(prop.get_value(), "Running") self.assertNotEqual(prop.nodeid, prop_t.nodeid) def test_instantiate_1(self): # Create device type dev_t = self.opc.nodes.base_object_type.add_object_type(0, "MyDevice") v_t = dev_t.add_variable(0, "sensor", 1.0) v_t.set_modelling_rule(True) p_t = dev_t.add_property(0, "sensor_id", "0340") p_t.set_modelling_rule(True) ctrl_t = dev_t.add_object(0, "controller") ctrl_t.set_modelling_rule(True) v_opt_t = dev_t.add_variable(0, "vendor", 1.0) v_opt_t.set_modelling_rule(False) v_none_t = dev_t.add_variable(0, "model", 1.0) v_none_t.set_modelling_rule(None) prop_t = ctrl_t.add_property(0, "state", "Running") prop_t.set_modelling_rule(True) # Create device sutype devd_t = dev_t.add_object_type(0, "MyDeviceDervived") v_t = devd_t.add_variable(0, "childparam", 1.0) v_t.set_modelling_rule(True) p_t = devd_t.add_property(0, "sensorx_id", "0340") p_t.set_modelling_rule(True) # instanciate device nodes = instantiate(self.opc.nodes.objects, dev_t, bname="2:Device0001") mydevice = nodes[0] self.assertEqual(mydevice.get_node_class(), ua.NodeClass.Object) self.assertEqual(mydevice.get_type_definition(), dev_t.nodeid) obj = mydevice.get_child(["0:controller"]) prop = mydevice.get_child(["0:controller", "0:state"]) with self.assertRaises(ua.UaError): mydevice.get_child(["0:controller", "0:vendor"]) with self.assertRaises(ua.UaError): mydevice.get_child(["0:controller", "0:model"]) self.assertEqual(prop.get_type_definition().Identifier, ua.ObjectIds.PropertyType) self.assertEqual(prop.get_value(), "Running") self.assertNotEqual(prop.nodeid, prop_t.nodeid) # instanciate device subtype nodes = instantiate(self.opc.nodes.objects, devd_t, bname="2:Device0002") mydevicederived = nodes[0] prop1 = mydevicederived.get_child(["0:sensorx_id"]) var1 = mydevicederived.get_child(["0:childparam"]) var_parent = mydevicederived.get_child(["0:sensor"]) prop_parent = mydevicederived.get_child(["0:sensor_id"]) def test_instantiate_string_nodeid(self): # Create device type dev_t = self.opc.nodes.base_object_type.add_object_type(0, "MyDevice2") v_t = dev_t.add_variable(0, "sensor", 1.0) v_t.set_modelling_rule(True) p_t = dev_t.add_property(0, "sensor_id", "0340") p_t.set_modelling_rule(True) ctrl_t = dev_t.add_object(0, "controller") ctrl_t.set_modelling_rule(True) prop_t = ctrl_t.add_property(0, "state", "Running") prop_t.set_modelling_rule(True) # instanciate device nodes = instantiate(self.opc.nodes.objects, dev_t, nodeid=ua.NodeId("InstDevice", 2, ua.NodeIdType.String), bname="2:InstDevice") mydevice = nodes[0] self.assertEqual(mydevice.get_node_class(), ua.NodeClass.Object) self.assertEqual(mydevice.get_type_definition(), dev_t.nodeid) obj = mydevice.get_child(["0:controller"]) obj_nodeid_ident = obj.nodeid.Identifier prop = mydevice.get_child(["0:controller", "0:state"]) self.assertEqual(obj_nodeid_ident, "InstDevice.controller") self.assertEqual(prop.get_type_definition().Identifier, ua.ObjectIds.PropertyType) self.assertEqual(prop.get_value(), "Running") self.assertNotEqual(prop.nodeid, prop_t.nodeid) def test_variable_with_datatype(self): v1 = self.opc.nodes.objects.add_variable(3, 'VariableEnumType1', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType)) tp1 = v1.get_data_type() self.assertEqual(ua.NodeId(ua.ObjectIds.ApplicationType), tp1) v2 = self.opc.nodes.objects.add_variable(3, 'VariableEnumType2', ua.ApplicationType.ClientAndServer, datatype=ua.NodeId(ua.ObjectIds.ApplicationType) ) tp2 = v2.get_data_type() self.assertEqual( ua.NodeId(ua.ObjectIds.ApplicationType), tp2) def test_enum(self): # create enum type enums = self.opc.get_root_node().get_child(["0:Types", "0:DataTypes", "0:BaseDataType", "0:Enumeration"]) myenum_type = enums.add_data_type(0, "MyEnum") es = myenum_type.add_variable(0, "EnumStrings", [ua.LocalizedText("String0"), ua.LocalizedText("String1"), ua.LocalizedText("String2")], ua.VariantType.LocalizedText) #es.set_value_rank(1) # instantiate o = self.opc.get_objects_node() myvar = o.add_variable(2, "MyEnumVar", ua.LocalizedText("String1"), datatype=myenum_type.nodeid) #myvar.set_writable(True) # tests self.assertEqual(myvar.get_data_type(), myenum_type.nodeid) myvar.set_value(ua.LocalizedText("String2")) def test_supertypes(self): nint32 = self.opc.get_node(ua.ObjectIds.Int32) node = ua_utils.get_node_supertype(nint32) self.assertEqual(node, self.opc.get_node(ua.ObjectIds.Integer)) nodes = ua_utils.get_node_supertypes(nint32) self.assertEqual(nodes[1], self.opc.get_node(ua.ObjectIds.Number)) self.assertEqual(nodes[0], self.opc.get_node(ua.ObjectIds.Integer)) # test custom dtype = nint32.add_data_type(0, "MyCustomDataType") node = ua_utils.get_node_supertype(dtype) self.assertEqual(node, nint32) dtype2 = dtype.add_data_type(0, "MyCustomDataType2") node = ua_utils.get_node_supertype(dtype2) self.assertEqual(node, dtype) def test_base_data_type(self): nint32 = self.opc.get_node(ua.ObjectIds.Int32) dtype = nint32.add_data_type(0, "MyCustomDataType") dtype2 = dtype.add_data_type(0, "MyCustomDataType2") self.assertEqual(ua_utils.get_base_data_type(dtype), nint32) self.assertEqual(ua_utils.get_base_data_type(dtype2), nint32) ext = self.opc.nodes.objects.add_variable(0, "MyExtensionObject", ua.Argument()) d = ext.get_data_type() d = self.opc.get_node(d) self.assertEqual(ua_utils.get_base_data_type(d), self.opc.get_node(ua.ObjectIds.Structure)) self.assertEqual(ua_utils.data_type_to_variant_type(d), ua.VariantType.ExtensionObject) def test_data_type_to_variant_type(self): test_data = { ua.ObjectIds.Boolean: ua.VariantType.Boolean, ua.ObjectIds.Byte: ua.VariantType.Byte, ua.ObjectIds.String: ua.VariantType.String, ua.ObjectIds.Int32: ua.VariantType.Int32, ua.ObjectIds.UInt32: ua.VariantType.UInt32, ua.ObjectIds.NodeId: ua.VariantType.NodeId, ua.ObjectIds.LocalizedText: ua.VariantType.LocalizedText, ua.ObjectIds.Structure: ua.VariantType.ExtensionObject, ua.ObjectIds.EnumValueType: ua.VariantType.ExtensionObject, ua.ObjectIds.Enumeration: ua.VariantType.Int32, # enumeration ua.ObjectIds.AttributeWriteMask: ua.VariantType.UInt32, ua.ObjectIds.AxisScaleEnumeration: ua.VariantType.Int32 # enumeration } for dt, vdt in test_data.items(): self.assertEqual(ua_utils.data_type_to_variant_type(self.opc.get_node(ua.NodeId(dt))), vdt)