test sdk: Increase test coverage

sdk/test/adapter/aasx/test_aasx.py

@@ -11,6 +11,7 @@
 import tempfile
 import unittest
 import warnings
+from pathlib import Path
 
 import pyecma376_2
 from basyx.aas import model
@@ -62,14 +63,14 @@ def test_supplementary_file_container(self) -> None:
 
         # Check metadata
         self.assertEqual("application/pdf", container.get_content_type("/TestFile.pdf"))
-        self.assertEqual("b18229b24a4ee92c6c2b6bc6a8018563b17472f1150d35d5a5945afeb447ed44",
+        self.assertEqual("142a0061de1ef5c22137ab05bb6001335596c0fc8693d33fa9b011ceac652342",
                          container.get_sha256("/TestFile.pdf").hex())
         self.assertIn("/TestFile.pdf", container)
 
         # Check contents
         file_content = io.BytesIO()
         container.write_file("/TestFile.pdf", file_content)
-        self.assertEqual(hashlib.sha1(file_content.getvalue()).hexdigest(), "78450a66f59d74c073bf6858db340090ea72a8b1")
+        self.assertEqual(hashlib.sha1(file_content.getvalue()).hexdigest(), "241e62aef8b4cdad0975f6c68a4ed8b3923d8db1")
 
         # Add same file again with different content_type to test reference counting
         with open(__file__, 'rb') as f:
@@ -90,6 +91,249 @@ def test_supplementary_file_container(self) -> None:
 
 
 class AASXWriterTest(unittest.TestCase):
+    def test_write_missing_aas_objects(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            data = example_aas.create_full_example()
+
+            # ---- Act & Assert -----
+            with self.assertLogs(level="WARNING") as log:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=True) as writer:
+                    # try to write non-existing object
+                    writer.write_aas_objects(
+                        "/aasx/selection.xml",
+                        ["https://example.org/Test_AssetAdministrationShell",
+                         "http://false-identifier.org/",
+                         "http://example.org/Submodels/Assets/TestAsset/Identification"],
+                        data, aasx.DictSupplementaryFileContainer()
+                    )
+
+            self.assertIn("Could not find identifiable http://false-identifier.org/ in IdentifiableStore",
+                          log.output[0])
+
+            # assert only the two existing objects have been written to aasx file
+            object_store = model.DictIdentifiableStore()
+            with aasx.AASXReader(tmpdir_path / "tmp.aasx") as reader:
+                reader.read_into(object_store, aasx.DictSupplementaryFileContainer())
+            self.assertEqual(len(object_store), 2)
+
+    def test_writing_with_missing_file(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            # data contains a submodel with a File submodel_element
+            # the empty_file_store does not contain the referenced file
+            data = example_aas.create_full_example()
+            empty_file_store = aasx.DictSupplementaryFileContainer()
+
+            # ---- Act & Assert ----
+            # assert warning is present in failsafe mode
+            with self.assertLogs(level="WARNING") as log:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=True) as writer:
+                    writer.write_all_aas_objects("/aasx/data.xml", data, empty_file_store)
+            self.assertIn("Could not find file", log.output[0])
+
+            # assert exception is rose in non-failsafe mode
+            with self.assertRaises(KeyError) as cm:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=False) as writer:
+                    writer.write_all_aas_objects("/aasx/data.xml", data, empty_file_store)
+            self.assertIn("Could not find file", cm.exception.args[0])
+
+    def test_writing_file_twice(self) -> None:
+        with (tempfile.TemporaryDirectory() as tmpdir):
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            file_store = aasx.DictSupplementaryFileContainer()
+            with open(Path(__file__).parent / "TestFile.pdf", "rb") as pdf:
+                resulting_file_name = file_store.add_file("/TestFile.pdf", pdf, "application/pdf")
+
+            # create two submodels that reference the same file in file_store
+            first_submodel = model.Submodel(
+                id_="http://example.org/First_Submodel",
+                submodel_element=[model.File(
+                    id_short="ExampleFile",
+                    content_type="application/pdf",
+                    value=resulting_file_name
+                )]
+            )
+            second_submodel = model.Submodel(
+                id_="http://example.org/SecondSubmodel",
+                submodel_element=[model.File(
+                    id_short="ExampleFile",
+                    content_type="application/pdf",
+                    value=resulting_file_name
+                )]
+            )
+            data: model.DictIdentifiableStore[model.Identifiable] \
+                = model.DictIdentifiableStore([first_submodel, second_submodel])
+
+            # ---- Act & Assert ----
+            with self.assertNoLogs(level="WARNING"):
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx") as writer:
+                    writer.write_all_aas_objects("/aasx/data.xml", data, file_store)
+
+    def test_write_non_aas(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            data = example_aas.create_full_example()
+            file_store = aasx.DictSupplementaryFileContainer()
+            with open(Path(__file__).parent / "TestFile.pdf", "rb") as pdf:
+                file_store.add_file("/TestFile.pdf", pdf, "application/pdf")
+
+            # ---- Act & Assert ----
+            # assert warning is present in failsafe mode
+            with self.assertLogs(level="WARNING") as log:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=True) as writer:
+                    # try to write a non AAS object
+                    writer.write_aas("https://example.org/Test_Submodel", data, file_store)
+            self.assertIn("Skipping AAS https://example.org/Test_Submodel", log.output[0])
+
+            # assert exception is rose in non-failsafe mode
+            with self.assertRaises(TypeError) as cm:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=False) as writer:
+                    # try to write a non AAS object
+                    writer.write_aas("https://example.org/Test_Submodel", data, file_store)
+            self.assertIn("Identifier https://example.org/Test_Submodel does not belong "
+                          "to an AssetAdministrationShell", cm.exception.args[0])
+
+    def test_write_aas_missing_submodel(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            # leave example_submodel out of object store
+            data = model.DictIdentifiableStore([
+                example_aas.create_example_asset_administration_shell(),
+                example_aas.create_example_asset_identification_submodel(),
+                example_aas.create_example_bill_of_material_submodel()
+            ])
+            empty_file_store = aasx.DictSupplementaryFileContainer()
+
+            # ---- Act & Assert ----
+            # assert warning is present in failsafe mode
+            with self.assertLogs(level="WARNING") as log:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=True) as writer:
+                    writer.write_aas("https://example.org/Test_AssetAdministrationShell", data, empty_file_store)
+            self.assertIn("Could not find Submodel", log.output[0])
+
+            # assert exception is rose in non-failsafe mode
+            with self.assertRaises(KeyError) as cm:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=False) as writer:
+                    writer.write_aas("https://example.org/Test_AssetAdministrationShell", data, empty_file_store)
+            self.assertIn("Could not find Submodel", cm.exception.args[0])
+
+    def test_write_aas_missing_concept_description(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            # leave example_concept_description out of object store
+            data = model.DictIdentifiableStore([
+                example_aas.create_example_asset_administration_shell(),
+                example_aas.create_example_submodel(),
+                example_aas.create_example_asset_identification_submodel(),
+                example_aas.create_example_bill_of_material_submodel()
+            ])
+            file_store = aasx.DictSupplementaryFileContainer()
+            with open(Path(__file__).parent / "TestFile.pdf", "rb") as pdf:
+                file_store.add_file("/TestFile.pdf", pdf, "application/pdf")
+
+            # ---- Act & Assert ----
+            # assert warning is present in failsafe mode
+            with self.assertLogs(level="WARNING") as log:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=True) as writer:
+                    writer.write_aas("https://example.org/Test_AssetAdministrationShell", data, file_store)
+            self.assertIn("https://example.org/Test_ConceptDescription", log.output[0])
+            self.assertRegex(log.output[0], "ConceptDescription .* not found")
+
+            # assert exception is rose in non-failsafe mode
+            with self.assertRaises(KeyError) as cm:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=False) as writer:
+                    writer.write_aas("https://example.org/Test_AssetAdministrationShell", data, file_store)
+            self.assertIn("https://example.org/Test_ConceptDescription", cm.exception.args[0])
+            self.assertRegex(cm.exception.args[0], "ConceptDescription .* not found")
+
+    def test_write_aas_false_semantic_id(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arange ----
+            # semanticId of submodel holds reference to an object
+            # that is no ContentDescription
+            second_submodel = model.Submodel(
+                id_="https://example.org/Second_Submodel"
+            )
+            submodel = model.Submodel(
+                id_="https://example.org/Test_Submodel",
+                semantic_id=model.ModelReference(
+                    key=(model.Key(type_=model.KeyTypes.SUBMODEL, value="https://example.org/Second_Submodel"),),
+                    type_=model.ConceptDescription
+                )
+            )
+            data = model.DictIdentifiableStore([
+                example_aas.create_example_asset_administration_shell(),
+                example_aas.create_example_asset_identification_submodel(),
+                example_aas.create_example_bill_of_material_submodel(),
+                submodel, second_submodel
+            ])
+            empty_file_store = aasx.DictSupplementaryFileContainer()
+
+            # ---- Act & Assert ----
+            # assert warning is present in failsafe mode
+            with self.assertLogs(level="WARNING") as log:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=True) as writer:
+                    writer.write_aas("https://example.org/Test_AssetAdministrationShell", data, empty_file_store)
+            self.assertIn("which is not a ConceptDescription", log.output[0])
+
+            # assert exception is rose in non-failsafe mode
+            with self.assertRaises(TypeError) as cm:
+                with aasx.AASXWriter(tmpdir_path / "tmp.aasx", failsafe=False) as writer:
+                    writer.write_aas("https://example.org/Test_AssetAdministrationShell", data, empty_file_store)
+            self.assertIn("which is not a ConceptDescription", cm.exception.args[0])
+
+    def test_write_core_properties_twice(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arrange ----
+            cp = pyecma376_2.OPCCoreProperties()
+            cp.created = datetime.datetime.now()
+            cp.creator = "Eclipse BaSyx Python Testing Framework"
+
+            # ---- Act & Assert ----
+            with aasx.AASXWriter(tmpdir_path / "tmp.aasx") as writer:
+                writer.write_core_properties(cp)
+
+                # expect RuntimeError on second write
+                with self.assertRaises(RuntimeError) as cm:
+                    writer.write_core_properties(cp)
+
+            self.assertIn("Core Properties have already been written", cm.exception.args[0])
+
+    def test_write_thumbnail_twice(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir_path = Path(tmpdir)
+
+            # ---- Arrange ----
+            with open(Path(__file__).parent / "test.png", "rb") as png:
+                thumbnail = png.read()
+
+            # ---- Act & Assert ----
+            with aasx.AASXWriter(tmpdir_path / "tmp.aasx") as writer:
+                writer.write_thumbnail("/aasx/thumbnail.png", bytearray(thumbnail), "image/png")
+
+                # expect RuntimeError on second write
+                with self.assertRaises(RuntimeError) as cm:
+                    writer.write_thumbnail("/aasx/thumbnail.png", bytearray(thumbnail), "image/png")
+
+            self.assertIn("package thumbnail has already been written", cm.exception.args[0])
+
     def test_writing_reading_example_aas(self) -> None:
         # Create example data and file_store
         data = example_aas.create_full_example()    # creates a complete, valid example AAS
@@ -147,7 +391,7 @@ def test_writing_reading_example_aas(self) -> None:
                 file_content = io.BytesIO()
                 new_files.write_file("/TestFile.pdf", file_content)
                 self.assertEqual(hashlib.sha1(file_content.getvalue()).hexdigest(),
-                                 "78450a66f59d74c073bf6858db340090ea72a8b1")
+                                 "241e62aef8b4cdad0975f6c68a4ed8b3923d8db1")
 
                 os.unlink(filename)
 
@@ -207,6 +451,50 @@ def test_reading_core_properties(self) -> None:
         finally:
             os.unlink(filename)
 
+    def test_get_thumbnail(self) -> None:
+        with tempfile.TemporaryDirectory() as tmpdir:
+            # ---- Arange ----
+            tmpdir_path = Path(tmpdir)
+
+            data: model.DictIdentifiableStore[model.Identifiable] = model.DictIdentifiableStore([
+                model.AssetAdministrationShell(
+                    id_="http://example.org/Test_AAS",
+                    asset_information=model.AssetInformation(
+                        global_asset_id="http://example.org/Test_Asset"
+                    )
+                )
+            ])
+
+            with aasx.AASXWriter(tmpdir_path / "test_thumbnail.aasx") as writer:
+                writer.write_aas(
+                    'http://example.org/Test_AAS',
+                    data, aasx.DictSupplementaryFileContainer(), write_json=False
+                )
+                with open(Path(__file__).parent / "test.png", "rb") as png:
+                    thumbnail = png.read()
+                    writer.write_thumbnail("/aasx/thumbnail.png", bytearray(thumbnail), "image/png")
+
+            # ---- Act ----
+            with aasx.AASXReader(tmpdir_path / "test_thumbnail.aasx") as reader:
+                new_thumbnail = reader.get_thumbnail()
+
+            # ---- Assert ----
+            self.assertEqual(new_thumbnail, thumbnail)
+
+    def test_missing_thumbnail(self) -> None:
+        # ---- Arange ----
+        filename = self._create_test_aasx()
+
+        try:
+            # ---- Act ----
+            with aasx.AASXReader(filename) as reader:
+                thumbnail = reader.get_thumbnail()
+
+            # ---- Assert ----
+            self.assertIsNone(thumbnail)
+        finally:
+            os.unlink(filename)
+
     def test_read_into(self) -> None:
         filename = self._create_test_aasx()
 
@@ -246,7 +534,7 @@ def test_supplementary_file_integrity(self) -> None:
 
             self.assertEqual(
                 hashlib.sha1(buf.getvalue()).hexdigest(),
-                "78450a66f59d74c073bf6858db340090ea72a8b1"
+                "241e62aef8b4cdad0975f6c68a4ed8b3923d8db1"
             )
         finally:
             os.unlink(filename)