Merge pull request #1 from rwth-iat/entityfix

improve `AssetInformation` and `Entity` constraint validation + add tests

basyx/aas/model/aas.py

@@ -60,32 +60,51 @@ def __init__(self,
 
         super().__init__()
         self.asset_kind: base.AssetKind = asset_kind
-        self.specific_asset_id: base.ConstrainedList[base.SpecificAssetId] = \
-            base.ConstrainedList(specific_asset_id, item_del_hook=self._check_constraint_del_spec_asset_id)
-        self.global_asset_id: Optional[base.Identifier] = global_asset_id
+        self._specific_asset_id: base.ConstrainedList[base.SpecificAssetId] = \
+            base.ConstrainedList(specific_asset_id, item_set_hook=self._check_constraint_set_spec_asset_id,
+                                 item_del_hook=self._check_constraint_del_spec_asset_id)
+        self._global_asset_id: Optional[base.Identifier]
+        # AASd-131 is validated via the global_asset_id setter
+        self.global_asset_id = global_asset_id
         self.asset_type: Optional[base.Identifier] = asset_type
         self.default_thumbnail: Optional[base.Resource] = default_thumbnail
 
+    def _check_constraint_set_spec_asset_id(self, old: List[base.SpecificAssetId], new: List[base.SpecificAssetId],
+                                            list_: List[base.SpecificAssetId]) -> None:
+        self._validate_asset_ids(self.global_asset_id,
+                                 # whether the list is nonempty after the set operation
+                                 len(old) < len(list_) or len(new) > 0)
+
     def _check_constraint_del_spec_asset_id(self, _item_to_del: base.SpecificAssetId,
-                                            _list: List[base.SpecificAssetId]) -> None:
-        if self.global_asset_id is None and len(_list) == 1:
-            raise base.AASConstraintViolation(
-                131, "An AssetInformation has to have a globalAssetId or a specificAssetId")
+                                            list_: List[base.SpecificAssetId]) -> None:
+        self._validate_asset_ids(self.global_asset_id, len(list_) > 1)
 
     @property
-    def global_asset_id(self):
+    def global_asset_id(self) -> Optional[base.Identifier]:
         return self._global_asset_id
 
     @global_asset_id.setter
-    def global_asset_id(self, global_asset_id: Optional[base.Identifier]):
-        if global_asset_id is None:
-            if not self.specific_asset_id:
-                raise base.AASConstraintViolation(
-                    131, "An AssetInformation has to have a globalAssetId or a specificAssetId")
-        else:
+    def global_asset_id(self, global_asset_id: Optional[base.Identifier]) -> None:
+        self._validate_asset_ids(global_asset_id, bool(self.specific_asset_id))
+        if global_asset_id is not None:
             _string_constraints.check_identifier(global_asset_id)
         self._global_asset_id = global_asset_id
 
