-
-
-parser = argparse.ArgumentParser()
-parser.add_argument("-t", "--test", dest="test", action='store',
- help="Test case or tier (group of tests) to be executed. "
- "It will run all the test if not specified.")
-parser.add_argument("-n", "--noclean", help="Do not clean OpenStack resources"
- " after running each test (default=false).",
- action="store_true")
-parser.add_argument("-r", "--report", help="Push results to database "
- "(default=false).", action="store_true")
-args = parser.parse_args()
-
-
-""" logging configuration """
-logger = ft_logger.Logger("run_tests").getLogger()
-
-
-""" global variables """
-EXEC_SCRIPT = ("%s/functest/ci/exec_test.sh" % ft_constants.FUNCTEST_REPO_DIR)
-
-# This will be the return code of this script. If any of the tests fails,
-# this variable will change to -1
-
-
-class GlobalVariables:
- EXECUTED_TEST_CASES = []
- OVERALL_RESULT = 0
- CLEAN_FLAG = True
- REPORT_FLAG = False
-
-
-def print_separator(str, count=45):
- line = ""
- for i in range(0, count - 1):
- line += str
- logger.info("%s" % line)
-
-
-def source_rc_file():
- rc_file = ft_constants.OPENSTACK_CREDS
- if not os.path.isfile(rc_file):
- logger.error("RC file %s does not exist..." % rc_file)
- sys.exit(1)
- logger.debug("Sourcing the OpenStack RC file...")
- creds = os_utils.source_credentials(rc_file)
- for key, value in creds.iteritems():
- if re.search("OS_", key):
- if key == 'OS_AUTH_URL':
- ft_constants.OS_AUTH_URL = value
- elif key == 'OS_USERNAME':
- ft_constants.OS_USERNAME = value
- elif key == 'OS_TENANT_NAME':
- ft_constants.OS_TENANT_NAME = value
- elif key == 'OS_PASSWORD':
- ft_constants.OS_PASSWORD = value
- logger.debug("OS_AUTH_URL:%s" % ft_constants.OS_AUTH_URL)
- logger.debug("OS_USERNAME:%s" % ft_constants.OS_USERNAME)
- logger.debug("OS_TENANT_NAME:%s" % ft_constants.OS_TENANT_NAME)
- logger.debug("OS_PASSWORD:%s" % ft_constants.OS_PASSWORD)
-
-
-def generate_os_snapshot():
- os_snapshot.main()
-
-
-def cleanup():
- os_clean.main()
-
-
-def update_test_info(test_name, result, duration):
- for test in GlobalVariables.EXECUTED_TEST_CASES:
- if test['test_name'] == test_name:
- test.update({"result": result,
- "duration": duration})
-
-
-def get_run_dict_if_defined(testname):
- try:
- dict = ft_utils.get_dict_by_test(testname)
- if not dict:
- logger.error("Cannot get {}'s config options".format(testname))
- elif 'run' in dict:
- return dict['run']
- return None
- except Exception:
- logger.exception("Cannot get {}'s config options".format(testname))
- return None
-
-
-def run_test(test, tier_name):
- result_str = "PASS"
- start = datetime.datetime.now()
- test_name = test.get_name()
- logger.info("\n") # blank line
- print_separator("=")
- logger.info("Running test case '%s'..." % test_name)
- print_separator("=")
- logger.debug("\n%s" % test)
-
- if GlobalVariables.CLEAN_FLAG:
- generate_os_snapshot()
-
- flags = (" -t %s" % (test_name))
- if GlobalVariables.REPORT_FLAG:
- flags += " -r"
-
- result = testcase_base.TestcaseBase.EX_RUN_ERROR
- run_dict = get_run_dict_if_defined(test_name)
- if run_dict:
+from functest.utils.constants import CONST
+
+# __name__ cannot be used here
+logger = logging.getLogger('functest.ci.run_tests')
+
+
+class Result(enum.Enum):
+ EX_OK = os.EX_OK
+ EX_ERROR = -1
+
+
+class BlockingTestFailed(Exception):
+ pass
+
+
+class TestNotEnabled(Exception):
+ pass
+
+
+class RunTestsParser(object):
+
+ def __init__(self):
+ self.parser = argparse.ArgumentParser()
+ self.parser.add_argument("-t", "--test", dest="test", action='store',
+ help="Test case or tier (group of tests) "
+ "to be executed. It will run all the test "
+ "if not specified.")
+ self.parser.add_argument("-n", "--noclean", help="Do not clean "
+ "OpenStack resources after running each "
+ "test (default=false).",
+ action="store_true")
+ self.parser.add_argument("-r", "--report", help="Push results to "
+ "database (default=false).",
+ action="store_true")
+
+ def parse_args(self, argv=[]):
+ return vars(self.parser.parse_args(argv))
+
+
+class Runner(object):
+
+ def __init__(self):
+ self.executed_test_cases = []
+ self.overall_result = Result.EX_OK
+ self.clean_flag = True
+ self.report_flag = False
+
+ @staticmethod
+ def print_separator(str, count=45):
+ line = ""
+ for i in range(0, count - 1):
+ line += str
+ logger.info("%s" % line)
+
+ @staticmethod
+ def source_rc_file():
+ rc_file = CONST.__getattribute__('openstack_creds')
+ if not os.path.isfile(rc_file):
+ raise Exception("RC file %s does not exist..." % rc_file)
+ logger.debug("Sourcing the OpenStack RC file...")
+ os_utils.source_credentials(rc_file)
+ for key, value in os.environ.iteritems():
+ if re.search("OS_", key):
+ if key == 'OS_AUTH_URL':
+ CONST.__setattr__('OS_AUTH_URL', value)
+ elif key == 'OS_USERNAME':
+ CONST.__setattr__('OS_USERNAME', value)
+ elif key == 'OS_TENANT_NAME':
+ CONST.__setattr__('OS_TENANT_NAME', value)
+ elif key == 'OS_PASSWORD':
+ CONST.__setattr__('OS_PASSWORD', value)
+
+ @staticmethod
+ def generate_os_snapshot():
+ os_snapshot.main()
+
+ @staticmethod
+ def cleanup():
+ os_clean.main()
+
+ @staticmethod
+ def get_run_dict(testname):