model.base: Copy Qualifier and Extension data in update_nss_from

sdk/basyx/aas/model/base.py

@@ -1777,6 +1777,25 @@ def name(self, name: NameType) -> None:
         # Redundant to the line above. However, this way, we make sure that we really update the _name
         self._name = name
 
+    def update_from(self, other: "Extension") -> None:
+        """
+        Internal function to update the object's attributes from a different version of the exact same object.
+
+        This function should not be used directly. It is typically used by backend implementations (database adapters,
+        protocol clients, etc.) to update the object's data, after ``update_nss_from()`` has been called.
+
+        :param other: The object to update from
+        """
+        # The name setter rejects the current name as a duplicate, so only assign a changed name
+        if self.name != other.name:
+            self.name = other.name
+        # Assign value_type first so the incoming value is cast against the new type.
+        self.value_type = other.value_type
+        self.value = other.value
+        self.refers_to = set(other.refers_to)
+        self.semantic_id = other.semantic_id
+        self.supplemental_semantic_id = other.supplemental_semantic_id
+
 
 class HasKind(metaclass=abc.ABCMeta):
     """
@@ -1933,6 +1952,26 @@ def type(self, type_: QualifierType) -> None:
         # Redundant to the line above. However, this way, we make sure that we really update the _type
         self._type = type_
 
+    def update_from(self, other: "Qualifier") -> None:
+        """
+        Internal function to update the object's attributes from a different version of the exact same object.
+
+        This function should not be used directly. It is typically used by backend implementations (database adapters,
+        protocol clients, etc.) to update the object's data, after ``update_nss_from()`` has been called.
+
+        :param other: The object to update from
+        """
+        # The type setter rejects the current type as a duplicate, so only assign a changed type
+        if self.type != other.type:
+            self.type = other.type
+        # Assign value_type first so the incoming value is cast against the new type.
+        self.value_type = other.value_type
+        self.value = other.value
+        self.value_id = other.value_id
+        self.kind = other.kind
+        self.semantic_id = other.semantic_id
+        self.supplemental_semantic_id = other.supplemental_semantic_id
+
 
 @_string_constraints.constrain_value_type_iec61360("value")
 class ValueReferencePair:
@@ -2367,7 +2406,6 @@ def get(
             attribute_value if case_sensitive else attribute_value.upper(), default
         )
 
-    # Todo: Implement function including tests
     def update_nss_from(self, other: "NamespaceSet"):
         """
         Update a NamespaceSet from a given NamespaceSet.
@@ -2390,20 +2428,20 @@ def update_nss_from(self, other: "NamespaceSet"):
                     referable.update_from(other_object)  # type: ignore
                 elif isinstance(other_object, Qualifier):
                     backend, case_sensitive = self._backend["type"]
