diff options
Diffstat (limited to 'functest/core')
| -rw-r--r-- | functest/core/feature.py | 18 | ||||
| -rw-r--r-- | functest/core/pytest_suite_runner.py | 42 | ||||
| -rw-r--r-- | functest/core/testcase.py | 68 | ||||
| -rw-r--r-- | functest/core/vnf.py (renamed from functest/core/vnf_base.py) | 63 |
4 files changed, 127 insertions, 64 deletions
diff --git a/functest/core/feature.py b/functest/core/feature.py index d65f5a3c..140c9bb2 100644 --- a/functest/core/feature.py +++ b/functest/core/feature.py @@ -13,11 +13,11 @@ Feature is considered as TestCase offered by Third-party. It offers helpers to run any python method or any bash command. """ +import logging import time import functest.core.testcase as base import functest.utils.functest_utils as ft_utils -import functest.utils.functest_logger as ft_logger from functest.utils.constants import CONST __author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, " @@ -27,11 +27,12 @@ __author__ = ("Serena Feng <feng.xiaowei@zte.com.cn>, " class Feature(base.TestCase): """Base model for single feature.""" + __logger = logging.getLogger(__name__) + def __init__(self, **kwargs): super(Feature, self).__init__(**kwargs) self.result_file = "{}/{}.log".format( - CONST.__getattribute__('dir_results'), self.project_name) - self.logger = ft_logger.Logger(self.project_name).getLogger() + CONST.__getattribute__('dir_results'), self.case_name) def execute(self, **kwargs): """Execute the Python method. @@ -82,10 +83,9 @@ class Feature(base.TestCase): ft_utils.logger_test_results( self.project_name, self.case_name, self.result, self.details) - self.logger.info("%s %s", self.project_name, self.result) except Exception: # pylint: disable=broad-except - self.logger.exception("%s FAILED", self.project_name) - self.logger.info("Test result is stored in '%s'", self.result_file) + self.__logger.exception("%s FAILED", self.project_name) + self.__logger.info("Test result is stored in '%s'", self.result_file) self.stop_time = time.time() return exit_code @@ -93,6 +93,8 @@ class Feature(base.TestCase): class BashFeature(Feature): """Class designed to run any bash command.""" + __logger = logging.getLogger(__name__) + def execute(self, **kwargs): """Execute the cmd passed as arg @@ -108,7 +110,7 @@ class BashFeature(Feature): cmd = kwargs["cmd"] ret = ft_utils.execute_command(cmd, output_file=self.result_file) except KeyError: - self.logger.error("Please give cmd as arg. kwargs: %s", kwargs) + self.__logger.error("Please give cmd as arg. kwargs: %s", kwargs) except Exception: # pylint: disable=broad-except - self.logger.exception("Execute cmd: %s failed", cmd) + self.__logger.exception("Execute cmd: %s failed", cmd) return ret diff --git a/functest/core/pytest_suite_runner.py b/functest/core/pytest_suite_runner.py index 8b5da05e..a6e47660 100644 --- a/functest/core/pytest_suite_runner.py +++ b/functest/core/pytest_suite_runner.py @@ -5,26 +5,46 @@ # # http://www.apache.org/licenses/LICENSE-2.0 -import testcase as base -import unittest +# pylint: disable=missing-docstring + +import logging import time +import unittest +import six -class PyTestSuiteRunner(base.TestCase): +from functest.core import testcase + + +class PyTestSuiteRunner(testcase.TestCase): """ This superclass is designed to execute pre-configured unittest.TestSuite() objects """ + def __init__(self, **kwargs): super(PyTestSuiteRunner, self).__init__(**kwargs) self.suite = None + self.logger = logging.getLogger(__name__) def run(self, **kwargs): """ Starts test execution from the functest framework """ + try: + name = kwargs["name"] + try: + self.suite = unittest.TestLoader().loadTestsFromName(name) + except ImportError: + self.logger.error("Can not import %s", name) + return testcase.TestCase.EX_RUN_ERROR + except KeyError: + pass self.start_time = time.time() - result = unittest.TextTestRunner(verbosity=2).run(self.suite) + stream = six.StringIO() + result = unittest.TextTestRunner( + stream=stream, verbosity=2).run(self.suite) + self.logger.debug("\n\n%s", stream.getvalue()) self.stop_time = time.time() if result.errors: @@ -44,14 +64,14 @@ class PyTestSuiteRunner(base.TestCase): # we shall distinguish Execution Error from FAIL results # TestCase.EX_RUN_ERROR means that the test case was not run # not that it was run but the result was FAIL - exit_code = base.TestCase.EX_OK - if ((result.errors and len(result.errors) > 0) - or (result.failures and len(result.failures) > 0)): - self.logger.info("%s FAILED" % self.case_name) - self.result = 'FAIL' + exit_code = testcase.TestCase.EX_OK + if ((result.errors and len(result.errors) > 0) or + (result.failures and len(result.failures) > 0)): + self.logger.info("%s FAILED", self.case_name) + self.result = 0 else: - self.logger.info("%s OK" % self.case_name) - self.result = 'PASS' + self.logger.info("%s OK", self.case_name) + self.result = 100 self.details = {} return exit_code diff --git a/functest/core/testcase.py b/functest/core/testcase.py index 3f191b40..43161525 100644 --- a/functest/core/testcase.py +++ b/functest/core/testcase.py @@ -9,9 +9,11 @@ """Define the parent class of all Functest TestCases.""" +import logging import os -import functest.utils.functest_logger as ft_logger +import prettytable + import functest.utils.functest_utils as ft_utils __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>" @@ -32,18 +34,53 @@ class TestCase(object): EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2 """results are false""" - logger = ft_logger.Logger(__name__).getLogger() + __logger = logging.getLogger(__name__) def __init__(self, **kwargs): self.details = {} self.project_name = kwargs.get('project_name', 'functest') self.case_name = kwargs.get('case_name', '') self.criteria = kwargs.get('criteria', 100) - self.result = "" - self.start_time = "" - self.stop_time = "" + self.result = 0 + self.start_time = 0 + self.stop_time = 0 + + def __str__(self): + try: + assert self.project_name + assert self.case_name + result = 'PASS' if(self.is_successful( + ) == TestCase.EX_OK) else 'FAIL' + msg = prettytable.PrettyTable( + header_style='upper', padding_width=5, + field_names=['test case', 'project', 'duration', + 'result']) + msg.add_row([self.case_name, self.project_name, + self.get_duration(), result]) + return msg.get_string() + except AssertionError: + self.__logger.error("We cannot print invalid objects") + return super(TestCase, self).__str__() + + def get_duration(self): + """Return the duration of the test case. + + Returns: + duration if start_time and stop_time are set + "XX:XX" otherwise. + """ + try: + assert self.start_time + assert self.stop_time + if self.stop_time < self.start_time: + return "XX:XX" + return "{0[0]:02.0f}:{0[1]:02.0f}".format(divmod( + self.stop_time - self.start_time, 60)) + except Exception: # pylint: disable=broad-except + self.__logger.error("Please run test before getting the duration") + return "XX:XX" - def check_result(self): + def is_successful(self): """Interpret the result of the test case. It allows getting the result of TestCase. It completes run() @@ -57,7 +94,9 @@ class TestCase(object): """ try: assert self.criteria - if isinstance(self.result, int) and isinstance(self.criteria, int): + assert self.result is not None + if (not isinstance(self.result, str) and + not isinstance(self.criteria, str)): if self.result >= self.criteria: return TestCase.EX_OK else: @@ -65,12 +104,12 @@ class TestCase(object): # It must be removed as soon as TestCase subclasses # stop setting result = 'PASS' or 'FAIL'. # In this case criteria is unread. - self.logger.warning( + self.__logger.warning( "Please update result which must be an int!") if self.result == 'PASS': return TestCase.EX_OK except AssertionError: - self.logger.error("Please run test before checking the results") + self.__logger.error("Please run test before checking the results") return TestCase.EX_TESTCASE_FAILED def run(self, **kwargs): @@ -96,7 +135,7 @@ class TestCase(object): TestCase.EX_RUN_ERROR. """ # pylint: disable=unused-argument - self.logger.error("Run must be implemented") + self.__logger.error("Run must be implemented") return TestCase.EX_RUN_ERROR def push_to_db(self): @@ -123,16 +162,17 @@ class TestCase(object): assert self.case_name assert self.start_time assert self.stop_time - pub_result = 'PASS' if self.check_result( + pub_result = 'PASS' if self.is_successful( ) == TestCase.EX_OK else 'FAIL' if ft_utils.push_results_to_db( self.project_name, self.case_name, self.start_time, self.stop_time, pub_result, self.details): - self.logger.info("The results were successfully pushed to DB") + self.__logger.info( + "The results were successfully pushed to DB") return TestCase.EX_OK else: - self.logger.error("The results cannot be pushed to DB") + self.__logger.error("The results cannot be pushed to DB") return TestCase.EX_PUSH_TO_DB_ERROR except Exception: # pylint: disable=broad-except - self.logger.exception("The results cannot be pushed to DB") + self.__logger.exception("The results cannot be pushed to DB") return TestCase.EX_PUSH_TO_DB_ERROR diff --git a/functest/core/vnf_base.py b/functest/core/vnf.py index fe4e427f..5667b299 100644 --- a/functest/core/vnf_base.py +++ b/functest/core/vnf.py @@ -8,30 +8,30 @@ # http://www.apache.org/licenses/LICENSE-2.0 import inspect +import logging import time import functest.core.testcase as base from functest.utils.constants import CONST -import functest.utils.functest_logger as ft_logger import functest.utils.functest_utils as ft_utils import functest.utils.openstack_utils as os_utils -class VnfOnBoardingBase(base.TestCase): +class VnfOnBoarding(base.TestCase): - logger = ft_logger.Logger(__name__).getLogger() + __logger = logging.getLogger(__name__) def __init__(self, **kwargs): - super(VnfOnBoardingBase, self).__init__(**kwargs) + super(VnfOnBoarding, self).__init__(**kwargs) self.repo = kwargs.get('repo', '') self.cmd = kwargs.get('cmd', '') self.details = {} - self.result_dir = CONST.dir_results + self.result_dir = CONST.__getattribute__('dir_results') self.details_step_mapping = dict( - deploy_orchestrator='orchestrator', - deploy_vnf='vnf', - test_vnf='test_vnf', - prepare='prepare_env') + deploy_orchestrator='orchestrator', + deploy_vnf='vnf', + test_vnf='test_vnf', + prepare='prepare_env') self.details['prepare_env'] = {} self.details['orchestrator'] = {} self.details['vnf'] = {} @@ -44,28 +44,28 @@ class VnfOnBoardingBase(base.TestCase): 'vnf_{}_tenant_description'.format(self.case_name)) except Exception: # raise Exception("Unknown VNF case=" + self.case_name) - self.logger.error("Unknown VNF case={}".format(self.case_name)) + self.__logger.error("Unknown VNF case={}".format(self.case_name)) try: self.images = CONST.__getattribute__( 'vnf_{}_tenant_images'.format(self.case_name)) except Exception: - self.logger.warn("No tenant image defined for this VNF") + self.__logger.warn("No tenant image defined for this VNF") def execute(self): self.start_time = time.time() # Prepare the test (Create Tenant, User, ...) try: - self.logger.info("Create VNF Onboarding environment") + self.__logger.info("Create VNF Onboarding environment") self.prepare() except Exception: - self.logger.error("Error during VNF Onboarding environment" + - "creation", exc_info=True) + self.__logger.error("Error during VNF Onboarding environment" + "creation", exc_info=True) return base.TestCase.EX_TESTCASE_FAILED # Deploy orchestrator try: - self.logger.info("Deploy orchestrator (if necessary)") + self.__logger.info("Deploy orchestrator (if necessary)") orchestrator_ready_time = time.time() res_orchestrator = self.deploy_orchestrator() # orchestrator is not mandatory @@ -77,11 +77,11 @@ class VnfOnBoardingBase(base.TestCase): self.details['orchestrator']['duration'] = round( orchestrator_ready_time - self.start_time, 1) except Exception: - self.logger.warn("Problem with the Orchestrator", exc_info=True) + self.__logger.warn("Problem with the Orchestrator", exc_info=True) # Deploy VNF try: - self.logger.info("Deploy VNF " + self.case_name) + self.__logger.info("Deploy VNF " + self.case_name) res_deploy_vnf = self.deploy_vnf() vnf_ready_time = time.time() self.details['vnf']['status'] = res_deploy_vnf['status'] @@ -89,12 +89,12 @@ class VnfOnBoardingBase(base.TestCase): self.details['vnf']['duration'] = round( vnf_ready_time - orchestrator_ready_time, 1) except Exception: - self.logger.error("Error during VNF deployment", exc_info=True) + self.__logger.error("Error during VNF deployment", exc_info=True) return base.TestCase.EX_TESTCASE_FAILED # Test VNF try: - self.logger.info("Test VNF") + self.__logger.info("Test VNF") res_test_vnf = self.test_vnf() test_vnf_done_time = time.time() self.details['test_vnf']['status'] = res_test_vnf['status'] @@ -102,7 +102,7 @@ class VnfOnBoardingBase(base.TestCase): self.details['test_vnf']['duration'] = round( test_vnf_done_time - vnf_ready_time, 1) except Exception: - self.logger.error("Error when running VNF tests", exc_info=True) + self.__logger.error("Error when running VNF tests", exc_info=True) return base.TestCase.EX_TESTCASE_FAILED # Clean the system @@ -121,12 +121,13 @@ class VnfOnBoardingBase(base.TestCase): self.creds = os_utils.get_credentials() self.keystone_client = os_utils.get_keystone_client() - self.logger.info("Prepare OpenStack plateform(create tenant and user)") + self.__logger.info( + "Prepare OpenStack plateform(create tenant and user)") admin_user_id = os_utils.get_user_id(self.keystone_client, self.creds['username']) if not admin_user_id: self.step_failure("Failed to get id of {0}".format( - self.creds['username'])) + self.creds['username'])) tenant_id = os_utils.get_tenant_id(self.keystone_client, self.tenant_name) @@ -136,7 +137,7 @@ class VnfOnBoardingBase(base.TestCase): self.tenant_description) if not tenant_id: self.step_failure("Failed to get or create {0} tenant".format( - self.tenant_name)) + self.tenant_name)) roles_name = ["admin", "Admin"] role_id = '' for role_name in roles_name: @@ -146,12 +147,12 @@ class VnfOnBoardingBase(base.TestCase): if not role_id: self.step_failure("Failed to get id for {0} role".format( - role_name)) + role_name)) if not os_utils.add_role_user(self.keystone_client, admin_user_id, role_id, tenant_id): self.step_failure("Failed to add {0} on tenant".format( - self.creds['username'])) + self.creds['username'])) user_id = os_utils.get_or_create_user(self.keystone_client, self.tenant_name, @@ -164,7 +165,7 @@ class VnfOnBoardingBase(base.TestCase): os_utils.add_role_user(self.keystone_client, user_id, role_id, tenant_id) - self.logger.info("Update OpenStack creds informations") + self.__logger.info("Update OpenStack creds informations") self.admin_creds = self.creds.copy() self.admin_creds.update({ "tenant": self.tenant_name @@ -183,21 +184,21 @@ class VnfOnBoardingBase(base.TestCase): # TODO see how to use built-in exception from releng module def deploy_vnf(self): - self.logger.error("VNF must be deployed") + self.__logger.error("VNF must be deployed") raise Exception("VNF not deployed") def test_vnf(self): - self.logger.error("VNF must be tested") + self.__logger.error("VNF must be tested") raise Exception("VNF not tested") # clean before openstack clean run def clean(self): - self.logger.info("test cleaning") + self.__logger.info("test cleaning") def parse_results(self): exit_code = self.EX_OK self.result = "PASS" - self.logger.info(self.details) + self.__logger.info(self.details) # The 2 VNF steps must be OK to get a PASS result if (self.details['vnf']['status'] is not "PASS" or self.details['test_vnf']['status'] is not "PASS"): @@ -213,7 +214,7 @@ class VnfOnBoardingBase(base.TestCase): def step_failure(self, error_msg): part = inspect.stack()[1][3] - self.logger.error("Step {0} failed: {1}".format(part, error_msg)) + self.__logger.error("Step {0} failed: {1}".format(part, error_msg)) try: step_name = self.details_step_mapping[part] part_info = self.details[step_name] |