#
# Copyright (C) 2019 Intel Corporation. All rights reserved.
# SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception
#
import datetime
import os
import pprint
import random
import re
import shlex
import subprocess
import signal
import sys
import time
import shutil
from .test_api import *
import this
'''
The run evironment dir structure:
run/
run{date-time}/
suites/
{suite name}/
-- target/ (the target software being tested)
-- tools/ (the tools for testing the target software)
'''
framework=None
def get_framework():
global framework
return framework
def my_import(name):
mod = __import__(name)
components = name.split('.')
for comp in components[1:]:
mod = getattr(mod, comp)
return mod
# we maintain a root path apart from framework location
# so the suites can be located in anywhere
class CTestFramework(object):
def __init__(self, path):
self.running_case = ''
self.running_suite = ''
self.target_suites = {}
self.target_cases = {}
self.root_path = path
self.running_folder=''
self.report = None
self.sucess_cases = 0
self.failed_cases = 0
self.setup_fails = 0
self.load_fails = 0;
global framework
framework = self
api_set_root_path(path)
print "root_path is " + self.root_path
def gen_execution_stats(self):
return '\nTest Execution Summary: ' \
'\n\tSuccess: {}' \
'\n\tCases fails: {}' \
'\n\tSetup fails: {}' \
'\n\tCase load fails: {}'.format(
self.sucess_cases, self.failed_cases, self.setup_fails, self.load_fails)
def report_result(self, success, message, case_description):
if self.report is None:
return
case_pass = "pass"
if not success:
case_pass = "fail"
self.report.write(case_pass + ": [" + self.running_case + "]\n\treason: " + \
message + "\n\tcase: " + case_description + "\n")
return
def get_running_path(self):
return self.root_path + "/run/" + self.running_folder
def load_suites(self):
self.target_suites = os.listdir(self.root_path + "/suites")
return
def run_case(self, suite_instance, case):
# load the test case module
case_description = ''
suite = suite_instance.m_name
api_log("\n>>start run [" + case + "] >>")
module_name = 'suites.' + suite + ".cases." + case + ".case"
try:
module = my_import(module_name)
except Exception, e:
report_fail("load case fail: " + str(e))
api_log_error("load case fail: " + str(e))
self.load_fails = self.load_fails +1
print(traceback.format_exc())
return False
try:
case = module.CTestCase(suite_instance)
except Exception, e:
report_fail("initialize case fail: " + str(e))
api_log_error("initialize case fail: " + str(e))
self.load_fails = self.load_fails +1
return False
# call the case on setup callback
try:
case_description = case.on_get_case_description()
result, message = case.on_setup_case()
except Exception, e:
result = False
message = str(e);
if not result:
api_log_error(message)
report_fail (message, case_description)
self.failed_cases = self.failed_cases+1
return False
# call the case execution callaback
try:
result, message = case.on_run_case()
except Exception, e:
result = False
message = str(e);
if not result:
report_fail (message, case_description)
api_log_error(message)
self.failed_cases = self.failed_cases+1
else:
report_success(case_description)
self.sucess_cases = self.sucess_cases +1
# call the case cleanup callback
try:
clean_result, message = case.on_cleanup_case()
except Exception, e:
clean_result = False
message = str(e)
if not clean_result:
api_log(message)
return result
def run_suite(self, suite, cases):
# suite setup
message = ''
api_log("\n>>> Suite [" + suite + "] starting >>>")
running_folder = self.get_running_path()+ "/suites/" + suite;
module_name = 'suites.' + suite + ".suite_setup"
try:
module = my_import(module_name)
except Exception, e:
report_fail("load suite [" + suite +"] fail: " + str(e))
self.load_fails = self.load_fails +1
return False
try:
suite_instance = module.CTestSuite(suite, \
self.root_path + '/suites/' + suite, running_folder)
except Exception, e:
report_fail("initialize suite fail: " + str(e))
self.load_fails = self.load_fails +1
return False
result, message = suite_instance.load_settings()
if not result:
report_fail("load settings fail: " + str(e))
self.load_fails = self.load_fails +1
return False
try:
result, message = suite_instance.on_suite_setup()
except Exception, e:
result = False
message = str(e);
if not result:
api_log_error(message)
report_fail (message)
self.setup_fails = self.setup_fails + 1
return False
self.running_suite = suite
cases.sort()
# run cases
for case in cases:
if not os.path.isdir(self.root_path + '/suites/' + suite + '/cases/' + case):
continue
self.running_case = case
self.run_case(suite_instance, case)
self.running_case = ''
# suites cleanup
self.running_suite = ''
try:
result, message = suite_instance.on_suite_cleanup()
except Exception, e:
result = False
message = str(e);
if not result:
api_log_error(message)
report_fail (message)
self.setup_fails = self.setup_fails + 1
return
def start_run(self):
if self.target_suites is None:
print "\n\nstart run: no target suites, exit.."
return
cur_time = time.localtime()
time_prefix = "{:02}-{:02}-{:02}-{:02}".format(
cur_time.tm_mon, cur_time.tm_mday, cur_time.tm_hour, cur_time.tm_min)
debug = api_get_value('debug', False)
if debug:
self.running_folder = 'debug'
else:
self.running_folder = 'run-' + time_prefix
folder = self.root_path + "/run/" +self.running_folder;
if os.path.exists(folder):
shutil.rmtree(folder, ignore_errors=True)
if not os.path.exists(folder):
os.makedirs(folder )
os.makedirs(folder + "/suites")
api_init_log(folder + "/test.log")
self.report = open(folder + "/report.txt", 'a')
self.target_suites.sort()
for suite in self.target_suites:
if not os.path.isdir(self.root_path + '/suites/' + suite):
continue
self.report.write("suite " + suite + " cases:\n")
if self.target_cases is None:
cases = os.listdir(self.root_path + "/suites/" + suite + "/cases")
self.run_suite(suite, cases)
else:
self.run_suite(suite, self.target_cases)
self.report.write("\n")
self.report.write("\n\n")
summary = self.gen_execution_stats()
self.report.write(summary);
self.report.flush()
self.report.close()
print summary
def report_fail(message, case_description=''):
global framework
if framework is not None:
framework.report_result(False, message, case_description)
api_log_error(message)
return
def report_success(case_description=''):
global framework
if framework is not None:
framework.report_result(True , "OK", case_description)
return