Source code for fastr.utils.verify
import os
import gzip
import fastr
from ..abc.serializable 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
log.info('Trying to read file with compression {}'.format('ON' if compressed else 'OFF'))
if compressed:
try:
with gzip.open(filename, 'r') as file_handle:
data = file_handle.read()
except:
log.error('Problem reading gzipped file: {}'.format(filename))
return None
else:
try:
with open(filename, 'r') as file_handle:
data = file_handle.read()
except:
log.error('Problem reading normal file: {}'.format(filename))
return None
log.info('Read data from file successfully')
# Try to read tool doc based on serializer matching the extension
serializer = ext[1:]
log.info('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
log.info('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
log.info(f'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
log.info('Loaded tool {} successfully'.format(tool))
if perform_tests:
log.info('Testing tool...')
try:
tool.test()
except fastr.exceptions.FastrValueError as e:
log.error('Tool is not valid: {}'.format(e))
return tool