-                    qualifier = backend[  # noqa: F841 qualifier currently unused
+                    qualifier = backend[
                         other_object.type
                         if case_sensitive
                         else other_object.type.upper()
                     ]
-                    # qualifier.update_from(other_object) # TODO: What should happend here? Remove noqa when done
+                    qualifier.update_from(other_object)  # type: ignore
                 elif isinstance(other_object, Extension):
                     backend, case_sensitive = self._backend["name"]
-                    extension = backend[  # noqa: F841 extension currently unused
+                    extension = backend[
                         other_object.name
                         if case_sensitive
                         else other_object.name.upper()
                     ]
-                    # extension.update_from(other_object) # TODO: What should happend here? Remove noqa when done
+                    extension.update_from(other_object)  # type: ignore
                 else:
                     raise TypeError("Type not implemented")
             except KeyError:

sdk/test/model/test_base.py

@@ -284,9 +284,17 @@ def test_get_id_short_path(self):
     def test_update_from(self):
         example_submodel = example_aas.create_example_submodel()
         example_relel = example_submodel.get_referable("ExampleRelationshipElement")
+        example_qualifier = model.Qualifier("q", model.datatypes.String, "old")
+        example_extension = model.Extension("e", model.datatypes.String, "old")
+        example_submodel.add_qualifier(example_qualifier)
+        example_submodel.add_extension(example_extension)
 
         other_submodel = example_aas.create_example_submodel()
         other_relel = other_submodel.get_referable("ExampleRelationshipElement")
+        other_qualifier = model.Qualifier("q", model.datatypes.String, "new")
+        other_extension = model.Extension("e", model.datatypes.String, "new")
+        other_submodel.add_qualifier(other_qualifier)
+        other_submodel.add_extension(other_extension)
 
         other_submodel.category = "NewCat"
         other_relel.category = "NewRelElCat"
@@ -295,6 +303,8 @@ def test_update_from(self):
         example_submodel.update_from(other_submodel)
         self.assertEqual("NewCat", example_submodel.category)
         self.assertEqual("NewRelElCat", example_relel.category)
+        self.assertEqual("new", example_qualifier.value)
+        self.assertEqual("new", example_extension.value)
         # References to Referable objects shall remain stable
         self.assertIs(
             example_relel, example_submodel.get_referable("ExampleRelationshipElement")
@@ -305,12 +315,17 @@ def test_update_from(self):
                 "id_short", "ExampleRelationshipElement"
             ),
         )
+        # The same holds for Qualifier and Extension objects
+        self.assertIs(example_qualifier, example_submodel.get_qualifier_by_type("q"))
+        self.assertIs(example_extension, example_submodel.get_extension_by_name("e"))
         # Check Namespace & parent consistency
         self.assertIs(
             example_submodel.namespace_element_sets[0],
             example_submodel.submodel_element,
         )
         self.assertIs(example_relel.parent, example_submodel)
+        self.assertIs(example_qualifier.parent, example_submodel)
+        self.assertIs(example_extension.parent, example_submodel)
 
     def test_update_commit_qualifier_extension_semantic_id(self):
         submodel = model.Submodel("https://example.org/Test_Submodel")
@@ -372,6 +387,12 @@ def __init__(self, values=()):
         self.set1 = model.NamespaceSet(self, [("type", False)], values)
 
 
+class ExampleNamespaceExtension(model.HasExtension):
+    def __init__(self, values=()):
+        super().__init__()
+        self.set1 = model.NamespaceSet(self, [("name", True)], values)
+
+
 class ModelNamespaceTest(unittest.TestCase):
     _namespace_class = ExampleNamespaceReferable
     _namespace_class_qualifier = ExampleNamespaceQualifier
@@ -843,6 +864,88 @@ def test_Namespaceset_update_from(self) -> None:
             namespace1.get_referable("Prop2")
         self.assertIsNone(prop2.parent)
 
+        # Qualifiers and Extensions are updated the same way, matched by type and name.
+        # ExampleNamespaceReferable also indexes semantic_id, so Qualifier/Extension
+        # objects belong in namespaces that only key on type/name.
+        qualifier_ns1 = self._namespace_class_qualifier()
+        qualifier1 = model.Qualifier(
+            "type1",
+            model.datatypes.Int,
+            1,
+            kind=model.QualifierKind.CONCEPT_QUALIFIER,
+        )
+        qualifier2 = model.Qualifier("type2", model.datatypes.Int, 2)
+        qualifier_ns1.set1.add(qualifier1)
+        qualifier_ns1.set1.add(qualifier2)
+
+        qualifier_ns2 = self._namespace_class_qualifier()
+        qualifier_ns2.set1.add(
+            model.Qualifier(
+                "type1",
+                model.datatypes.String,
+                "updated",
+                kind=model.QualifierKind.VALUE_QUALIFIER,
+                semantic_id=self.propSemanticID,
+            )
+        )
+        qualifier_ns2.set1.add(model.Qualifier("type3", model.datatypes.Int, 3))
+        qualifier_ns1.set1.update_nss_from(qualifier_ns2.set1)
+
+        self.assertIs(
+            qualifier_ns1.set1.get_object_by_attribute("type", "type1"), qualifier1
+        )
+        self.assertEqual("updated", qualifier1.value)
+        self.assertIs(qualifier1.value_type, model.datatypes.String)
+        self.assertEqual(model.QualifierKind.VALUE_QUALIFIER, qualifier1.kind)
+        self.assertEqual(self.propSemanticID, qualifier1.semantic_id)
+        self.assertIs(qualifier1.parent, qualifier_ns1)
+        qualifier3 = qualifier_ns1.set1.get_object_by_attribute("type", "type3")
+        self.assertIs(qualifier3.parent, qualifier_ns1)
+        assert isinstance(qualifier3, model.Qualifier)
+        self.assertEqual(3, qualifier3.value)
+        self.assertFalse(qualifier_ns1.set1.contains_id("type", "type2"))
+        self.assertIsNone(qualifier2.parent)
+
+        extension_ns1 = ExampleNamespaceExtension()
+        extension1 = model.Extension("Ext1", model.datatypes.Int, 1)
+        extension2 = model.Extension("Ext2", model.datatypes.Int, 2)
+        extension_ns1.set1.add(extension1)
+        extension_ns1.set1.add(extension2)
+
+        refers_to = {
+            model.ModelReference(
+                (model.Key(model.KeyTypes.SUBMODEL, "urn:x-test:submodel"),),
+                model.Submodel,
+            )
+        }
+        extension_ns2 = ExampleNamespaceExtension()
+        extension1_new = model.Extension(
+            "Ext1",
+            model.datatypes.String,
+            "updated",
+            refers_to=refers_to,
+            semantic_id=self.propSemanticID,
+        )
+        extension_ns2.set1.add(extension1_new)
+        extension_ns2.set1.add(model.Extension("Ext3", model.datatypes.Int, 3))
+        extension_ns1.set1.update_nss_from(extension_ns2.set1)
+
+        self.assertIs(
+            extension_ns1.set1.get_object_by_attribute("name", "Ext1"), extension1
+        )
+        self.assertEqual("updated", extension1.value)
+        self.assertIs(extension1.value_type, model.datatypes.String)
+        self.assertEqual(refers_to, extension1.refers_to)
+        self.assertIsNot(extension1_new.refers_to, extension1.refers_to)
+        self.assertEqual(self.propSemanticID, extension1.semantic_id)
+        self.assertIs(extension1.parent, extension_ns1)
+        extension3 = extension_ns1.set1.get_object_by_attribute("name", "Ext3")
+        self.assertIs(extension3.parent, extension_ns1)
+        assert isinstance(extension3, model.Extension)
+        self.assertEqual(3, extension3.value)
+        self.assertFalse(extension_ns1.set1.contains_id("name", "Ext2"))
+        self.assertIsNone(extension2.parent)
+
     def test_qualifiable_id_short_namespace(self) -> None:
         prop1 = model.Property("Prop1", model.datatypes.Int, 1)
         qualifier1 = model.Qualifier("Qualifier1", model.datatypes.Int, 2)