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/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()