Merge pull request #6 from acplt/feature/compliance_tool/aasx_schema_check

Feature/compliance tool/aasx schema check

Author: FrankSchnicke

basyx/aas/compliance_tool/compliance_check_aasx.py

@@ -22,9 +22,12 @@
 import datetime
 import logging
 from typing import Optional, Tuple
+import io
+from lxml import etree  # type: ignore
 
 import pyecma376_2
 
+from . import compliance_check_json, compliance_check_xml
 from .. import model
 from ..adapter import aasx
 from ..examples.data import example_aas, create_example_aas_binding
@@ -90,6 +93,69 @@ def check_deserialization(file_path: str, state_manager: ComplianceToolStateMana
     return obj_store, files, new_cp
 
 
+def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> None:
+    """
+    Checks a given file against the official json schema and reports any issues using the given
+    :class:`~aas.compliance_tool.state_manager.ComplianceToolStateManager`
+
+    Opens the file and checks if the data inside is stored in XML or JSON. Then calls the respective compliance tool
+    schema check
+    """
+    logger = logging.getLogger('compliance_check')
+    logger.addHandler(state_manager)
+    logger.propagate = False
+    logger.setLevel(logging.INFO)
+
+    # create handler to get logger info
+    logger_deserialization = logging.getLogger(aasx.__name__)
+    logger_deserialization.addHandler(state_manager)
+    logger_deserialization.propagate = False
+    logger_deserialization.setLevel(logging.INFO)
+
+    state_manager.add_step('Open file')
+    try:
+        # open given file
+        reader = aasx.AASXReader(file_path)
+        state_manager.set_step_status_from_log()
+    except ValueError as error:
+        logger.error(error)
+        state_manager.set_step_status_from_log()
+        state_manager.add_step('Read file')
+        state_manager.set_step_status(Status.NOT_EXECUTED)
+        return
+
+    try:
+        # read given file (Find XML and JSON parts)
+        state_manager.add_step('Read file')
+        core_rels = reader.reader.get_related_parts_by_type()
+        try:
+            aasx_origin_part = core_rels[aasx.RELATIONSHIP_TYPE_AASX_ORIGIN][0]
+        except IndexError as e:
+            raise ValueError("Not a valid AASX file: aasx-origin Relationship is missing.") from e
+        state_manager.set_step_status(Status.SUCCESS)
+        for aas_part in reader.reader.get_related_parts_by_type(aasx_origin_part)[
+                aasx.RELATIONSHIP_TYPE_AAS_SPEC]:
+            content_type = reader.reader.get_content_type(aas_part)
+            extension = aas_part.split("/")[-1].split(".")[-1]
+            with reader.reader.open_part(aas_part) as p:
+                if content_type.split(";")[0] in (
+                        "text/xml", "application/xml") or content_type == "" and extension == "xml":
+                    logger.debug("Parsing AAS objects from XML stream in OPC part {} ...".format(aas_part))
+                    compliance_check_xml._check_schema(p, state_manager)
+                elif content_type.split(";")[0] == "application/json" \
+                        or content_type == "" and extension == "json":
+                    logger.debug("Parsing AAS objects from JSON stream in OPC part {} ...".format(aas_part))
+                    compliance_check_json._check_schema(io.TextIOWrapper(p, encoding='utf-8-sig'), state_manager)
+                else:
+                    raise ValueError("Could not determine part format of AASX part {} (Content Type: {}, extension: {}"
+                                     .format(aas_part, content_type, extension))
+    except ValueError as error:
+        logger.error(error)
+        state_manager.set_step_status(Status.FAILED)
+    finally:
+        reader.close()
+
+
 def check_aas_example(file_path: str, state_manager: ComplianceToolStateManager) -> None:
     """
     Checks if a file contains all elements of the aas example and reports any issues using the given

basyx/aas/compliance_tool/compliance_check_json.py

@@ -24,7 +24,7 @@
 """
 import json
 import logging
-from typing import Optional
+from typing import Optional, IO
 
 from .. import model
 from ..adapter.json import json_deserialization, JSON_SCHEMA_FILE
@@ -61,6 +61,15 @@ def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> N
         state_manager.add_step('Validate file against official json schema')
         state_manager.set_step_status(Status.NOT_EXECUTED)
         return
+    return _check_schema(file_to_be_checked, state_manager)
+
+
+def _check_schema(file_to_be_checked: IO[str], state_manager: ComplianceToolStateManager):
+    logger = logging.getLogger('compliance_check')
+    logger.addHandler(state_manager)
+    logger.propagate = False
+    logger.setLevel(logging.INFO)
+
     try:
         with file_to_be_checked:
             state_manager.set_step_status(Status.SUCCESS)

basyx/aas/compliance_tool/compliance_check_xml.py

@@ -54,6 +54,7 @@ def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> N
     try:
         # open given file
         file_to_be_checked = open(file_path, 'rb')
+        state_manager.set_step_status(Status.SUCCESS)
     except IOError as error:
         state_manager.set_step_status(Status.FAILED)
         logger.error(error)
@@ -62,20 +63,31 @@ def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> N
         state_manager.add_step('Validate file against official xml schema')
         state_manager.set_step_status(Status.NOT_EXECUTED)
         return
+    return _check_schema(file_to_be_checked, state_manager)
+
+
+def _check_schema(file_to_be_checked, state_manager):
+    logger = logging.getLogger('compliance_check')
+    logger.addHandler(state_manager)
+    logger.propagate = False
+    logger.setLevel(logging.INFO)
+
+    state_manager.add_step('Read file and check if it is conform to the xml syntax')
     try:
-        with file_to_be_checked:
-            state_manager.set_step_status(Status.SUCCESS)
-            # read given file and check if it is conform to the xml syntax
-            state_manager.add_step('Read file and check if it is conform to the xml syntax')
-            parser = etree.XMLParser(remove_blank_text=True, remove_comments=True)
-            etree.parse(file_to_be_checked, parser)
-            state_manager.set_step_status(Status.SUCCESS)
+        # read given file and check if it is conform to the xml syntax
+        parser = etree.XMLParser(remove_blank_text=True, remove_comments=True)
+        etree.parse(file_to_be_checked, parser)
+        state_manager.set_step_status(Status.SUCCESS)
     except etree.XMLSyntaxError as error:
         state_manager.set_step_status(Status.FAILED)
         logger.error(error)
         state_manager.add_step('Validate file against official xml schema')
         state_manager.set_step_status(Status.NOT_EXECUTED)
+        file_to_be_checked.close()
         return
+    except Exception:
+        file_to_be_checked.close()
+        raise
 
     # load aas xml schema
     aas_xml_schema = etree.XMLSchema(file=XML_SCHEMA_FILE)
@@ -84,7 +96,8 @@ def check_schema(file_path: str, state_manager: ComplianceToolStateManager) -> N
     state_manager.add_step('Validate file against official xml schema')
     # validate given file against schema
     try:
-        with open(file_path, 'rb') as file_to_be_checked:
+        file_to_be_checked.seek(0)  # Reset reading file offset (cursor) to the beginning of the file
+        with file_to_be_checked:
             etree.parse(file_to_be_checked, parser=parser)
     except etree.ParseError as error:
         state_manager.set_step_status(Status.FAILED)

test/compliance_tool/test_compliance_check_aasx.py

@@ -7,6 +7,7 @@
 # SPDX-License-Identifier: EPL-2.0 OR Apache-2.0
 import os
 import unittest
+import sys
 
 from basyx.aas.compliance_tool import compliance_check_aasx as compliance_tool
 from basyx.aas.compliance_tool.state_manager import ComplianceToolStateManager, Status
@@ -130,3 +131,28 @@ def test_check_aasx_files_equivalence(self) -> None:
                       '[Identifier(IRI=https://acplt.org/Test_AssetAdministrationShell)] must be ==',
                       manager.format_step(4, verbose_level=1))
         self.assertEqual(Status.NOT_EXECUTED, manager.steps[5].status)
+
+    @unittest.skipIf(sys.version_info < (3, 7), "The XML schema check fails for Python <= 3.6")
+    def test_check_schema(self):
+        manager = ComplianceToolStateManager()
+        script_dir = os.path.dirname(__file__)
+
+        file_path_2 = os.path.join(script_dir, 'files/test_demo_full_example.aasx')
+        compliance_tool.check_schema(file_path_2, manager)
+        self.assertEqual(10, len(manager.steps))
+        for i in range(10):
+            self.assertEqual(Status.SUCCESS, manager.steps[i].status)
+
+        manager.steps = []
+        file_path_3 = os.path.join(script_dir, 'files/test_demo_full_example2.aasx')
+        compliance_tool.check_schema(file_path_3, manager)
+        self.assertEqual(4, len(manager.steps))
+        for i in range(4):
+            self.assertEqual(Status.SUCCESS, manager.steps[i].status)
+
+        manager.steps = []
+        file_path_4 = os.path.join(script_dir, 'files/test_demo_full_example_wrong_attribute.aasx')
+        compliance_tool.check_schema(file_path_4, manager)
+        self.assertEqual(10, len(manager.steps))
+        for i in range(10):
+            self.assertEqual(Status.SUCCESS, manager.steps[i].status)