use SubElement where possible when creating dom trees

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 - basic_logging_setup only handles sdc logger, no more side effect due to calling logging.basicConfig. 
+- fix possible invalid prefix if QName is a node text.
 - fixed wrong response for SetContextState message. [#287](https://github.com/Draegerwerk/sdc11073/issues/287
 - fixed connection problem when provider closes socket after first request. [#289](https://github.com/Draegerwerk/sdc11073/issues/289
 - change default in ContainerBase.mk_copy to not copy node due to performance problems. [#294](https://github.com/Draegerwerk/sdc11073/issues/294

src/sdc11073/mdib/containerbase.py

@@ -4,7 +4,7 @@
 import inspect
 from typing import Any
 
-from lxml.etree import Element, QName
+from lxml.etree import Element, SubElement, QName
 
 from sdc11073 import observableproperties as properties
 from sdc11073.namespaces import QN_TYPE, NamespaceHelper
@@ -34,18 +34,27 @@ def get_actual_value(self, attr_name: str) -> Any:
         """Ignore default value and implied value, e.g. return None if value is not present in xml."""
         return getattr(self.__class__, attr_name).get_actual_value(self)
 
-    def mk_node(self, tag: QName, ns_helper: NamespaceHelper, set_xsi_type: bool = False) -> xml_utils.LxmlElement:
+    def mk_node(self,
+                tag: QName,
+                ns_helper: NamespaceHelper,
+                parent_node: xml_utils.LxmlElement | None = None,
+                set_xsi_type: bool = False) -> xml_utils.LxmlElement:
         """Create an etree node from instance data.
 
         :param tag: tag of the newly created node
         :param ns_helper: namespaces.NamespaceHelper instance
+        :param parent_node: optional parent node
         :param set_xsi_type: if True, adds Type attribute to node
         :return: etree node
         """
         ns_map = ns_helper.partial_map(ns_helper.PM,
                                        ns_helper.MSG,
                                        ns_helper.XSI)
-        node = Element(tag, nsmap=ns_map)
+        if parent_node is not None:
+            node = SubElement(parent_node, tag, nsmap=ns_map)
+        else:
+            node = Element(tag, nsmap=ns_map)
+
         self.update_node(node, ns_helper, set_xsi_type)
         return node
 

src/sdc11073/mdib/descriptorcontainers.py

@@ -204,17 +204,6 @@ def diff(self, other: AbstractDescriptorContainer, ignore_property_names: list[s
                     ret.append(f'parent_handle={my_value}, other={other_value}')
         return None if len(ret) == 0 else ret
 
-    def mk_descriptor_node(self, tag: etree_.QName,
-                           ns_helper: NamespaceHelper, set_xsi_type: bool = True) -> xml_utils.LxmlElement:
-        """Create a lxml etree node from instance data.
-
-        :param tag: tag of node
-        :param ns_helper: namespaces.DocNamespaceHelper instance
-        :param set_xsi_type: if True, adds Type attribute to node
-        :return: an etree node
-        """
-        return self.mk_node(tag, ns_helper, set_xsi_type)
-
     def tag_name_for_child_descriptor(self, node_type: etree_.QName) -> (etree_.QName, bool):
         """Determine the tag name of a child descriptor.
 

src/sdc11073/mdib/mdibbase.py

@@ -358,33 +358,34 @@ def _reconstruct_md_description(self) -> xml_utils.LxmlElement:
                                              attrib={'DescriptionVersion': str(self.mddescription_version)},
                                              nsmap=doc_nsmap)
         for root_container in root_containers:
-            node = self.make_descriptor_node(root_container, tag=pm.Mds, set_xsi_type=False)
-            md_description_node.append(node)
+            self.make_descriptor_node(root_container, md_description_node, tag=pm.Mds, set_xsi_type=False)
         return md_description_node
 
     def make_descriptor_node(self,
                              descriptor_container: AbstractDescriptorContainer,
+                             parent_node: xml_utils.LxmlElement,
                              tag: etree_.QName,
                              set_xsi_type: bool = True) -> xml_utils.LxmlElement:
         """Create a lxml etree node with subtree from instance data.
 
         :param descriptor_container: a descriptor container instance
+        :param parent_node: parent node
         :param tag: tag of node
         :param set_xsi_type: if true, the NODETYPE will be used to set the xsi:type attribute of the node
         :return: an etree node.
         """
         ns_map = self.nsmapper.partial_map(self.nsmapper.PM, self.nsmapper.XSI) \
             if set_xsi_type else self.nsmapper.partial_map(self.nsmapper.PM)
-        node = etree_.Element(tag,
-                              attrib={'Handle': descriptor_container.Handle},
-                              nsmap=ns_map)
+        node = etree_.SubElement(parent_node,
+                                 tag,
+                                 attrib={'Handle': descriptor_container.Handle},
+                                 nsmap=ns_map)
         descriptor_container.update_node(node, self.nsmapper, set_xsi_type)  # create all
         child_list = self.descriptions.parent_handle.get(descriptor_container.Handle, [])
         # append all child containers, then bring all child elements in correct order
         for child in child_list:
             child_tag, set_xsi = descriptor_container.tag_name_for_child_descriptor(child.NODETYPE)
-            child_node = self.make_descriptor_node(child, child_tag, set_xsi)
-            node.append(child_node)
+            self.make_descriptor_node(child, node, child_tag, set_xsi)
         descriptor_container.sort_child_nodes(node)
         return node
 

src/sdc11073/xml_types/addressing_types.py

@@ -121,9 +121,11 @@ def mk_reply_header_block(self,
         reply_address.Action = action
         return reply_address
 
-    def as_etree_node(self, q_name: QName, ns_map: dict[str, str]) -> xml_utils.LxmlElement:
+    def as_etree_node(self,
+                      q_name: QName, ns_map: dict[str, str],
+                      parent_node: xml_utils.LxmlElement | None = None) -> xml_utils.LxmlElement:
         """Create etree Element form instance data."""
-        node = super().as_etree_node(q_name, ns_map)
+        node = super().as_etree_node(q_name, ns_map, parent_node)
         for param in self.reference_parameters:
             tmp = copy.deepcopy(param)
             tmp.set(_is_reference_parameter, 'true')

src/sdc11073/xml_types/basetypes.py

@@ -35,8 +35,11 @@ def __init__(self):
         for _, prop in self.sorted_container_properties():
             prop.init_instance_data(self)
 
-    def as_etree_node(self, q_name: etree_.QName, ns_map: dict):
-        node = etree_.Element(q_name, nsmap=ns_map)
+    def as_etree_node(self, q_name: etree_.QName, ns_map: dict, parent_node: etree_.Element | None = None):
+        if parent_node is not None:
+            node = etree_.SubElement(parent_node, q_name, nsmap=ns_map)
+        else:
+            node = etree_.Element(q_name, nsmap=ns_map)
         self.update_node(node)
         return node
 

src/sdc11073/xml_types/xml_structure.py

@@ -822,8 +822,7 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
                 sub_node.text = None
         else:
             sub_node = self._get_element_by_child_name(node, self._sub_element_name, create_missing_nodes=True)
-            value = docname_from_qname(py_value, sub_node.nsmap)
-            sub_node.text = value
+            sub_node.text = py_value  # this adds the namesoace to sube_node.nsmap
 
 
 def _compare_extension(left: xml_utils.LxmlElement, right: xml_utils.LxmlElement) -> bool:
@@ -986,12 +985,11 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
                     raise ValueError(f'mandatory value {self._sub_element_name} missing')
                 etree_.SubElement(node, self._sub_element_name, nsmap=node.nsmap)
         else:
-            sub_node = py_value.as_etree_node(self._sub_element_name, node.nsmap)
+            sub_node = py_value.as_etree_node(self._sub_element_name, node.nsmap, node)
             if hasattr(py_value, 'NODETYPE') and hasattr(self.value_class, 'NODETYPE') \
                     and py_value.NODETYPE != self.value_class.NODETYPE:
                 # set xsi type
                 sub_node.set(QN_TYPE, docname_from_qname(py_value.NODETYPE, node.nsmap))
-            node.append(sub_node)
 
 
 class ContainerProperty(_ElementBase):
@@ -1045,11 +1043,10 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
                 etree_.SubElement(node, self._sub_element_name, nsmap=node.nsmap)
         else:
             self.remove_sub_element(node)
-            sub_node = py_value.mk_node(self._sub_element_name, self._ns_helper)
+            sub_node = py_value.mk_node(self._sub_element_name, self._ns_helper, node)
             if py_value.NODETYPE != self.value_class.NODETYPE:
                 # set xsi type
                 sub_node.set(QN_TYPE, docname_from_qname(py_value.NODETYPE, node.nsmap))
-            node.append(sub_node)
 
 
 class _ElementListProperty(_ElementBase, ABC):
@@ -1106,12 +1103,11 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
 
         if py_value is not None:
             for val in py_value:
-                sub_node = val.as_etree_node(self._sub_element_name, node.nsmap)
+                sub_node = val.as_etree_node(self._sub_element_name, node.nsmap, node)
                 if hasattr(val, 'NODETYPE') and hasattr(self.value_class, 'NODETYPE') \
                         and val.NODETYPE != self.value_class.NODETYPE:
                     # set xsi type
                     sub_node.set(QN_TYPE, docname_from_qname(val.NODETYPE, node.nsmap))
-                node.append(sub_node)
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__} datatype {self.value_class.__name__} in subelement {self._sub_element_name}'
@@ -1172,11 +1168,10 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
         # ... and create new ones
         if py_value is not None:
             for val in py_value:
-                sub_node = val.mk_node(self._sub_element_name, self._ns_helper)
+                sub_node = val.mk_node(self._sub_element_name, self._ns_helper, node)
                 if val.NODETYPE != self.value_class.NODETYPE:
                     # set xsi type
                     sub_node.set(QN_TYPE, docname_from_qname(val.NODETYPE, node.nsmap))
-                node.append(sub_node)
 
     def __repr__(self) -> str:
         return f'{self.__class__.__name__} datatype {self.value_class.__name__} in subelement {self._sub_element_name}'
@@ -1269,7 +1264,7 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
         if py_value is None or py_value.is_empty():
             return
         self.remove_sub_element(node)
-        node.append(py_value.as_etree_node(self._sub_element_name, node.nsmap))
+        py_value.as_etree_node(self._sub_element_name, node.nsmap, node)  # creates a sub-node
 
     def __set__(self, instance: Any, py_value: Any):
         if isinstance(py_value, self.value_class):
@@ -1313,7 +1308,7 @@ def update_xml_value(self, instance: Any, node: xml_utils.LxmlElement):
         sub_node.extend(py_value)
 
     def __str__(self) -> str:
-        return f'{self.__class__.__name__} in subelement {self._sub_element_name}'
+        return f'{self.__class__.__name__} in sub-element {self._sub_element_name}'
 
 
 class NodeTextListProperty(_ElementListProperty):

tests/test_mdib.py

@@ -1,6 +1,7 @@
 import os
 import unittest
-
+from dataclasses import dataclass
+from lxml.etree import QName
 from sdc11073.xml_types import pm_qnames as pm
 from sdc11073.exceptions import ApiUsageError
 from sdc11073.mdib import ProviderMdib
@@ -122,3 +123,90 @@ def test_get_mixed_states(self):
             state = mgr.get_state('numeric.ch0.vmd0')
             self.assertEqual(state.DescriptorHandle, 'numeric.ch0.vmd0')
             self.assertRaises(ApiUsageError, mgr.get_state, 'ch0.vmd0')
+
+
+
+    def test_activate_operation_argument(self):
+        """Test that pm:ActivateOperationDescriptor/pm:argument/pm:Arg is handled correctly.
+
+        QName as node text is beyond what xml libraries can handle automatically,
+        it must be handled specifically in sdc11073 code.
+        """
+
+        @dataclass
+        class TestData:
+            mdib_text: str
+            expected_qname: QName
+
+        mdib_dummy = """<msg:GetMdibResponse 
+                        xmlns:msg="http://standards.ieee.org/downloads/11073/11073-10207-2017/message"
+                         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+                         {0}
+                         xmlns:pm="http://standards.ieee.org/downloads/11073/11073-10207-2017/participant"
+                         xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+                         xmlns:ext="http://standards.ieee.org/downloads/11073/11073-10207-2017/extension"
+                         MdibVersion="174" SequenceId="urn:uuid:6f4ff7de-6809-4883-9938-54dd4f6f9173">
+                        <msg:Mdib MdibVersion="174" SequenceId="urn:uuid:6f4ff7de-6809-4883-9938-54dd4f6f9173">
+                            <pm:MdDescription DescriptionVersion="8">
+                                <pm:Mds Handle="mds0" DescriptorVersion="5" SafetyClassification="MedA">
+                                    <pm:Sco Handle="Sco.mds0" DescriptorVersion="2">
+                                        <pm:Operation {1} xsi:type="pm:ActivateOperationDescriptor" Handle="SVO.38.3569" DescriptorVersion="2"
+                                                      SafetyClassification="MedA" OperationTarget="3569" MaxTimeToFinish="PT00H00M02S"
+                                                      Retriggerable="false">
+                                            <pm:Type Code="193821">
+                                                <pm:ConceptDescription Lang="en-US">An operation to cancel global all audio pause
+                                                </pm:ConceptDescription>
+                                            </pm:Type>
+                                            <pm:Argument>
+                                                <pm:ArgName Code="codeForArgumentName"></pm:ArgName>
+                                                <pm:Arg {2}>{3}duration</pm:Arg>
+                                            </pm:Argument>
+                                        </pm:Operation>
+                                    </pm:Sco>
+                                    <pm:SystemContext Handle="SC.mds0" DescriptorVersion="2">
+                                        <ext:Extension>
+                                        </ext:Extension>
+                                        <pm:PatientContext Handle="PC.mds0" DescriptorVersion="2">
+                                        </pm:PatientContext>
+                                        <pm:LocationContext Handle="LC.mds0" DescriptorVersion="2">
+                                        </pm:LocationContext>
+                                    </pm:SystemContext>
+                                </pm:Mds>
+                            </pm:MdDescription>
+                        </msg:Mdib>
+                    </msg:GetMdibResponse>"""
+
+        my_prefix = "my"
+        xsd_prefix = "xsd"
+        delaration = 'xmlns:{0}="http://www.w3.org/2001/XMLSchema"'
+        delaration_any_uri = 'xmlns:{0}="urn:oid:1.23.3.123.2"'
+        expected_qname_xsd = QName("http://www.w3.org/2001/XMLSchema", "duration")
+        expected_qname_any_uri = QName("urn:oid:1.23.3.123.2", "duration")
+
+        mdibs = [TestData(mdib_text=mdib_dummy.format('', '', delaration.format(my_prefix), f"{my_prefix}:"),
+                          expected_qname=expected_qname_xsd),
+                 TestData(mdib_text=mdib_dummy.format('', delaration.format(my_prefix), '', f"{my_prefix}:"),
+                          expected_qname=expected_qname_xsd),
+                 TestData(mdib_text=mdib_dummy.format(delaration.format(my_prefix), '', '', f"{my_prefix}:"),
+                          expected_qname=expected_qname_xsd),
+                 TestData(mdib_text=mdib_dummy.format('', '', delaration.format(xsd_prefix), f"{xsd_prefix}:"),
+                          expected_qname=expected_qname_xsd),
+                 TestData(mdib_text=mdib_dummy.format('', '', 'xmlns="http://www.w3.org/2001/XMLSchema"', ''),
+                          expected_qname=expected_qname_xsd),
+                 TestData(mdib_text=mdib_dummy.format('', '', delaration_any_uri.format(xsd_prefix), f"{xsd_prefix}:"),
+                          expected_qname=expected_qname_any_uri),
+                 TestData(mdib_text=mdib_dummy.format('', delaration_any_uri.format(xsd_prefix), '', f"{xsd_prefix}:"),
+                          expected_qname=expected_qname_any_uri)]
+
+        for test_data in mdibs:
+            # parse mdib data into container and reconstruct mdib data back to a msg:GetMdibResponse
+            # so that it can be validated by xml schema validator
+            mdib_text = test_data.mdib_text.encode('utf-8')
+            mdib_container = ProviderMdib.from_string(mdib_text)
+            mdib_node, mdib_version_group = mdib_container.reconstruct_mdib_with_context_states()
+            arg_nodes = mdib_node.xpath('//*/pm:Arg', namespaces={'pm': "http://standards.ieee.org/downloads/11073/11073-10207-2017/participant"})
+            arg_node = arg_nodes[0]
+            prefix = arg_node.text.split(':')[0]
+            self.assertTrue(prefix in arg_node.nsmap)
+            self.assertEqual(test_data.expected_qname.namespace, arg_node.nsmap[prefix])
+            self.assertEqual(test_data.expected_qname.localname, arg_node.text.split(':')[1])

tests/test_pmtypes.py

@@ -1,6 +1,5 @@
 import unittest
 from unittest import mock
-
 from lxml.etree import QName, fromstring, tostring
 
 from sdc11073.xml_types import pm_types, xml_structure
@@ -68,6 +67,12 @@ def test_activate_operation_descriptor_argument(self):
         arg = pm_types.ActivateOperationDescriptorArgument.from_node(node)
         self.assertEqual(arg.ArgName, pm_types.CodedValue("202890"))
         self.assertEqual(arg.Arg, QName("dummy", "Something"))
+        # verify that as_etree_node -> from_node conversion creates an identical arg
+        node2 = arg.as_etree_node(
+            QName("http://standards.ieee.org/downloads/11073/11073-10207-2017/participant", 'Argument'),
+            ns_map={"pm": "http://standards.ieee.org/downloads/11073/11073-10207-2017/participant"})
+        arg2 = pm_types.ActivateOperationDescriptorArgument.from_node(node2)
+        self.assertEqual(arg, arg2)
 
 
 class TestExtensions(unittest.TestCase):