Skip to content

O
opcua-asyncio
Commit 1c6e5d26 authored by Marcel's avatar Marcel Committed by ORD
Browse files
Options

Move nsrel to xmlimport (#330) 

* move ns relative functionality from parser to importer

* move rel ns handling to xml_import

* also move working version of _sort_nodes_by_parentid to xmlimport

* because moved methods updated test_unit.py
parent c97137f7
Hide whitespace changes
Inline Side-by-side
Showing with 213 additions and 172 deletions
opcua/common/xmlimporter.py
View file @ 1c6e5d26
"""
"""G
add node defined in XML to address space
format is the one from opc-ua specification
"""
import logging
import sys
import re
from opcua import ua
from opcua.common import xmlparser
......@@ -38,6 +38,30 @@ class XmlImporter(object):
self.logger = logging.getLogger(__name__)
self.parser = None
self.server = server
self.namespaces = {}
self.aliases = {}
self._re_nodeid = re.compile(r"^ns=(?P<ns>\d+[^;]*);i=(?P<i>\d+)")
def _map_namespaces(self, namespaces_uris, act_server):
"""
creates a mapping between the namespaces in the xml file and in the server.
if not present the namespace is registered.
"""
namespaces = {}
for ns_index, ns_uri in enumerate(namespaces_uris):
ns_server_index = act_server.register_namespace(ns_uri)
namespaces[ns_index + 1] = (ns_server_index, ns_uri)
self.logger.info("namespace offset", ns_index + 1, (ns_server_index, ns_uri))
return namespaces
def _map_aliases(self, aliases):
"""
maps the import aliases to the correct namespaces
"""
aliases_mapped = {}
for alias, node_id in aliases.items():
aliases_mapped[alias] = self._get_node_id(node_id)
return aliases_mapped
def import_xml(self, xmlpath, act_server):
"""
......@@ -45,8 +69,17 @@ class XmlImporter(object):
"""
self.logger.info("Importing XML file %s", xmlpath)
self.parser = xmlparser.XMLParser(xmlpath, act_server)
self.namespaces = self._map_namespaces(self.parser.get_used_namespaces(), act_server)
self.aliases = self._map_aliases(self.parser.get_aliases())
# The ordering of nodes currently only works if namespaces are
# defined in XML.
# Also, it is recommended not to use node ids without namespace prefix!
nodes_parsed = self._sort_nodes_by_parentid(self.parser)
nodes = []
for nodedata in self.parser:
for nodedata in nodes_parsed: # self.parser:
if nodedata.nodetype == 'UAObject':
node = self.add_object(nodedata)
elif nodedata.nodetype == 'UAObjectType':
......@@ -67,17 +100,59 @@ class XmlImporter(object):
nodes.append(node)
return nodes
def _split_node_id(self, value):
"""
Split the fq node id into namespace and id part.
:returns: (namespace, id)
"""
if not value:
return (None, value)
r_match = self._re_nodeid.search(value)
if r_match:
return r_match.groups()
return (None, value)
def _parse_bname(self, bname):
"""
Parse a browsename and correct the namespace index.
"""
if bname.find(':') != -1:
browse_ns, browse_name = bname.split(':')
if browse_ns:
ns_server = self.namespaces.get(int(browse_ns), None)
if ns_server:
return '%d:%s' % (ns_server[0], browse_name)
return bname
def _get_node_id(self, value):
"""
Check if the nodeid given in the xml model file must be converted
to a already existing namespace id based on the files namespace uri
:returns: NodeId (str)
"""
result = value
node_ns, node_id = self._split_node_id(value)
if node_ns:
ns_server = self.namespaces.get(int(node_ns), None)
if ns_server:
result = "ns={};i={}".format(ns_server[0], node_id)
return result
def _get_node(self, obj):
node = ua.AddNodesItem()
node.RequestedNewNodeId = ua.NodeId.from_string(obj.nodeid)
node.BrowseName = ua.QualifiedName.from_string(obj.browsename)
node.RequestedNewNodeId = ua.NodeId.from_string(self._get_node_id(obj.nodeid))
node.BrowseName = ua.QualifiedName.from_string(self._parse_bname(obj.browsename))
node.NodeClass = getattr(ua.NodeClass, obj.nodetype[2:])
if obj.parent:
node.ParentNodeId = ua.NodeId.from_string(obj.parent)
node.ParentNodeId = ua.NodeId.from_string(self._get_node_id(obj.parent))
if obj.parentlink:
node.ReferenceTypeId = self.to_nodeid(obj.parentlink)
node.ReferenceTypeId = self.to_nodeid(self._get_node_id(obj.parentlink))
if obj.typedef:
node.TypeDefinition = ua.NodeId.from_string(obj.typedef)
node.TypeDefinition = ua.NodeId.from_string(self._get_node_id(obj.typedef))
return node
def to_nodeid(self, nodeid):
......@@ -88,8 +163,8 @@ class XmlImporter(object):
elif hasattr(ua.ObjectIds, nodeid):
return ua.NodeId(getattr(ua.ObjectIds, nodeid))
else:
if nodeid in self.parser.aliases:
nodeid = self.parser.aliases[nodeid]
if nodeid in self.aliases:
nodeid = self.aliases[nodeid]
else:
nodeid = "i={}".format(getattr(ua.ObjectIds, nodeid))
return ua.NodeId.from_string(nodeid)
......@@ -127,7 +202,7 @@ class XmlImporter(object):
attrs.DataType = self.to_nodeid(obj.datatype)
# if obj.value and len(obj.value) == 1:
if obj.value is not None:
attrs.Value = self._add_variable_value(obj, )
attrs.Value = self._add_variable_value(obj,)
if obj.rank:
attrs.ValueRank = obj.rank
if obj.accesslevel:
......@@ -142,7 +217,7 @@ class XmlImporter(object):
res = self.server.add_nodes([node])
self._add_refs(obj)
return res[0].AddedNodeId
def _make_ext_obj(sefl, obj):
ext = getattr(ua, obj.objname)()
for name, val in obj.body:
......@@ -151,7 +226,7 @@ class XmlImporter(object):
else:
for attname, v in val:
# tow possible values:
# either we get value directly
# either we get value directly
# or a dict if it s an object or a list
if type(v) is str:
setattr(ext, attname, to_python(v, ext, attname))
......@@ -272,8 +347,61 @@ class XmlImporter(object):
ref = ua.AddReferencesItem()
ref.IsForward = True
ref.ReferenceTypeId = self.to_nodeid(data.reftype)
ref.SourceNodeId = ua.NodeId.from_string(obj.nodeid)
ref.SourceNodeId = ua.NodeId.from_string(self._get_node_id(obj.nodeid))
ref.TargetNodeClass = ua.NodeClass.DataType
ref.TargetNodeId = ua.NodeId.from_string(data.target)
ref.TargetNodeId = ua.NodeId.from_string(self._get_node_id(data.target))
refs.append(ref)
self.server.add_references(refs)
# FIX: wrong order of node sorting .. need to find out what is wrong
def _sort_nodes_by_parentid(self, nodes):
"""
Sort the list of nodes according theire parent node in order to respect
the depency between nodes.
:param nodes: list of NodeDataObjects
:returns: list of sorted nodes
"""
_nodes = list(nodes)
# list of node ids that are already sorted / inserted
sorted_nodes_ids = []
# list of sorted nodes (i.e. XML Elements)
sorted_nodes = []
# list of namespace indexes that are relevant for this import
# we can only respect ordering nodes for namespaces indexes that
# are defined in the xml file itself. Thus we assume that all other
# references namespaces are already known to the server and should
# not create any dependency problems (like "NodeNotFound")
relevant_namespaces = [str(ns) for ns in self.namespaces.keys()]
while len(_nodes) > 0:
pop_nodes = []
for node in _nodes:
insert = None
# Get the node and parent node namespace and id parts
node_ns, node_id = self._split_node_id(node.nodeid)
parent_ns, parent_id = self._split_node_id(node.parent)
# Insert nodes that
# (1) have no parent / parent_ns is None (e.g. namespace 0)
# (2) ns is not in list of relevant namespaces
if (parent_ns is None or
(len(relevant_namespaces) >= 1 and node_ns not in relevant_namespaces) or
parent_id is None):
insert = 0
else:
# Check if the nodes parent is already in the list of
# inserted nodes
if node.parent in sorted_nodes_ids:
insert = -1
if insert == 0:
sorted_nodes.insert(insert, node)
sorted_nodes_ids.insert(insert, node.nodeid)
pop_nodes.append(node)
elif insert == -1:
sorted_nodes.append(node)
sorted_nodes_ids.append(node.nodeid)
pop_nodes.append(node)
# Remove inserted nodes from the list
for node in pop_nodes:
_nodes.pop(_nodes.index(node))
return sorted_nodes
opcua/common/xmlparser.py
View file @ 1c6e5d26
......@@ -75,38 +75,44 @@ class XMLParser(object):
self.logger = logging.getLogger(__name__)
self._retag = re.compile(r"(\{.*\})(.*)")
self.path = xmlpath
self.aliases = {}
self.tree = ET.parse(xmlpath)
self.root = self.tree.getroot()
self.it = None
self.namespaces = {}
self._re_nodeid = re.compile(r"^ns=(?P<ns>\d+[^;]*);i=(?P<i>\d+)")
def get_used_namespaces(self):
"""
Return the used namespace uris in this import file
"""
namespaces_uris = []
for child in self.root:
name = self._retag.match(child.tag).groups()[1]
if name == 'NamespaceUris':
namespaces_uris = [ns_element.text for ns_element in child]
break
return namespaces_uris
def get_aliases(self):
"""
Return the used node aliases in this import file
"""
aliases = {}
for child in self.root:
name = self._retag.match(child.tag).groups()[1]
if name == 'Aliases':
for el in child:
aliases[el.attrib["Alias"]] = el.text
break
return aliases
def __iter__(self):
nodes = []
for child in self.root:
name = self._retag.match(child.tag).groups()[1]
if name == "Aliases":
for el in child:
self.aliases[el.attrib["Alias"]] = self._get_node_id(el.text)
elif name == 'NamespaceUris':
for ns_index, ns_element in enumerate(child):
ns_uri = ns_element.text
ns_server_index = self.server.register_namespace(ns_uri)
self.namespaces[ns_index + 1] = (ns_server_index, ns_uri)
print("namespace offset", ns_index + 1, (ns_server_index, ns_uri))
else:
if name not in ["Aliases", "NamespaceUris"]:
node = self._parse_node(name, child)
nodes.append(node)
# The ordering of nodes currently only works if namespaces are
# defined in XML.
# Also, it is recommended not to use node ids without namespace prefix!
if self.namespaces:
nodes = self._sort_nodes_by_parentid(nodes)
self.it = iter(nodes)
return self
......@@ -121,101 +127,6 @@ class XMLParser(object):
def next(self): # support for python2
return self.__next__()
def _sort_nodes_by_parentid(self, nodes):
"""
Sort the list of nodes according theire parent node in order to respect
the depency between nodes.
:param nodes: list of NodeDataObjects
:returns: list of sorted nodes
"""
_nodes = list(nodes)
# list of node ids that are already sorted / inserted
sorted_nodes_ids = []
# list of sorted nodes (i.e. XML Elements)
sorted_nodes = []
# list of namespace indexes that are relevant for this import
# we can only respect ordering nodes for namespaces indexes that
# are defined in the xml file itself. Thus we assume that all other
# references namespaces are already known to the server and should
# not create any dependency problems (like "NodeNotFound")
relevant_namespaces = [str(i[0]) for i in self.namespaces.values()]
while len(_nodes) > 0:
pop_nodes = []
for node in _nodes:
insert = None
# Get the node and parent node namespace and id parts
node_ns, node_id = self._split_node_id(node.nodeid)
parent_ns, parent_id = self._split_node_id(node.parent)
# Insert nodes that
# (1) have no parent / parent_ns is None (e.g. namespace 0)
# (2) ns is not in list of relevant namespaces
if (parent_ns is None or node_ns not in relevant_namespaces or
parent_id is None):
insert = 0
else:
# Check if the nodes parent is already in the list of
# inserted nodes
if node.parent in sorted_nodes_ids:
insert = -1
if insert == 0:
sorted_nodes.insert(insert, node)
sorted_nodes_ids.insert(insert, node.nodeid)
pop_nodes.append(node)
elif insert == -1:
sorted_nodes.append(node)
sorted_nodes_ids.append(node.nodeid)
pop_nodes.append(node)
# Remove inserted nodes from the list
for node in pop_nodes:
_nodes.pop(_nodes.index(node))
return sorted_nodes
def _split_node_id(self, value):
"""
Split the fq node id into namespace and id part.
:returns: (namespace, id)
"""
if not value:
return (None, value)
r_match = self._re_nodeid.search(value)
if r_match:
return r_match.groups()
return (None, value)
def _get_node_id(self, value):
"""
Check if the nodeid given in the xml model file must be converted
to a already existing namespace id based on the files namespace uri
:returns: NodeId (str)
"""
result = value
node_ns, node_id = self._split_node_id(value)
if node_ns:
ns_server = self.namespaces.get(int(node_ns), None)
if ns_server:
result = "ns={};i={}".format(ns_server[0], node_id)
return result
def _parse_bname(self, bname):
"""
Parse a browsename and correct the namespace index.
"""
if bname.find(':') != -1:
browse_ns, browse_name = bname.split(':')
if browse_ns:
ns_server = self.namespaces.get(int(browse_ns), None)
if ns_server:
return '%d:%s' % (ns_server[0], browse_name)
return bname
def _parse_node(self, name, child):
"""
Parse a XML node and create a NodeData object.
......@@ -232,14 +143,13 @@ class XMLParser(object):
def _set_attr(self, key, val, obj):
if key == "NodeId":
obj.nodeid = self._get_node_id(val)
print("PARSING", obj.nodeid)
obj.nodeid = val
elif key == "BrowseName":
obj.browsename = self._parse_bname(val)
obj.browsename = val
elif key == "SymbolicName":
obj.symname = val
elif key == "ParentNodeId":
obj.parent = self._get_node_id(val)
obj.parent = val
elif key == "DataType":
obj.datatype = val
elif key == "IsAbstract":
......@@ -373,16 +283,16 @@ class XMLParser(object):
def _parse_refs(self, el, obj):
for ref in el:
if ref.attrib["ReferenceType"] == "HasTypeDefinition":
obj.typedef = self._get_node_id(ref.text)
obj.typedef = ref.text
elif "IsForward" in ref.attrib and ref.attrib["IsForward"] in ("false", "False"):
# if obj.parent:
# sys.stderr.write("Parent is already set with: "+ obj.parent + " " + ref.text + "\n")
obj.parent = self._get_node_id(ref.text)
obj.parent = ref.text
obj.parentlink = ref.attrib["ReferenceType"]
else:
struct = RefStruct()
if "IsForward" in ref.attrib:
struct.forward = ref.attrib["IsForward"]
struct.target = self._get_node_id(ref.text)
struct.target = ref.text
struct.reftype = ref.attrib["ReferenceType"]
obj.refs.append(struct)
tests/custom_nodes.xml
View file @ 1c6e5d26
......@@ -11,7 +11,7 @@
<Alias Alias="MyEnumVal">ns=1;i=3011</Alias>
</Aliases>
<UAObject NodeId="i=30001" BrowseName="MyXMLFolder" >
<UAObject NodeId="ns=1;i=30001" BrowseName="1:MyXMLFolder" >
<Description>A custom folder.</Description>
<References>
<Reference ReferenceType="HasTypeDefinition">i=61</Reference>
......@@ -20,40 +20,40 @@
</UAObject>
<UAObject NodeId="i=30002" BrowseName="MyXMLObject">
<UAObject NodeId="ns=1;i=30002" BrowseName="1:MyXMLObject">
<Description>A custom object node.</Description>
<References>
<Reference ReferenceType="HasTypeDefinition">i=58</Reference>
<Reference ReferenceType="Organizes" IsForward="false">i=30001</Reference>
<Reference ReferenceType="Organizes" IsForward="false">ns=1;i=30001</Reference>
</References>
</UAObject>
<UAVariable NodeId="i=30004" BrowseName="MyXMLVariable" DataType="String">
<UAVariable NodeId="ns=1;i=30004" BrowseName="1:MyXMLVariable" DataType="String">
<References>
<Reference ReferenceType="HasTypeDefinition">i=69</Reference>
<Reference ReferenceType="Organizes" IsForward="false">i=30002</Reference>
<Reference ReferenceType="Organizes" IsForward="false">ns=1;i=30002</Reference>
</References>
<Value>
<String>StringValue</String>
</Value>
</UAVariable>
<UAVariable NodeId="i=30005" BrowseName="MyXMLProperty" DataType="UInt32">
<UAVariable NodeId="ns=1;i=30005" BrowseName="1:MyXMLProperty" DataType="UInt32">
<References>
<Reference ReferenceType="HasTypeDefinition">i=68</Reference>
<Reference ReferenceType="HasProperty" IsForward="false">i=30002</Reference>
<Reference ReferenceType="HasProperty" IsForward="false">ns=1;i=30002</Reference>
</References>
<Value>
<UInt32>76</UInt32>
</Value>
</UAVariable>
<UAVariable NodeId="i=30006" BrowseName="MyXMLVariableWithoutValue" DataType="String">
<UAVariable NodeId="ns=1;i=30006" BrowseName="1:MyXMLVariableWithoutValue" DataType="String">
<References>
<Reference ReferenceType="HasTypeDefinition">i=69</Reference>
<Reference ReferenceType="Organizes" IsForward="false">i=30002</Reference>
<Reference ReferenceType="Organizes" IsForward="false">ns=1;i=30002</Reference>
</References>
</UAVariable>
......@@ -64,10 +64,10 @@
</References>
</UADataType>
<UAVariable NodeId="i=30007" BrowseName="MyCustomTypeVar" DataType="MyCustomString">
<UAVariable NodeId="ns=1;i=30007" BrowseName="1:MyCustomTypeVar" DataType="MyCustomString">
<References>
<Reference ReferenceType="HasTypeDefinition">i=69</Reference>
<Reference ReferenceType="Organizes" IsForward="false">i=30002</Reference>
<Reference ReferenceType="Organizes" IsForward="false">ns=1;i=30002</Reference>
</References>
</UAVariable>
......
tests/tests_unit.py
View file @ 1c6e5d26
......@@ -13,7 +13,7 @@ from opcua.ua.uatypes import flatten, get_shape, reshape
from opcua.server.internal_subscription import WhereClauseEvaluator
from opcua.common.event_objects import BaseEvent
from opcua.common.ua_utils import string_to_variant, variant_to_string, string_to_val, val_to_string
from opcua.common.xmlparser import XMLParser
from opcua.common.xmlimporter import XmlImporter
from opcua.ua.uatypes import _MaskEnum
......@@ -46,7 +46,7 @@ class TestUnit(unittest.TestCase):
s_datetime = "2014-05-3"
self.assertEqual(val_to_string(arr_string), s_arr_datetime)
self.assertEqual(string_to_val(s_arr_datetime, ua.VariantType.String), arr_string)
self.assertEqual(string_to_val(s_arr_datetime, ua.VariantType.DateTime), arr_datetime )
self.assertEqual(string_to_val(s_arr_datetime, ua.VariantType.DateTime), arr_datetime)
def test_string_to_variant_nodeid(self):
s_arr_nodeid = "[ns=2;i=56, i=45]"
......@@ -69,19 +69,19 @@ class TestUnit(unittest.TestCase):
def test_string_to_variant_localized_text(self):
string = "_This is my string"
#string = "_This is my nøåæ"FIXME: does not work with python2 ?!?!
# string = "_This is my nøåæ"FIXME: does not work with python2 ?!?!
obj = ua.LocalizedText(string)
self.assertEqual(string_to_val(string, ua.VariantType.LocalizedText), obj)
self.assertEqual(val_to_string(obj), string)
def test_variant_dimensions(self):
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]]]
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 = ua.Variant(l)
self.assertEqual(v.Dimensions, [2, 3, 4])
v2 = ua.Variant.from_binary(ua.utils.Buffer(v.to_binary()))
self.assertEqual(v, v2)
self.assertEqual(v.Dimensions, v2.Dimensions)
# very special case
l = [[[], [], []], [[], [], []]]
v = ua.Variant(l, ua.VariantType.UInt32)
......@@ -91,13 +91,13 @@ class TestUnit(unittest.TestCase):
self.assertEqual(v, v2)
def test_flatten(self):
l = [[[1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0]],[[1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0]]]
l = [[[1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0]], [[1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0], [1.0, 1.0, 1.0, 1.0]]]
l2 = flatten(l)
dims = get_shape(l)
self.assertEqual(dims, [2, 3, 4])
self.assertNotEqual(l, l2)
l3 = reshape(l2, (2,3,4))
l3 = reshape(l2, (2, 3, 4))
self.assertEqual(l, l3)
......@@ -178,7 +178,7 @@ class TestUnit(unittest.TestCase):
self.assertNotEqual(nid, ua.NodeId.from_string("i=5; ns=10"))
# not sure the next one is correct...
self.assertEqual(nid, ua.NodeId.from_string("i=45; ns=10; srv=serverid"))
nid1 = ua.NodeId("myid.mynodeid", 7)
self.assertEqual(nid1, ua.NodeId.from_string("ns=7; s=myid.mynodeid"))
with self.assertRaises(ua.UaError):
......@@ -298,7 +298,7 @@ class TestUnit(unittest.TestCase):
def test_strrepr_nodeid(self):
nid = ua.NodeId.from_string('ns=2;s=PLC1.Manufacturer;')
self.assertEqual(nid.to_string(), 'ns=2;s=PLC1.Manufacturer')
#self.assertEqual(repr(nid), 'ns=2;s=PLC1.Manufacturer;')
# self.assertEqual(repr(nid), 'ns=2;s=PLC1.Manufacturer;')
def test_qualified_name(self):
qn = ua.QualifiedName('qname', 2)
......@@ -346,10 +346,10 @@ class TestUnit(unittest.TestCase):
self.assertEqual(v.VariantType, v2.VariantType)
def test_variant_array_dim(self):
v = ua.Variant([1, 2, 3, 4, 5, 6], dimensions = [2, 3])
v = ua.Variant([1, 2, 3, 4, 5, 6], dimensions=[2, 3])
self.assertEqual(v.Value[1], 2)
v2 = ua.Variant.from_binary(ua.utils.Buffer(v.to_binary()))
self.assertEqual(reshape(v.Value, (2,3)), v2.Value)
self.assertEqual(reshape(v.Value, (2, 3)), v2.Value)
self.assertEqual(v.VariantType, v2.VariantType)
self.assertEqual(v.Dimensions, v2.Dimensions)
self.assertEqual(v2.Dimensions, [2, 3])
......@@ -442,29 +442,26 @@ class TestUnit(unittest.TestCase):
def test_xmlparser_get_node_id(self):
server = None
xmlpath = 'tests/custom_nodes.xml'
parser = XMLParser(xmlpath, server)
importer = XmlImporter(server)
res1 = parser._get_node_id('i=1001')
res1 = importer._get_node_id('i=1001')
self.assertEqual(res1, 'i=1001')
res2 = parser._get_node_id('ns=1;i=1001')
res2 = importer._get_node_id('ns=1;i=1001')
self.assertEqual(res2, 'ns=1;i=1001')
parser.namespaces = {1: [3, 'http://someuri.com']}
res3 = parser._get_node_id('ns=1;i=1001')
importer.namespaces = {1: [3, 'http://someuri.com']}
res3 = importer._get_node_id('ns=1;i=1001')
self.assertEqual(res3, 'ns=3;i=1001')
parser.namespaces = {1: [3, 'http://someuri.com']}
res4 = parser._get_node_id('ns=2;i=1001')
importer.namespaces = {1: [3, 'http://someuri.com']}
res4 = importer._get_node_id('ns=2;i=1001')
self.assertEqual(res4, 'ns=2;i=1001')
def test_xmlparser_sort_nodes_by_parentid(self):
NodeMock = namedtuple('NodeMock', 'nodeid parent')
server = None
# We actually dont need this. Thus we just pass the available file
xml_path = 'tests/custom_nodes.xml'
unordered_nodes = [
NodeMock('ns=1;i=1001', None),
......@@ -481,9 +478,9 @@ class TestUnit(unittest.TestCase):
]
namespaces = {'1': (1, 'http://someuri.com')}
parser = XMLParser(xml_path, server)
parser.namespaces = namespaces
res = parser._sort_nodes_by_parentid(unordered_nodes)
importer = XmlImporter(server)
importer.namespaces = namespaces
res = importer._sort_nodes_by_parentid(unordered_nodes)
self.assertEqual(res, ordered_nodes)
......
tests/tests_xml.py
View file @ 1c6e5d26
......@@ -13,7 +13,7 @@ class XmlTests(object):
def test_xml_import(self):
self.srv.import_xml("tests/custom_nodes.xml")
o = self.opc.get_objects_node()
v = o.get_child(["MyXMLFolder", "MyXMLObject", "MyXMLVariable"])
v = o.get_child(["1:MyXMLFolder", "1:MyXMLObject", "1:MyXMLVariable"])
val = v.get_value()
self.assertEqual(val, "StringValue")
......@@ -51,6 +51,12 @@ class XmlTests(object):
v1 = o.get_child(["%d:MyBaseObject" % ns, "%d:MyVar" % ns])
self.assertIsNotNone(v1)
r1 = o2.get_references(refs=ua.ObjectIds.HasComponent)[0]
self.assertEqual(r1.NodeId.NamespaceIndex, ns)
r3 = v1.get_references(refs=ua.ObjectIds.HasComponent)[0]
self.assertEqual(r3.NodeId.NamespaceIndex, ns)
def test_xml_method(self):
o = self.opc.nodes.objects.add_object(2, "xmlexportobj")
m = o.add_method(2, "callme", func, [ua.VariantType.Double, ua.VariantType.String], [ua.VariantType.Float])
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7