00001 import uuid
00002 import logging
00003
00004 from opcua import ua
00005 from opcua import uamethod
00006 from opcua.ua import uaerrors
00007
00008
00009 logger = logging.getLogger("opcua.common.xmlimporter")
00010 logger.setLevel(logging.DEBUG)
00011 logger = logging.getLogger("opcua.common.xmlparser")
00012 logger.setLevel(logging.DEBUG)
00013
00014
00015 @uamethod
00016 def func(parent, value, string):
00017 return value * 2
00018
00019
00020 class XmlTests(object):
00021 srv = None
00022 opc = None
00023 assertEqual = dir
00024
00025 def test_xml_import(self):
00026 self.srv.import_xml("tests/custom_nodes.xml")
00027 o = self.opc.get_objects_node()
00028 v = o.get_child(["1:MyXMLFolder", "1:MyXMLObject", "1:MyXMLVariable"])
00029 val = v.get_value()
00030 self.assertEqual(val, "StringValue")
00031
00032 node_path = ["Types", "DataTypes", "BaseDataType", "Enumeration",
00033 "1:MyEnum", "0:EnumStrings"]
00034 o = self.opc.get_root_node().get_child(node_path)
00035 self.assertEqual(len(o.get_value()), 3)
00036
00037
00038 node_path = ["Types", "ObjectTypes", "BaseObjectType",
00039 "1:MyObjectType", "1:MyMethod"]
00040 o = self.opc.get_root_node().get_child(node_path)
00041 self.assertEqual(len(o.get_referenced_nodes()), 4)
00042
00043
00044 node_path = ["Types", "ObjectTypes", "BaseObjectType",
00045 "1:MyObjectType", "1:MyMethod", "InputArguments"]
00046 o = self.opc.get_root_node().get_child(node_path)
00047 input_arg = o.get_data_value().Value.Value[0]
00048 self.assertEqual(input_arg.Name, 'Context')
00049
00050 def test_xml_import_additional_ns(self):
00051 self.srv.register_namespace('http://placeholder.toincrease.nsindex')
00052
00053
00054 self.srv.import_xml("tests/custom_nodesns.xml")
00055
00056 ns = self.srv.get_namespace_index("http://examples.freeopcua.github.io/")
00057 o = self.opc.get_objects_node()
00058
00059 o2 = o.get_child(["{0:d}:MyBaseObject".format(ns)])
00060
00061 self.assertIsNotNone(o2)
00062
00063 v1 = o.get_child(["{0:d}:MyBaseObject".format(ns), "{0:d}:MyVar".format(ns)])
00064 self.assertIsNotNone(v1)
00065
00066 r1 = o2.get_references(refs=ua.ObjectIds.HasComponent)[0]
00067 self.assertEqual(r1.NodeId.NamespaceIndex, ns)
00068
00069 r3 = v1.get_references(refs=ua.ObjectIds.HasComponent)[0]
00070 self.assertEqual(r3.NodeId.NamespaceIndex, ns)
00071
00072 def test_xml_method(self):
00073 self.opc.register_namespace("tititi")
00074 self.opc.register_namespace("whatthefuck")
00075 o = self.opc.nodes.objects.add_object(2, "xmlexportmethod")
00076 m = o.add_method(2, "callme", func, [ua.VariantType.Double, ua.VariantType.String], [ua.VariantType.Float])
00077
00078 inputs = m.get_child("InputArguments")
00079 val = inputs.get_value()
00080 val[0].ArrayDimensions = [2, 2]
00081 desc = b"My nce description"
00082 val[0].Description = ua.LocalizedText(desc)
00083 inputs.set_value(val)
00084
00085
00086 nodes = [o, m]
00087 nodes.extend(m.get_children())
00088 self.opc.export_xml(nodes, "export.xml")
00089
00090 self.opc.delete_nodes(nodes)
00091 self.opc.import_xml("export.xml")
00092
00093
00094 val = inputs.get_value()
00095 self.assertEqual(len(val), 2)
00096
00097 self.assertEqual(val[0].ArrayDimensions, [2, 2])
00098 self.assertEqual(val[0].Description.Text, desc)
00099
00100 def test_xml_vars(self):
00101 self.opc.register_namespace("tititi")
00102 self.opc.register_namespace("whatthexxx")
00103 o = self.opc.nodes.objects.add_object(2, "xmlexportobj")
00104 v = o.add_variable(3, "myxmlvar", 6.78, ua.VariantType.Float)
00105 a = o.add_variable(3, "myxmlvar-array", [6, 1], ua.VariantType.UInt16)
00106 a2 = o.add_variable(3, "myxmlvar-2dim", [[1, 2], [3, 4]], ua.VariantType.UInt32)
00107 a3 = o.add_variable(3, "myxmlvar-2dim", [[]], ua.VariantType.ByteString)
00108
00109 nodes = [o, v, a, a2, a3]
00110 self.opc.export_xml(nodes, "export-vars.xml")
00111 self.opc.delete_nodes(nodes)
00112 self.opc.import_xml("export-vars.xml")
00113
00114 self.assertEqual(v.get_value(), 6.78)
00115 self.assertEqual(v.get_data_type(), ua.NodeId(ua.ObjectIds.Float))
00116
00117 self.assertEqual(a.get_data_type(), ua.NodeId(ua.ObjectIds.UInt16))
00118 self.assertIn(a.get_value_rank(), (0, 1))
00119 self.assertEqual(a.get_value(), [6, 1])
00120
00121 self.assertEqual(a2.get_value(), [[1, 2], [3, 4]])
00122 self.assertEqual(a2.get_data_type(), ua.NodeId(ua.ObjectIds.UInt32))
00123 self.assertIn(a2.get_value_rank(), (0, 2))
00124 self.assertEqual(a2.get_attribute(ua.AttributeIds.ArrayDimensions).Value.Value, [2, 2])
00125
00126 self.assertEqual(a3.get_data_type(), ua.NodeId(ua.ObjectIds.ByteString))
00127 self.assertIn(a3.get_value_rank(), (0, 2))
00128 self.assertEqual(a3.get_attribute(ua.AttributeIds.ArrayDimensions).Value.Value, [1, 0])
00129
00130 def test_xml_ns(self):
00131 """
00132 This test is far too complicated but catches a lot of things...
00133 """
00134 ns_array = self.opc.get_namespace_array()
00135 if len(ns_array) < 3:
00136 self.opc.register_namespace("dummy_ns")
00137 print("ARRAY", self.opc.get_namespace_array())
00138
00139 ref_ns = self.opc.register_namespace("ref_namespace")
00140 new_ns = self.opc.register_namespace("my_new_namespace")
00141 bname_ns = self.opc.register_namespace("bname_namespace")
00142
00143 o = self.opc.nodes.objects.add_object(0, "xmlns0")
00144 o50 = self.opc.nodes.objects.add_object(50, "xmlns20")
00145 o200 = self.opc.nodes.objects.add_object(200, "xmlns200")
00146 onew = self.opc.nodes.objects.add_object(new_ns, "xmlns_new")
00147 vnew = onew.add_variable(new_ns, "xmlns_new_var", 9.99)
00148 o_no_export = self.opc.nodes.objects.add_object(ref_ns, "xmlns_parent")
00149 v_no_parent = o_no_export.add_variable(new_ns, "xmlns_new_var_no_parent", 9.99)
00150 o_bname = onew.add_object("ns={0};i=4000".format(new_ns), "{0}:BNAME".format(bname_ns))
00151
00152 nodes = [o, o50, o200, onew, vnew, v_no_parent, o_bname]
00153 print("CREATED", nodes, o_no_export)
00154 self.opc.export_xml(nodes, "export-ns.xml")
00155
00156 self.opc.delete_nodes(nodes)
00157 ns_node = self.opc.get_node(ua.NodeId(ua.ObjectIds.Server_NamespaceArray))
00158 nss = ns_node.get_value()
00159 nss.remove("my_new_namespace")
00160
00161 nss.remove("bname_namespace")
00162 ns_node.set_value(nss)
00163 new_ns = self.opc.register_namespace("my_new_namespace_offsett")
00164 new_ns = self.opc.register_namespace("my_new_namespace")
00165
00166 new_nodes = self.opc.import_xml("export-ns.xml")
00167 print("NEW NODES", new_nodes)
00168
00169 for i in [o, o50, o200]:
00170 print(i)
00171 i.get_browse_name()
00172 with self.assertRaises(uaerrors.BadNodeIdUnknown):
00173 onew.get_browse_name()
00174
00175
00176 nss = self.opc.get_namespace_array()
00177 self.assertIn("bname_namespace", nss)
00178
00179 new_ns = self.opc.register_namespace("my_new_namespace")
00180 bname_ns = self.opc.register_namespace("bname_namespace")
00181 print("ARRAY 2", self.opc.get_namespace_array())
00182
00183 print("NEW NS", new_ns, onew)
00184 onew.nodeid.NamespaceIndex = new_ns
00185 print("OENE", onew)
00186 onew.get_browse_name()
00187 vnew2 = onew.get_children()[0]
00188 self.assertEqual(new_ns, vnew2.nodeid.NamespaceIndex)
00189
00190 def test_xml_float(self):
00191 o = self.opc.nodes.objects.add_variable(2, "xmlfloat", 5.67)
00192 dtype = o.get_data_type()
00193 dv = o.get_data_value()
00194
00195 self.opc.export_xml([o], "export-float.xml")
00196 self.opc.delete_nodes([o])
00197 new_nodes = self.opc.import_xml("export-float.xml")
00198 o2 = self.opc.get_node(new_nodes[0])
00199
00200 self.assertEqual(o, o2)
00201 self.assertEqual(dtype, o2.get_data_type())
00202 self.assertEqual(dv.Value, o2.get_data_value().Value)
00203
00204 def test_xml_bool(self):
00205 o = self.opc.nodes.objects.add_variable(2, "xmlbool", True)
00206 self._test_xml_var_type(o, "bool")
00207
00208 def test_xml_string(self):
00209 o = self.opc.nodes.objects.add_variable(2, "xmlstring", "mystring")
00210 self._test_xml_var_type(o, "string")
00211
00212 def test_xml_string_array(self):
00213 o = self.opc.nodes.objects.add_variable(2, "xmlstringarray", ["mystring2", "mystring3"])
00214 node2 = self._test_xml_var_type(o, "stringarray")
00215 dv = node2.get_data_value()
00216
00217 def test_xml_guid(self):
00218 o = self.opc.nodes.objects.add_variable(2, "xmlguid", uuid.uuid4())
00219 self._test_xml_var_type(o, "guid")
00220
00221 def test_xml_localizedtext(self):
00222 o = self.opc.nodes.objects.add_variable(2, "xmlltext", ua.LocalizedText("mytext"))
00223 self._test_xml_var_type(o, "localized_text")
00224
00225 def test_xml_localizedtext_array(self):
00226 o = self.opc.nodes.objects.add_variable(2, "xmlltext_array", [ua.LocalizedText("erert"), ua.LocalizedText("erert33")])
00227 self._test_xml_var_type(o, "localized_text_array")
00228
00229 def test_xml_nodeid(self):
00230 o = self.opc.nodes.objects.add_variable(2, "xmlnodeid", ua.NodeId("mytext", 1))
00231 self._test_xml_var_type(o, "nodeid")
00232
00233 def test_xml_ext_obj(self):
00234 arg = ua.Argument()
00235 arg.DataType = ua.NodeId(ua.ObjectIds.Float)
00236 arg.Description = ua.LocalizedText(b"Nice description")
00237 arg.ArrayDimensions = [1, 2, 3]
00238 arg.Name = "MyArg"
00239
00240 node = self.opc.nodes.objects.add_variable(2, "xmlexportobj2", arg)
00241 node2 = self._test_xml_var_type(node, "ext_obj", test_equality=False)
00242 arg2 = node2.get_value()
00243
00244 self.assertEqual(arg.Name, arg2.Name)
00245 self.assertEqual(arg.ArrayDimensions, arg2.ArrayDimensions)
00246 self.assertEqual(arg.Description, arg2.Description)
00247 self.assertEqual(arg.DataType, arg2.DataType)
00248
00249 def test_xml_enum(self):
00250 o = self.opc.nodes.objects.add_variable(2, "xmlenum", 0, varianttype=ua.VariantType.Int32, datatype=ua.ObjectIds.ApplicationType)
00251 self._test_xml_var_type(o, "enum")
00252
00253 def test_xml_enumvalues(self):
00254 o = self.opc.nodes.objects.add_variable(2, "xmlenumvalues", 0, varianttype=ua.VariantType.Int32, datatype=ua.ObjectIds.AttributeWriteMask)
00255 self._test_xml_var_type(o, "enumvalues")
00256
00257 def test_xml_custom_uint32(self):
00258 t = self.opc.create_custom_data_type(2, 'MyCustomUint32', ua.ObjectIds.UInt32)
00259 o = self.opc.nodes.objects.add_variable(2, "xmlcustomunit32", 0, varianttype=ua.VariantType.UInt32, datatype=t.nodeid)
00260 self._test_xml_var_type(o, "cuint32")
00261
00262 def _test_xml_var_type(self, node, typename, test_equality=True):
00263 dtype = node.get_data_type()
00264 dv = node.get_data_value()
00265 rank = node.get_value_rank()
00266 dim = node.get_array_dimensions()
00267 nclass = node.get_node_class()
00268
00269 path = "export-{0}.xml".format(typename)
00270 self.opc.export_xml([node], path)
00271 self.opc.delete_nodes([node])
00272 new_nodes = self.opc.import_xml(path)
00273 node2 = self.opc.get_node(new_nodes[0])
00274
00275 self.assertEqual(node, node2)
00276 self.assertEqual(dtype, node2.get_data_type())
00277 if test_equality:
00278 self.assertEqual(dv.Value, node2.get_data_value().Value)
00279 self.assertEqual(rank, node2.get_value_rank())
00280 self.assertEqual(dim, node2.get_array_dimensions())
00281 self.assertEqual(nclass, node2.get_node_class())
00282 return node2