From e15d4a53621f8eabeb013a8065d41a0e3b8fbe92 Mon Sep 17 00:00:00 2001 From: Sankalp Thakur Date: Sat, 5 Sep 2026 19:26:31 +0400 Subject: [PATCH] model.base: Copy Qualifier and Extension data in update_nss_from `NamespaceSet.update_nss_from()` is how backends refresh a namespace from a newer copy of the same objects. Matching Referable objects get their attributes copied in place. Matching Qualifier and Extension objects were only looked up; their value and the rest of the payload stayed on the old object. A PUT of a Submodel with the same qualifier type therefore kept the previous value. Backends already treat a Qualifier of the same type and an Extension of the same name as the same object, so those fields have to be copied onto the object that already lives in the target set. Fixes #602 --- sdk/basyx/aas/model/base.py | 42 ++++++++++++-- sdk/test/model/test_base.py | 108 ++++++++++++++++++++++++++++++++++++ 2 files changed, 145 insertions(+), 5 deletions(-) diff --git a/sdk/basyx/aas/model/base.py b/sdk/basyx/aas/model/base.py index 84650fe17..630b171c0 100644 --- a/sdk/basyx/aas/model/base.py +++ b/sdk/basyx/aas/model/base.py @@ -1777,6 +1777,22 @@ 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"): + """ + 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 + """ + # 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 = other.refers_to + self.semantic_id = other.semantic_id + self.supplemental_semantic_id = other.supplemental_semantic_id + class HasKind(metaclass=abc.ABCMeta): """ @@ -1933,6 +1949,23 @@ 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"): + """ + 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 + """ + # 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 +2400,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 +2422,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: diff --git a/sdk/test/model/test_base.py b/sdk/test/model/test_base.py index 2e2e75ca0..000280bd8 100644 --- a/sdk/test/model/test_base.py +++ b/sdk/test/model/test_base.py @@ -312,6 +312,26 @@ def test_update_from(self): ) self.assertIs(example_relel.parent, example_submodel) + def test_update_from_qualifier_and_extension(self): + submodel = model.Submodel("https://example.org/Test_Submodel") + qualifier = model.Qualifier("q", model.datatypes.String, "old") + extension = model.Extension("e", model.datatypes.String, "old") + submodel.add_qualifier(qualifier) + submodel.add_extension(extension) + + other = model.Submodel("https://example.org/Test_Submodel") + other.add_qualifier(model.Qualifier("q", model.datatypes.String, "new")) + other.add_extension(model.Extension("e", model.datatypes.String, "new")) + + submodel.update_from(other) + + self.assertIs(submodel.get_qualifier_by_type("q"), qualifier) + self.assertEqual("new", qualifier.value) + self.assertIs(qualifier.parent, submodel) + self.assertIs(submodel.get_extension_by_name("e"), extension) + self.assertEqual("new", extension.value) + self.assertIs(extension.parent, submodel) + def test_update_commit_qualifier_extension_semantic_id(self): submodel = model.Submodel("https://example.org/Test_Submodel") qualifier = model.Qualifier("test", model.datatypes.String) @@ -372,6 +392,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 +869,88 @@ def test_Namespaceset_update_from(self) -> None: namespace1.get_referable("Prop2") self.assertIsNone(prop2.parent) + def test_Namespaceset_update_from_qualifier_and_extension(self) -> None: + # 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() + extension_ns2.set1.add( + model.Extension( + "Ext1", + model.datatypes.String, + "updated", + refers_to=refers_to, + semantic_id=self.propSemanticID, + ) + ) + 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.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)