Commit 0e00d3e4 authored by Torben Miny's avatar Torben Miny
Browse files

Merge branch 'fix/compliance_tool' into 'master'

Fix/compliance tool

See merge request !52
parents 0ae7a41d e5f72f00
Pipeline #335194 failed with stage
in 38 seconds
......@@ -15,12 +15,18 @@ Command line script which is a compliance tool for creating and checking json an
examples.data.__init__.py
"""
import argparse
import datetime
import logging
import pyecma376_2
from aas import model
from aas.adapter import aasx
from aas.adapter.xml import write_aas_xml_file
from aas.compliance_tool import compliance_check_json as compliance_tool_json
from aas.compliance_tool import compliance_check_xml as compliance_tool_xml
from aas.compliance_tool import compliance_check_aasx as compliance_tool_aasx
from aas.adapter.json import write_aas_json_file
from aas.examples.data import create_example
from aas.compliance_tool.state_manager import ComplianceToolStateManager, Status
......@@ -32,11 +38,11 @@ def main():
description='Compliance tool for creating and checking json and xml files in compliance with "Details of the '
'Asset Administration Shell" specification of Plattform Industrie 4.0. \n\n'
'This tool has five features: \n'
'1. create an xml or json file with example aas elements\n'
'1. create a xml or json file or an AASX file using xml or json files with example aas elements\n'
'2. check a given xml or json file if it is compliant with the official json or xml aas schema\n'
'3. check if a given xml or json file is deserializable\n'
'4. check if the data in a given xml or json file is the same as the example data\n'
'5. check if two given xml or json files contain the same aas elements in any order\n\n'
'3. check if a given xml, json or aasx file is deserializable\n'
'4. check if the data in a given xml, json or aasx file is the same as the example data\n'
'5. check if two given xml, json or aasx files contain the same aas elements in any order\n\n'
'As a first argument, the feature must be specified (create, schema, deserialization, example, '
'files) or in short (c, s, d, e or f).\n'
'Depending the chosen feature, different additional arguments must be specified:\n'
......@@ -44,9 +50,11 @@ def main():
'schema or s: file to be checked (file_1)\n'
'deserialization or d: file to be checked (file_1)\n'
'example or e: file to be checked (file_1)\n'
'file_compare or f: files to compare (file_1, file_2)\n'
'file_compare or f: files to compare (file_1, file_2)\n,'
'In any case, it must be specified whether the (given or created) files are json (--json) or '
'xml (--xml).\n\n'
'In all cases except of schema, it could be specified if the (given or created) files should be '
'aasx files or only simple json or xml files (--aasx).\n\n'
'Additionally, the tool offers some extra features for more convenient usage:\n'
'a. Different levels of verbosity:\n'
' Default output is just the status for each step performed. With -v or --verbose, additional '
......@@ -76,6 +84,7 @@ def main():
group.add_argument('--json', help="Use AAS json format when checking or creating files", action='store_true')
group.add_argument('--xml', help="Use AAS xml format when checking or creating files", action='store_true')
parser.add_argument('-l', '--logfile', help="Log file to be created in addition to output to stdout", default=None)
parser.add_argument('--aasx', help="Create or read AASX files", action='store_true')
args = parser.parse_args()
......@@ -92,7 +101,31 @@ def main():
manager.set_step_status(Status.SUCCESS)
try:
manager.add_step('Open file')
if args.json:
if args.aasx:
with aasx.AASXWriter(args.file_1) as writer:
manager.set_step_status(Status.SUCCESS)
manager.add_step('Write data to file')
# Todo add Example TestFile.pdf
files = aasx.DictSupplementaryFileContainer()
# Create OPC/AASX core properties
cp = pyecma376_2.OPCCoreProperties()
cp.created = datetime.datetime.now()
cp.creator = "PyI40AAS Testing Framework"
# Decide wether write json or xml files
if args.json:
write_json = True
elif args.xml:
write_json = False
for identifiable in data:
if isinstance(identifiable, model.AssetAdministrationShell):
writer.write_aas(identifiable.identification, data, files, write_json=write_json)
writer.write_core_properties(cp)
manager.set_step_status(Status.SUCCESS)
elif args.json:
with open(args.file_1, 'w', encoding='utf-8-sig') as file:
manager.set_step_status(Status.SUCCESS)
manager.add_step('Write data to file')
......@@ -113,18 +146,24 @@ def main():
if args.xml:
compliance_tool_xml.check_schema(args.file_1, manager)
elif args.action == 'deserialization' or args.action == 'd':
if args.json:
if args.aasx:
compliance_tool_aasx.check_deserialization(args.file_1, manager)
elif args.json:
compliance_tool_json.check_deserialization(args.file_1, manager)
elif args.xml:
compliance_tool_xml.check_deserialization(args.file_1, manager)
elif args.action == 'example' or args.action == 'e':
if args.json:
if args.aasx:
compliance_tool_aasx.check_aas_example(args.file_1, manager)
elif args.json:
compliance_tool_json.check_aas_example(args.file_1, manager)
elif args.xml:
compliance_tool_xml.check_aas_example(args.file_1, manager)
elif args.action == 'files' or args.action == 'f':
if args.file_2:
if args.json:
if args.aasx:
compliance_tool_aasx.check_aasx_files_equivalence(args.file_1, args.file_2, manager)
elif args.json:
compliance_tool_json.check_json_files_equivalence(args.file_1, args.file_2, manager)
elif args.xml:
compliance_tool_xml.check_xml_files_equivalence(args.file_1, args.file_2, manager)
......
# Copyright 2020 PyI40AAS Contributors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
"""
Module which offers functions to use in a confirmation tool related to AASX files
check_deserialization: Checks if a AASX file can be deserialized
check_aas_example: Checks if a AASX file consist the data of the example data defined in
aas.examples.data.example_aas.py
check_aasx_files_equivalence: Checks if two AASX files have the same data regardless of their order
All functions reports any issues using the given StateManager by adding new steps and associated LogRecords
"""
import json
import logging
from typing import Optional, Tuple
import pyecma376_2
from .. import model
from ..adapter import aasx
from ..adapter.json import json_deserialization, JSON_SCHEMA_FILE
from ..examples.data import example_aas, create_example
from ..examples.data._helper import AASDataChecker
from .state_manager import ComplianceToolStateManager, Status
def check_deserialization(file_path: str, state_manager: ComplianceToolStateManager,
file_info: Optional[str] = None) \
-> Tuple[model.DictObjectStore, aasx.DictSupplementaryFileContainer, pyecma376_2.OPCCoreProperties]:
"""
Read a AASX file and reports any issues using the given StateManager
add the steps: 'Open {} file' and 'Read {} file'
:param file_path: given file which should be deserialized
:param state_manager: manager to log the steps
:param file_info: additional information about the file for name of the steps
:return: returns the read object store
"""
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)
if file_info:
state_manager.add_step('Open {} file'.format(file_info))
else:
state_manager.add_step('Open file')
try:
# open given file
obj_store: model.DictObjectStore[model.Identifiable] = model.DictObjectStore()
files = aasx.DictSupplementaryFileContainer()
with aasx.AASXReader(file_path) as reader:
reader.read_into(obj_store, files)
new_cp = reader.get_core_properties()
except ValueError as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
state_manager.set_step_status_from_log()
return model.DictObjectStore(), aasx.DictSupplementaryFileContainer(), pyecma376_2.OPCCoreProperties()
state_manager.set_step_status_from_log()
return obj_store, files, new_cp
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 StateManager
calls the check_deserialization and add the steps: 'Check if data is equal to example data'
:param file_path: given file which should be checked
:param state_manager: manager to log the steps
"""
# create handler to get logger info
logger_example = logging.getLogger(example_aas.__name__)
logger_example.addHandler(state_manager)
logger_example.propagate = False
logger_example.setLevel(logging.INFO)
obj_store, files, cp = check_deserialization(file_path, state_manager)
if state_manager.status in (Status.FAILED, Status.NOT_EXECUTED):
state_manager.add_step('Check if data is equal to example data')
state_manager.set_step_status(Status.NOT_EXECUTED)
return
checker = AASDataChecker(raise_immediately=False)
state_manager.add_step('Check if data is equal to example data')
checker.check_object_store(obj_store, create_example())
state_manager.add_log_records_from_data_checker(checker)
def check_aasx_files_equivalence(file_path_1: str, file_path_2: str, state_manager: ComplianceToolStateManager) -> None:
"""
Checks if two aasx files contain the same elements in any order and reports any issues using the given StateManager
calls the check_deserialization for ech file and add the steps: 'Check if data in files are equal'
:param file_path_1: given first file which should be checked
:param file_path_2: given second file which should be checked
:param state_manager: manager to log the steps
"""
logger = logging.getLogger('compliance_check')
logger.addHandler(state_manager)
logger.propagate = False
logger.setLevel(logging.INFO)
obj_store_1, files_1, cp_1 = check_deserialization(file_path_1, state_manager, 'first')
obj_store_2, files_2, cp_2 = check_deserialization(file_path_2, state_manager, 'second')
if state_manager.status is Status.FAILED:
state_manager.add_step('Check if data in files are equal')
state_manager.set_step_status(Status.NOT_EXECUTED)
return
checker = AASDataChecker(raise_immediately=False)
try:
state_manager.add_step('Check if data in files are equal')
checker.check_object_store(obj_store_1, obj_store_2)
except (KeyError, AssertionError) as error:
state_manager.set_step_status(Status.FAILED)
logger.error(error)
return
state_manager.add_log_records_from_data_checker(checker)
......@@ -25,9 +25,10 @@ All functions reports any issues using the given StateManager by adding new step
"""
import json
import logging
from typing import Optional
from typing import Optional, Tuple
from .. import model
from ..adapter import aasx
from ..adapter.json import json_deserialization, JSON_SCHEMA_FILE
from ..examples.data import example_aas, create_example
from ..examples.data._helper import AASDataChecker
......
# Copyright 2020 PyI40AAS Contributors
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
# specific language governing permissions and limitations under the License.
import os
import unittest
import aas.compliance_tool.compliance_check_aasx as compliance_tool
from aas.compliance_tool.state_manager import ComplianceToolStateManager, Status
class ComplianceToolAASXTest(unittest.TestCase):
def test_check_deserialization(self) -> None:
manager = ComplianceToolStateManager()
script_dir = os.path.dirname(__file__)
file_path_1 = os.path.join(script_dir, 'files/test_not_found.aasx')
compliance_tool.check_deserialization(file_path_1, manager)
self.assertEqual(1, len(manager.steps))
self.assertEqual(Status.FAILED, manager.steps[0].status)
self.assertIn("is not a valid ECMA376-2 (OPC) file", manager.format_step(0, verbose_level=1))
# Todo add more tests for checking wrong aasx files
manager.steps = []
file_path_5 = os.path.join(script_dir, 'files/test_demo_full_example.aasx')
compliance_tool.check_deserialization(file_path_5, manager)
self.assertEqual(1, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
def test_check_aas_example(self) -> None:
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_aas_example(file_path_2, manager)
self.assertEqual(2, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
# Todo update AASX library for saving also sm and cds which are not linked to an aas
self.assertEqual(Status.FAILED, manager.steps[1].status)
manager.steps = []
file_path_3 = os.path.join(script_dir, 'files/test_demo_full_example_wrong_attribute.aasx')
compliance_tool.check_aas_example(file_path_3, manager)
self.assertEqual(2, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.FAILED, manager.steps[1].status)
self.assertIn('Attribute id_short of AssetAdministrationShell[Identifier(IRI=https://acplt.org/'
'Test_AssetAdministrationShell)] must be == TestAssetAdministrationShell',
manager.format_step(1, verbose_level=1))
def test_check_aasx_files_equivalence(self) -> None:
manager = ComplianceToolStateManager()
script_dir = os.path.dirname(__file__)
file_path_1 = os.path.join(script_dir, 'files/test_demo_full_example.aasx')
file_path_2 = os.path.join(script_dir, 'files/test_empty.aasx')
compliance_tool.check_aasx_files_equivalence(file_path_1, file_path_2, manager)
self.assertEqual(3, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
self.assertEqual(Status.FAILED, manager.steps[2].status)
manager.steps = []
compliance_tool.check_aasx_files_equivalence(file_path_2, file_path_1, manager)
self.assertEqual(3, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
self.assertEqual(Status.FAILED, manager.steps[2].status)
manager.steps = []
file_path_3 = os.path.join(script_dir, 'files/test_demo_full_example.aasx')
file_path_4 = os.path.join(script_dir, 'files/test_demo_full_example.aasx')
compliance_tool.check_aasx_files_equivalence(file_path_3, file_path_4, manager)
self.assertEqual(3, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
self.assertEqual(Status.SUCCESS, manager.steps[2].status)
manager.steps = []
file_path_3 = os.path.join(script_dir, 'files/test_demo_full_example.aasx')
file_path_4 = os.path.join(script_dir, 'files/test_demo_full_example_wrong_attribute.aasx')
compliance_tool.check_aasx_files_equivalence(file_path_3, file_path_4, manager)
self.assertEqual(3, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
self.assertEqual(Status.FAILED, manager.steps[2].status)
self.assertIn('Attribute id_short of AssetAdministrationShell'
'[Identifier(IRI=https://acplt.org/Test_AssetAdministrationShell)] must be ==',
manager.format_step(2, verbose_level=1))
manager.steps = []
compliance_tool.check_aasx_files_equivalence(file_path_4, file_path_3, manager)
self.assertEqual(3, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
self.assertEqual(Status.FAILED, manager.steps[2].status)
self.assertIn('Attribute id_short of AssetAdministrationShell'
'[Identifier(IRI=https://acplt.org/Test_AssetAdministrationShell)] must be ==',
manager.format_step(2, verbose_level=1))
......@@ -98,8 +98,8 @@ class ComplianceToolJsonTest(unittest.TestCase):
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
manager.steps = []
file_path_4 = os.path.join(script_dir, 'files/test_empty.json')
compliance_tool.check_deserialization(file_path_4, manager)
file_path_5 = os.path.join(script_dir, 'files/test_demo_full_example.json')
compliance_tool.check_deserialization(file_path_5, manager)
self.assertEqual(2, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
......
......@@ -88,8 +88,8 @@ class ComplianceToolXmlTest(unittest.TestCase):
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
manager.steps = []
file_path_4 = os.path.join(script_dir, 'files/test_empty.xml')
compliance_tool.check_deserialization(file_path_4, manager)
file_path_5 = os.path.join(script_dir, 'files/test_demo_full_example.xml')
compliance_tool.check_deserialization(file_path_5, manager)
self.assertEqual(2, len(manager.steps))
self.assertEqual(Status.SUCCESS, manager.steps[0].status)
self.assertEqual(Status.SUCCESS, manager.steps[1].status)
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment