Add SAMM version validation to SAMMGraph load_aspect_model

Author: Oleksandr Muzyka (EPAM)

core/esmf-aspect-meta-model-python/esmf_aspect_meta_model_python/loader/samm_graph.py

@@ -213,12 +213,41 @@ def load_aspect_model(self) -> Aspect:
 
             graph = self.rdf_graph + self.samm_graph
             self._reader.prepare_aspect_model(graph)
+            self._validate_samm_namespace_version(graph)
 
             model_element_factory = ModelElementFactory(self.samm_version, graph, self._cache)
             self.aspect = model_element_factory.create_element(aspect_urn)
 
         return self.aspect
 
+    def _validate_samm_namespace_version(self, graph: Graph) -> None:
+        """
+        Validates that the SAMM version in the given RDF graph matches the detected SAMM version.
+
+        This method iterates through the namespaces in the RDF graph and checks if any namespace
+        corresponds to the SAMM namespace prefix. If a SAMM namespace is found, it extracts the version
+        from the namespace and compares it to the `samm_version` attribute of the class. If the versions
+        do not match, a `ValueError` is raised.
+
+        Args:
+            graph (Graph): The RDF graph whose namespaces are to be validated.
+
+        Raises:
+            ValueError: If the SAMM version in the graph's namespace does not match the detected SAMM version.
+        """
+        for prefix, namespace in graph.namespace_manager.namespaces():
+            if prefix.startswith(self.samm_namespace_prefix):
+                namespace_info = namespace.split(":")  # [urn, namespace_id, namespace_specific_str, entity, version#]
+
+                if len(namespace_info) == 5 and namespace_info[2] == "org.eclipse.esmf.samm":
+                    version = namespace_info[-1].replace("#", "")
+
+                    if version != self.samm_version:
+                        raise ValueError(
+                            f"SAMM version mismatch. Found '{version}', but expected '{self.samm_version}'. "
+                            "Ensure all RDF files use a single, consistent SAMM version"
+                        )
+
     def _get_aspect_from_elements(self):
         """Geta and save the Aspect element from the model elements."""
         if self.model_elements:

core/esmf-aspect-meta-model-python/tests/integration/aspect_model_loader/resources/org.eclipse.esmf.test.general_with_references/2.2.0/AspectWithEarlierSubmodelVersionReferences.ttl

@@ -0,0 +1,19 @@
+#
+# Copyright (c) 2023 Robert Bosch Manufacturing Solutions GmbH, Germany. All rights reserved.
+#
+
+@prefix : <urn:samm:org.eclipse.esmf.test.general_with_references:2.2.0#>.
+@prefix type: <urn:samm:org.eclipse.esmf.test.types:2.1.0#> .
+@prefix samm: <urn:samm:org.eclipse.esmf.samm:meta-model:2.2.0#> .
+
+:AspectWithEarlierSubmodelVersionReferences a samm:Aspect ;
+    samm:preferredName "Aspect with references"@en ;
+    samm:description "Test aspect with references from different files."@en ;
+    samm:properties ( :testProperty ) ;
+    samm:operations ( ) .
+
+:testProperty a samm:Entity ;
+            samm:preferredName "Test property"@en ;
+            samm:description "Test property description."@en ;
+            samm:properties ( :ExternalPartId
+                              type:TypeList ) .

core/esmf-aspect-meta-model-python/tests/integration/aspect_model_loader/test_resolve_elements_referencies.py

@@ -9,18 +9,22 @@
 #
 #   SPDX-License-Identifier: MPL-2.0
 
-from os import getcwd
 from pathlib import Path
 
+import pytest
+
 from esmf_aspect_meta_model_python import SAMMGraph
 
