diff --git a/sdk/test/adapter/aasx/TestFile.pdf b/sdk/test/adapter/aasx/TestFile.pdf index 2bccbec5..4c97666a 100644 Binary files a/sdk/test/adapter/aasx/TestFile.pdf and b/sdk/test/adapter/aasx/TestFile.pdf differ diff --git a/sdk/test/adapter/aasx/test.png b/sdk/test/adapter/aasx/test.png new file mode 100644 index 00000000..58634849 Binary files /dev/null and b/sdk/test/adapter/aasx/test.png differ diff --git a/sdk/test/adapter/aasx/test_aasx.py b/sdk/test/adapter/aasx/test_aasx.py index d4b0ff24..271b992c 100644 --- a/sdk/test/adapter/aasx/test_aasx.py +++ b/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) diff --git a/sdk/test/adapter/test_load_directory.py b/sdk/test/adapter/test_load_directory.py new file mode 100644 index 00000000..46b87f5d --- /dev/null +++ b/sdk/test/adapter/test_load_directory.py @@ -0,0 +1,100 @@ +import unittest +import tempfile +from pathlib import Path + +from basyx.aas import model +from basyx.aas import adapter + + +class LoadDirectoryTest(unittest.TestCase): + def test_reading_all_files(self): + # ----- Arange ---- + # Create an AAS per file type + json_aas = model.AssetAdministrationShell( + id_="http://example.org/JSON_AAS", + asset_information=model.AssetInformation( + global_asset_id="http://example.org/JSON_Asset" + ) + ) + + xml_aas = model.AssetAdministrationShell( + id_="http://example.org/XML_AAS", + asset_information=model.AssetInformation( + global_asset_id="http://example.org/XML_Asset" + ) + ) + + aasx_aas = model.AssetAdministrationShell( + id_="http://example.org/aasx_AAS", + asset_information=model.AssetInformation( + global_asset_id="http://example.org/aasx_Asset" + ) + ) + + # load TestFile.pdf to save into aasx + file_container = adapter.aasx.DictSupplementaryFileContainer() + with open(Path(__file__).parent / "aasx" / "TestFile.pdf", "rb") as pdf: + resulting_file_name = file_container.add_file( + "/aasx/suppl/file.pdf", pdf, "application/json") + + # create submodel for aasx_aas that refers to pdf + sm_with_file = model.Submodel( + id_="http://example.org/tmp_Submodel", + submodel_element={ + model.File( + id_short="SampleFile", + content_type="application/json", + value=resulting_file_name + ) + } + ) + aasx_aas.submodel.add(model.ModelReference.from_referable(sm_with_file)) + + with tempfile.TemporaryDirectory() as temp_dir: + temp_dir_path = Path(temp_dir) + + # save to json file + adapter.json.write_aas_json_file(temp_dir_path / "testAAS.json", + model.DictIdentifiableStore([json_aas])) + # save to xml file + adapter.xml.write_aas_xml_file(temp_dir_path / "testAAS.xml", + model.DictIdentifiableStore([xml_aas])) + # save to aasx file + with adapter.aasx.AASXWriter(temp_dir_path / "testAAS.aasx") as writer: + writer.write_aas( + aas_ids=["http://example.org/aasx_AAS"], + object_store=model.DictIdentifiableStore([aasx_aas, sm_with_file]), + file_store=file_container + ) + + # ---- Act ---- + new_object_store, new_file_store = adapter.load_directory(temp_dir_path) + + # ---- Assert ----- + # check for all three AAS + self.assertIn("http://example.org/JSON_AAS", new_object_store) + self.assertIn("http://example.org/XML_AAS", new_object_store) + self.assertIn("http://example.org/aasx_AAS", new_object_store) + + # check pdf is loaded + self.assertIn(resulting_file_name, new_file_store) + + def test_skipping_other_files(self): + with tempfile.TemporaryDirectory() as tmp_dir: + # ---- Arange ---- + tmp_dir_path = Path(tmp_dir) + + # create empty file + open(tmp_dir_path / "test.txt", "a").close() + + # create directory + (tmp_dir_path / "empty").mkdir() + + # ---- Act ---- + # assert no exception is occurring + object_store, file_store = adapter.load_directory(tmp_dir_path) + + # ---- Assert ---- + # check stores are empty + self.assertEqual(len(object_store), 0) + self.assertEqual(len(file_store), 0)