#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Managing test setup by reading configurations, handling test objects and performing test in class TestManager.
Basic test steps include
* adding test from test configuration files or manually
* adding test objects
* creating the test setup
* running the tests
* returning the test results
"""
import yaml
import csv
import os
import pathlib
import pickle
from .testing import KM3Test, TestCriteria
from .reporting import Reporter
[docs]class TestManager:
"""Class to manage test execution
Objects:
* self.configs: holding configuration for tests
* self.inputs: holding input objects to be tested (e.g. files paths)
* self.tests: full test scenarios with according configuration and inputs (of type KM3Test)
* self.outputs: output objects from tests (e.g. summaries, optional)
"""
def __init__(self):
self.configs = {}
self.inputs = {}
self.tests = {}
self.outputs = {"config": {}}
self.read_standards()
[docs] def add_configuration(self, configname, config):
self.configs.setdefault(configname, config)
[docs] def add_report(self, reportname, reportconfig):
self.outputs["config"].setdefault(reportname, reportconfig)
[docs] def read_standards(self, folderpath = ""):
if not folderpath:
folderpath = os.path.dirname(__file__)+"/testconfigs/"
filelist = os.listdir(folderpath)
for file in filelist:
if file.find(".y")>0:
fshort = file[0:file.rfind(".")]
self.read_configfile(folderpath + file, fshort)
# standard report to return a summary message
self.outputs["config"].setdefault("onlyone", {"datatype": "bool"})
[docs] def read_configfile(self, infilename, prefix = ""):
"""Auxiliary to read configurations from yaml file"""
with open(infilename, 'r') as file:
config = yaml.safe_load(file)
if prefix:
prefix = prefix + "_"
if "tests" in config:
for entry in config["tests"]:
self.add_configuration(prefix+entry, config["tests"][entry])
if "inputs" in config:
for entry in config["inputs"]:
self.add_input(prefix+entry, config["inputs"][entry])
if "reports" in config:
self.outputs["config"] = config["reports"]
[docs] def create(self):
"""Creates KM3Tests from configurations"""
for entry in self.inputs:
iconf = self.inputs[entry]
if not "tests" in iconf:
print ("Need to define tests when adding input")
return
for testconf in iconf["tests"]:
if not testconf in self.configs:
print ("Did not find test configuration for "+testconf+". Will not add test.")
continue
else:
fpath = iconf["filepath"]
# create all input file combinations
if fpath.find("{")>-1:
if not "parameters" in iconf:
print ("Cannot find parameters in for input in configuration.")
return
elif type(iconf["parameters"]) is str:
try:
with open(iconf["parameters"], 'r') as parfile:
parreader = csv.DictReader(parfile)
params = {}
for key in parreader.fieldnames:
params.setdefault(key.strip(), [])
for row in parreader:
for key in parreader.fieldnames:
if row[key].strip() != "":
params[key.strip()].append(row[key].strip())
iconf["parameters"] = params
except:
print ("Failed to read parameter file "+iconf["parameters"])
fnames = self._expand_filelist(fpath, iconf["parameters"])
else:
fnames = [[fpath, {}]]
# create files & tests
for fileinfo in fnames:
config = self.configs[testconf].copy()
config.setdefault("objectfilepath", fileinfo[0])
config.setdefault("objectinfo", fileinfo[1])
if "objecttype" in iconf:
config.setdefault("objecttype", iconf["objecttype"])
elif "objecttype" in self.configs[testconf]:
config.setdefault("objecttype", self.configs[testconf]["objecttype"])
test = KM3Test()
test.set_config(config)
self.tests.setdefault(testconf+"_"+fileinfo[0], test)
@staticmethod
def _expand_filelist(fpath, paramdict):
"""Creating full file list from configuration. Returns list of files and individual configuration parameters"""
paramnames = []
starter = 0
while fpath.find("{", starter) > -1:
paramnames.append(fpath[fpath.find("{", starter)+1:fpath.find("}", starter)])
starter = fpath.find("}", starter) + 1
for param in paramnames:
if not param in paramdict:
print ("Cannot find parameter "+param+" in configuration.")
pcombi = [{}]
counter = 0
for param in paramnames:
for combi in pcombi:
for entry in paramdict[param]:
if not param in combi and len(combi) >= counter:
new = combi.copy()
new.setdefault(param, entry)
pcombi.append(new)
counter += 1
fnames = []
for entry in pcombi:
if len(entry)==len(paramnames):
fname = fpath
for param in entry:
fname = fname.replace("{"+param+"}", str(entry[param]))
if not [fname, entry] in fnames and fname.find("{")<0:
fnames.append([fname, entry])
return fnames
[docs] def run(self):
"""Runs created tests in self.test"""
for key in self.tests:
self.tests[key].perform_test()
[docs] def report(self, display = False):
"""Returns results according to configuration"""
repo = Reporter(self.outputs["config"], self.tests)
repo.create_report()
self.outputs["reports"] = repo
[docs] def save_all(self, filepath):
"""Saves all configurations, tests and reports as python object"""
with open(filepath, 'wb') as output:
pickle.dump(self.configs, output, pickle.HIGHEST_PROTOCOL)
pickle.dump(self.inputs, output, pickle.HIGHEST_PROTOCOL)
pickle.dump(self.tests, output, pickle.HIGHEST_PROTOCOL)
pickle.dump(self.outputs, output, pickle.HIGHEST_PROTOCOL)
[docs] def read_all(self, filepath):
with open(filepath, 'rb') as pickles:
self.configs = pickle.load(pickles)
self.inputs = pickle.load(pickles)
self.tests = pickle.load(pickles)
self.outputs = pickle.load(pickles)