Source code for fastr.utils.verify

import os
import gzip

import fastr
from import Serializable
from ..core.tool import Tool

[docs]def verify_resource_loading(filename: str, log=fastr.log): """ Verify that a resource file can be loaded. Returns loaded object. :param filename: path of the object to load :param log: the logger to use to send messages to :return: loaded resource """ name, ext = os.path.splitext(filename) # Check if file is gzipped if ext == '.gz': compressed = True name, ext = os.path.splitext(filename) else: compressed = False # Read file data'Trying to read file with compression {}'.format('ON' if compressed else 'OFF')) if compressed: try: with, 'r') as file_handle: data = except: log.error('Problem reading gzipped file: {}'.format(filename)) return None else: try: with open(filename, 'r') as file_handle: data = except: log.error('Problem reading normal file: {}'.format(filename)) return None'Read data from file successfully') # Try to read tool doc based on serializer matching the extension serializer = ext[1:]'Trying to load file using serializer "{}"'.format(serializer)) if serializer not in Tool.dumpfuncs: log.error('No matching serializer found for "{}"'.format(serializer)) return None load_func = Serializable.dumpfuncs[serializer].loads try: doc = load_func(data) except Exception as exception: log.error('Could not load data using serializer "{}", encountered exception: {}'.format(serializer, exception)) return None return doc
[docs]def verify_tool(filename, log=fastr.log, perform_tests=True): """ Verify that a file """ # Load the file doc = verify_resource_loading(filename, log) if not doc: log.error('Could not load data successfully from {}'.format(filename)) return None # Match the data to the schema for Tools'Validating data against Tool schema') serializer = Tool.get_serializer() try: doc = serializer.instantiate(doc) except Exception as exception: log.error('Encountered a problem when verifying the Tool schema: {}'.format(exception)) return None # Create the Tool object as the final test'Instantiating Tool object') try: tool = Tool(doc) tool.filename = filename except Exception as exception: log.error('Encountered a problem when creating the Tool object: {}'.format(exception)) return None'Loaded tool {} successfully'.format(tool)) if perform_tests:'Testing tool...') try: tool.test() except fastr.exceptions.FastrValueError as e: log.error('Tool is not valid: {}'.format(e)) return tool