feat: enhance submodel management with validation in add/remove funct…

aas_http_client/utilities/sdk_tools.py

@@ -10,12 +10,34 @@
 _logger = logging.getLogger(__name__)
 
 
+def get_submodel_ids(shell: model.AssetAdministrationShell) -> list[str]:
+    """Get all IDs from the submodels referenced in the given AAS.
+
+    :param shell: The Asset Administration Shell to extract submodel IDs from.
+    :return: A list of submodel IDs referenced in the AAS.
+    """
+    submodel_ids = []
+    for submodel in shell.submodel:
+        if len(submodel.key) < 1 or submodel.key[0].type != model.KeyTypes.SUBMODEL:
+            _logger.warning(f"Submodel reference {submodel} does not start with SUBMODEL key type.")
+            continue
+
+        submodel_ids.append(submodel.key[0].value)
+
+    return submodel_ids
+
+
 def add_submodel_to_aas(aas: model.AssetAdministrationShell, submodel: model.Submodel) -> None:
     """Add a given Submodel correctly to a provided AssetAdministrationShell.
 
     :param aas: provided AssetAdministrationShell to which the Submodel should be added
     :param submodel: given Submodel to add
     """
+    existing_submodel_ids = get_submodel_ids(aas)
+    if submodel.id in existing_submodel_ids:
+        _logger.warning(f"Submodel with ID {submodel.id} is already referenced in the AAS. Skipping addition.")
+        return
+
     aas.submodel.add(model.ModelReference.from_referable(submodel))
 
 
@@ -25,6 +47,11 @@ def remove_submodel_from_aas(aas: model.AssetAdministrationShell, submodel: mode
     :param aas: provided AssetAdministrationShell from which the Submodel should be removed
     :param submodel: given Submodel to remove
     """
+    existing_submodel_ids = get_submodel_ids(aas)
+    if submodel.id not in existing_submodel_ids:
+        _logger.warning(f"Submodel with ID {submodel.id} is not referenced in the AAS. Skipping removal.")
+        return
+
     aas.submodel.remove(model.ModelReference.from_referable(submodel))
 
 

tests/test_sdk_tools.py

@@ -0,0 +1,51 @@
+from basyx.aas import model
+from aas_http_client.utilities import encoder
+import aas_http_client.utilities.model_builder as model_builder
+import aas_http_client.utilities.sdk_tools as sdk_tools
+import pytest
+
+SM_ID = "fluid40/sm_sdk_tools_unit_tests"
+SHELL_ID = "fluid40/aas_sdk_tools_unit_tests"
+
+@pytest.fixture(scope="module")
+def shared_sm() -> model.Submodel:
+    # create a Submodel
+    return model_builder.create_base_submodel(identifier=SM_ID, id_short="sm_http_client_unit_tests", display_name="Submodel HTTP Client Unit Tests", description="This is a sample Submodel created for unit testing of the AAS HTTP Client.")
+
+@pytest.fixture(scope="module")
+def shared_sm_temp() -> model.Submodel:
+    # create a Submodel
+    return model_builder.create_base_submodel(identifier=f"{SM_ID}_temp", id_short="sm_http_client_unit_tests", display_name="Submodel HTTP Client Unit Tests", description="This is a sample Submodel created for unit testing of the AAS HTTP Client.")
+
+@pytest.fixture(scope="module")
+def shared_aas() -> model.AssetAdministrationShell:
+    # create an AAS
+    aas = model_builder.create_base_aas(identifier=SHELL_ID, id_short="aas_http_client_unit_tests", global_asset_identifier=SHELL_ID, display_name="AAS HTTP Client Unit Tests", description="This is a sample AAS created for unit testing of the AAS HTTP Client.")
+    return aas
+
+def test_001a_add_sm_to_shell(shared_aas: model.AssetAdministrationShell, shared_sm: model.Submodel):
+    assert len(shared_aas.submodel) == 0
+
+    sdk_tools.add_submodel_to_aas(shared_aas, shared_sm)
+    assert len(shared_aas.submodel) == 1
+
+    assert shared_sm.id in sdk_tools.get_submodel_ids(shared_aas)
+
+
+def test_001b_add_sm_to_shell(shared_aas: model.AssetAdministrationShell, shared_sm: model.Submodel):
+    assert len(shared_aas.submodel) == 1
+
+    sdk_tools.add_submodel_to_aas(shared_aas, shared_sm)
+    assert len(shared_aas.submodel) == 1
+
+def test_002a_remove_sm_from_shell(shared_aas: model.AssetAdministrationShell, shared_sm_temp: model.Submodel):
+    assert len(shared_aas.submodel) == 1
+
+    sdk_tools.remove_submodel_from_aas(shared_aas, shared_sm_temp)
+    assert len(shared_aas.submodel) == 1
+
+def test_002b_remove_sm_from_shell(shared_aas: model.AssetAdministrationShell, shared_sm: model.Submodel):
+    assert len(shared_aas.submodel) == 1
+
+    sdk_tools.remove_submodel_from_aas(shared_aas, shared_sm)
+    assert len(shared_aas.submodel) == 0