-RESOURCE_PATH = getcwd() / Path(
-    "tests/integration/aspect_model_loader/resources/org.eclipse.esmf.test.general_with_references/2.1.0"
+ASPECT_MODELS_DIR = (
+    Path.cwd() / "tests/integration/aspect_model_loader/resources/org.eclipse.esmf.test.general_with_references"
 )
 
+ASPECT_MODELS_V2_1_0 = ASPECT_MODELS_DIR / "2.1.0"
+ASPECT_MODELS_V2_2_0 = ASPECT_MODELS_DIR / "2.2.0"
+
 
 def test_resolve_elements_references():
-    file_path = RESOURCE_PATH / "AspectWithReferences.ttl"
+    file_path = ASPECT_MODELS_V2_1_0 / "AspectWithReferences.ttl"
     samm_graph = SAMMGraph()
     samm_graph.parse(file_path)
     aspect = samm_graph.load_aspect_model()
@@ -51,3 +55,18 @@ def test_resolve_elements_references():
     assert property_4.get_preferred_name("en") == "Test List"
     assert property_4.get_description("en") == "This is a test list."
     assert property_4.see == ["http://example.com/"]
+
+
+def test_aspect_with_diff_meta_model_submodel_version():
+    file_path = ASPECT_MODELS_V2_2_0 / "AspectWithEarlierSubmodelVersionReferences.ttl"
+    samm_graph = SAMMGraph()
+    samm_graph.parse(file_path)
+
+    with pytest.raises(
+        ValueError,
+        match=(
+            "SAMM version mismatch. Found '2.1.0', but expected '2.2.0'. "
+            "Ensure all RDF files use a single, consistent SAMM version"
+        ),
+    ):
+        samm_graph.load_aspect_model()

core/esmf-aspect-meta-model-python/tests/unit/loader/test_samm_graph.py

@@ -216,7 +216,10 @@ def test_load_aspect_model(self):
 
     @mock.patch("esmf_aspect_meta_model_python.loader.samm_graph.ModelElementFactory")
     @mock.patch("esmf_aspect_meta_model_python.loader.samm_graph.SAMMGraph.get_aspect_urn")
-    def test_load_aspect_model_create_element(self, get_aspect_urn_mock, model_element_factory_mock):
+    @mock.patch("esmf_aspect_meta_model_python.loader.samm_graph.SAMMGraph._validate_samm_namespace_version")
+    def test_load_aspect_model_create_element(
+        self, validate_samm_namespace_version_mock, get_aspect_urn_mock, model_element_factory_mock
+    ):
         reader_mock = mock.MagicMock(name="reader")
         cache_mock = mock.MagicMock(name="cache")
         samm_graph = SAMMGraph()
@@ -229,13 +232,53 @@ def test_load_aspect_model_create_element(self, get_aspect_urn_mock, model_eleme
         get_aspect_urn_mock.return_value = "aspect_urn"
         model_element_factory_mock.return_value = model_element_factory_mock
         model_element_factory_mock.create_element.return_value = "aspect"
+
         result = samm_graph.load_aspect_model()
 
         assert result == "aspect"
         get_aspect_urn_mock.assert_called_once()
         reader_mock.prepare_aspect_model.assert_called_once_with("rdf_graph_samm_graph")
         model_element_factory_mock.assert_called_once_with("1.2.3", "rdf_graph_samm_graph", cache_mock)
         model_element_factory_mock.create_element.assert_called_once_with("aspect_urn")
+        validate_samm_namespace_version_mock.assert_called_once_with("rdf_graph_samm_graph")
+
+    class TestValidateSammNamespaceVersion:
+        """Test suite for SAMMGraph._validate_samm_namespace_version method."""
+
+        def test_valid_version(self):
+            version = "1.2.3"
+            samm_graph = SAMMGraph()
+            samm_graph.samm_version = version
+            graph = mock.MagicMock(name="rdf_graph")
+            graph.namespace_manager.namespaces.return_value = [
+                ("rdf", "http://www.w3.org/1999/02/22-rdf-syntax-ns#"),
+                ("samm", f"urn:samm:org.eclipse.esmf.samm:meta-model:{version}#"),  # True
+                ("samm1", f"urn:samm:org.eclipse.esmf.samm:{version}#"),
+                ("samm-e1", f"urn:samm:org.eclipse.esmf.samm:entity:{version}#"),  # True
+                ("samm-c1", "urn:samm:org.china.zes.samm:extension:1.1.1#"),
+            ]
+
+            # Should not raise an exception
+            samm_graph._validate_samm_namespace_version(graph)
+
+        def test_version_mismatch(self):
+            samm_version = "2.2.0"
+            earlier_version = "2.1.0"
+            samm_graph = SAMMGraph()
+            samm_graph.samm_version = samm_version
+            graph = mock.MagicMock(name="rdf_graph")
+            graph.namespace_manager.namespaces.return_value = [
+                ("rdf", "http://www.w3.org/1999/02/22-rdf-syntax-ns#"),
+                ("samm", f"urn:samm:org.eclipse.esmf.samm:meta-model:{samm_version}#"),
+                ("samm-e1", f"urn:samm:org.eclipse.esmf.samm:extension:{earlier_version}#"),  # Exception
+            ]
+
+            with pytest.raises(
+                ValueError,
+                match=f"SAMM version mismatch. Found '{earlier_version}', but expected '{samm_version}'. "
+                "Ensure all RDF files use a single, consistent SAMM version",
+            ):
+                samm_graph._validate_samm_namespace_version(graph)
 
     def test_load_model_elements(self):
         samm_graph = SAMMGraph()