+    @property
+    def specific_asset_id(self) -> base.ConstrainedList[base.SpecificAssetId]:
+        return self._specific_asset_id
+
+    @specific_asset_id.setter
+    def specific_asset_id(self, specific_asset_id: Iterable[base.SpecificAssetId]) -> None:
+        # constraints are checked via _check_constraint_set_spec_asset_id() in this case
+        self._specific_asset_id[:] = specific_asset_id
+
+    @staticmethod
+    def _validate_asset_ids(global_asset_id: Optional[base.Identifier], specific_asset_id_nonempty: bool) -> None:
+        if global_asset_id is None and not specific_asset_id_nonempty:
+            raise base.AASConstraintViolation(131,
+                                              "An AssetInformation has to have a globalAssetId or a specificAssetId")
+
     def __repr__(self) -> str:
         return "AssetInformation(assetKind={}, globalAssetId={}, specificAssetId={}, assetType={}, " \
                "defaultThumbnail={})".format(self.asset_kind, self._global_asset_id, str(self.specific_asset_id),

basyx/aas/model/base.py

@@ -1311,6 +1311,10 @@ def extend(self, values: Iterable[_T]) -> None:
                 self._item_add_hook(v, self._list + v_list[:idx])
         self._list = self._list + v_list
 
+    def clear(self) -> None:
+        # clear() repeatedly deletes the last item by default, making it not atomic
+        del self[:]
+
     @overload
     def __getitem__(self, index: int) -> _T: ...
 

basyx/aas/model/submodel.py

@@ -1048,7 +1048,6 @@ def __init__(self,
                          supplemental_semantic_id, embedded_data_specifications)
 
 
-@_string_constraints.constrain_identifier("global_asset_id")
 class Entity(SubmodelElement, base.UniqueIdShortNamespace):
     """
     An entity is a :class:`~.SubmodelElement` that is used to model entities
@@ -1107,47 +1106,75 @@ def __init__(self,
         super().__init__(id_short, display_name, category, description, parent, semantic_id, qualifier, extension,
                          supplemental_semantic_id, embedded_data_specifications)
         self.statement = base.NamespaceSet(self, [("id_short", True)], statement)
-        self.specific_asset_id: base.ConstrainedList[base.SpecificAssetId] = \
-            base.ConstrainedList(specific_asset_id, item_del_hook=self._check_constraint_del_spec_asset_id)
-        self.global_asset_id: Optional[base.Identifier] = global_asset_id
-        self.entity_type: base.EntityType = entity_type
+        self._specific_asset_id: base.ConstrainedList[base.SpecificAssetId] = \
+            base.ConstrainedList(specific_asset_id, item_set_hook=self._check_constraint_set_spec_asset_id,
+                                 item_del_hook=self._check_constraint_del_spec_asset_id)
+        # assign private attributes, bypassing setters, as constraints will be checked below
+        self._entity_type: base.EntityType = entity_type
+        # add item_add_hook after items have been added, because checking the constraints requires the global_asset_id
+        # to be set
+        self._specific_asset_id._item_add_hook = self._check_constraint_add_spec_asset_id
+        # use setter for global_asset_id, as it also checks the string constraint,
+        # which hasn't been checked at this point
+        # furthermore, the setter also validates AASd-014
+        self._global_asset_id: Optional[base.Identifier]
+        self.global_asset_id = global_asset_id
+
+    def _check_constraint_add_spec_asset_id(self, _new: base.SpecificAssetId, _list: List[base.SpecificAssetId]) \
+            -> None:
+        self._validate_asset_ids_for_entity_type(self.entity_type, self.global_asset_id, True)
+
+    def _check_constraint_set_spec_asset_id(self, old: List[base.SpecificAssetId], new: List[base.SpecificAssetId],
+                                            list_: List[base.SpecificAssetId]) -> None:
+        self._validate_asset_ids_for_entity_type(self.entity_type, self.global_asset_id,
+                                                 # whether the list is nonempty after the set operation
+                                                 len(old) < len(list_) or len(new) > 0)
 
     def _check_constraint_del_spec_asset_id(self, _item_to_del: base.SpecificAssetId,
-                                            _list: List[base.SpecificAssetId]) -> None:
-        if self.global_asset_id is None and len(_list) == 1:
-            raise base.AASConstraintViolation(
-                131, "An AssetInformation has to have a globalAssetId or a specificAssetId")
+                                            list_: List[base.SpecificAssetId]) -> None:
+        self._validate_asset_ids_for_entity_type(self.entity_type, self.global_asset_id, len(list_) > 1)
 
     @property
     def entity_type(self) -> base.EntityType:
         return self._entity_type
 
     @entity_type.setter
     def entity_type(self, entity_type: base.EntityType) -> None:
-        self._validate_asset_ids_for_entity_type(entity_type, self.global_asset_id, self.specific_asset_id)
-        self._entity_type: base.EntityType = entity_type
+        self._validate_asset_ids_for_entity_type(entity_type, self.global_asset_id, bool(self.specific_asset_id))
+        self._entity_type = entity_type
 
     @property
-    def global_asset_id(self):
+    def global_asset_id(self) -> Optional[base.Identifier]:
         return self._global_asset_id
 
     @global_asset_id.setter
-    def global_asset_id(self, global_asset_id: Optional[base.Identifier]):
-        self._validate_asset_ids_for_entity_type(self.entity_type, global_asset_id, self.specific_asset_id)
+    def global_asset_id(self, global_asset_id: Optional[base.Identifier]) -> None:
+        self._validate_asset_ids_for_entity_type(self.entity_type, global_asset_id, bool(self.specific_asset_id))
+        if global_asset_id is not None:
+            _string_constraints.check_identifier(global_asset_id)
         self._global_asset_id = global_asset_id
 
+    @property
+    def specific_asset_id(self) -> base.ConstrainedList[base.SpecificAssetId]:
+        return self._specific_asset_id
+
+    @specific_asset_id.setter
+    def specific_asset_id(self, specific_asset_id: Iterable[base.SpecificAssetId]) -> None:
+        # constraints are checked via _check_constraint_set_spec_asset_id() in this case
+        self._specific_asset_id[:] = specific_asset_id
+
     @staticmethod
     def _validate_asset_ids_for_entity_type(entity_type: base.EntityType,
                                             global_asset_id: Optional[base.Identifier],
-                                            specific_asset_id: Optional[base.ConstrainedList[base.SpecificAssetId]]):
-        if entity_type == base.EntityType.SELF_MANAGED_ENTITY and global_asset_id is None and not specific_asset_id:
+                                            specific_asset_id_nonempty: bool) -> None:
+        if entity_type == base.EntityType.SELF_MANAGED_ENTITY and global_asset_id is None \
+                and not specific_asset_id_nonempty:
             raise base.AASConstraintViolation(
                 14, "A self-managed entity has to have a globalAssetId or a specificAssetId")
-        elif entity_type == base.EntityType.CO_MANAGED_ENTITY and (global_asset_id or specific_asset_id):
+        elif entity_type == base.EntityType.CO_MANAGED_ENTITY and (global_asset_id is not None
+                                                                   or specific_asset_id_nonempty):
             raise base.AASConstraintViolation(
                 14, "A co-managed entity has to have neither a globalAssetId nor a specificAssetId")
-        if global_asset_id:
-            _string_constraints.check_identifier(global_asset_id)
 
 
 class EventElement(SubmodelElement, metaclass=abc.ABCMeta):

test/model/test_aas.py

@@ -0,0 +1,86 @@
+# Copyright (c) 2023 the Eclipse BaSyx Authors
+#
+# This program and the accompanying materials are made available under the terms of the MIT License, available in
+# the LICENSE file of this project.
+#
+# SPDX-License-Identifier: MIT
+
+import unittest
+
+from basyx.aas import model
+
+
+class AssetInformationTest(unittest.TestCase):
+    def test_aasd_131_init(self) -> None:
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            model.AssetInformation(model.AssetKind.INSTANCE)
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))
+        model.AssetInformation(model.AssetKind.INSTANCE, global_asset_id="https://acplt.org/TestAsset")
+        model.AssetInformation(model.AssetKind.INSTANCE, specific_asset_id=(model.SpecificAssetId("test", "test"),))
+        model.AssetInformation(model.AssetKind.INSTANCE, global_asset_id="https://acplt.org/TestAsset",
+                               specific_asset_id=(model.SpecificAssetId("test", "test"),))
+
+    def test_aasd_131_set(self) -> None:
+        asset_information = model.AssetInformation(model.AssetKind.INSTANCE,
+                                                   global_asset_id="https://acplt.org/TestAsset",
+                                                   specific_asset_id=(model.SpecificAssetId("test", "test"),))
+        asset_information.global_asset_id = None
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            asset_information.specific_asset_id = model.ConstrainedList(())
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))
+
+        asset_information = model.AssetInformation(model.AssetKind.INSTANCE,
+                                                   global_asset_id="https://acplt.org/TestAsset",
+                                                   specific_asset_id=(model.SpecificAssetId("test", "test"),))
+        asset_information.specific_asset_id = model.ConstrainedList(())
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            asset_information.global_asset_id = None
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))
+
+    def test_aasd_131_specific_asset_id_add(self) -> None:
+        asset_information = model.AssetInformation(model.AssetKind.INSTANCE,
+                                                   global_asset_id="https://acplt.org/TestAsset")
+        specific_asset_id1 = model.SpecificAssetId("test", "test")
+        specific_asset_id2 = model.SpecificAssetId("test", "test")
+        asset_information.specific_asset_id.append(specific_asset_id1)
+        asset_information.specific_asset_id.extend((specific_asset_id2,))
+        self.assertIs(asset_information.specific_asset_id[0], specific_asset_id1)
+        self.assertIs(asset_information.specific_asset_id[1], specific_asset_id2)
+
+    def test_aasd_131_specific_asset_id_set(self) -> None:
+        asset_information = model.AssetInformation(model.AssetKind.INSTANCE,
+                                                   specific_asset_id=(model.SpecificAssetId("test", "test"),))
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            asset_information.specific_asset_id[:] = ()
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))
+        specific_asset_id = model.SpecificAssetId("test", "test")
+        self.assertIsNot(asset_information.specific_asset_id[0], specific_asset_id)
+        asset_information.specific_asset_id[:] = (specific_asset_id,)
+        self.assertIs(asset_information.specific_asset_id[0], specific_asset_id)
+        asset_information.specific_asset_id[0] = model.SpecificAssetId("test", "test")
+        self.assertIsNot(asset_information.specific_asset_id[0], specific_asset_id)
+
+    def test_aasd_131_specific_asset_id_del(self) -> None:
+        specific_asset_id = model.SpecificAssetId("test", "test")
+        asset_information = model.AssetInformation(model.AssetKind.INSTANCE,
+                                                   specific_asset_id=(model.SpecificAssetId("test1", "test1"),
+                                                                      specific_asset_id))
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            del asset_information.specific_asset_id[:]
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            asset_information.specific_asset_id.clear()
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))
+        self.assertIsNot(asset_information.specific_asset_id[0], specific_asset_id)
+        del asset_information.specific_asset_id[0]
+        self.assertIs(asset_information.specific_asset_id[0], specific_asset_id)
+        with self.assertRaises(model.AASConstraintViolation) as cm:
+            del asset_information.specific_asset_id[0]
+        self.assertEqual("An AssetInformation has to have a globalAssetId or a specificAssetId (Constraint AASd-131)",
+                         str(cm.exception))

test/model/test_base.py

@@ -1213,7 +1213,15 @@ def hook(itm: int, _list: List[int]) -> None:
         self.assertEqual(c_list, [1, 2, 3])
         with self.assertRaises(ValueError):
             c_list.clear()
-        self.assertEqual(c_list, [1, 2, 3])
+        # the default clear() implementation seems to repeatedly delete the last item until the list is empty
+        # in this case, the last item is 3, which cannot be deleted because it is > 2, thus leaving it unclear whether
+        # clear() really is atomic. to work around this, the list is reversed, making 1 the last item, and attempting
+        # to clear again.
+        c_list.reverse()
+        with self.assertRaises(ValueError):
+            c_list.clear()
+        self.assertEqual(c_list, [3, 2, 1])
+        c_list.reverse()
         del c_list[0:2]
         self.assertEqual(c_list, [3])