diff options
Diffstat (limited to 'functest/core')
| -rw-r--r-- | functest/core/feature.py | 149 | ||||
| -rw-r--r-- | functest/core/pytest_suite_runner.py | 46 | ||||
| -rw-r--r-- | functest/core/testcase.py | 127 | ||||
| -rw-r--r-- | functest/core/vnf.py (renamed from functest/core/vnf_base.py) | 170 |
4 files changed, 305 insertions, 187 deletions
diff --git a/functest/core/feature.py b/functest/core/feature.py index 325c10d4..140c9bb2 100644 --- a/functest/core/feature.py +++ b/functest/core/feature.py @@ -1,63 +1,116 @@ +#!/usr/bin/env python + +# Copyright (c) 2016 ZTE Corp and others. +# +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 + +"""Define the parent class of all Functest Features. + +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 testcase as base +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>, " + "Cedric Ollivier <cedric.ollivier@orange.com>") + 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.case_name) + + def execute(self, **kwargs): + """Execute the Python method. + + The subclasses must override the default implementation which + is false on purpose. + + The new implementation must return 0 if success or anything + else if failure. + + Args: + kwargs: Arbitrary keyword arguments. - def __init__(self, project='functest', case='', repo='', cmd=''): - super(Feature, self).__init__() - self.project_name = project - self.case_name = case - self.cmd = cmd - self.repo = CONST.__getattribute__(repo) - self.result_file = self.get_result_file() - self.logger = ft_logger.Logger(project).getLogger() + Returns: + -1. + """ + # pylint: disable=unused-argument,no-self-use + return -1 def run(self, **kwargs): - self.prepare() + """Run the feature. + + It allows executing any Python method by calling execute(). + + It sets the following attributes required to push the results + to DB: + + * result, + * start_time, + * stop_time. + + It doesn't fulfill details when pushing the results to the DB. + + Args: + kwargs: Arbitrary keyword arguments. + + Returns: + TestCase.EX_OK if execute() returns 0, + TestCase.EX_RUN_ERROR otherwise. + """ self.start_time = time.time() - ret = self.execute() + exit_code = base.TestCase.EX_RUN_ERROR + self.result = 0 + try: + if self.execute(**kwargs) == 0: + exit_code = base.TestCase.EX_OK + self.result = 100 + ft_utils.logger_test_results( + self.project_name, self.case_name, + self.result, self.details) + 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.stop_time = time.time() - self.post() - self.parse_results(ret) - self.log_results() - self.logger.info("Test result is stored in '%s'" % self.result_file) - return base.TestCase.EX_OK - - def execute(self): - ''' - Executer method that can be overwritten - By default it executes a shell command. - ''' - return ft_utils.execute_command(self.cmd, output_file=self.result_file) - - def prepare(self, **kwargs): - pass - - def post(self, **kwargs): - pass - - def parse_results(self, ret): - exit_code = base.TestCase.EX_OK - if ret == 0: - self.logger.info("{} OK".format(self.project_name)) - self.criteria = 'PASS' - else: - self.logger.info("{} FAILED".format(self.project_name)) - exit_code = base.TestCase.EX_RUN_ERROR - self.criteria = "FAIL" - return exit_code - def get_result_file(self): - return "{}/{}.log".format(CONST.dir_results, self.project_name) - def log_results(self): - ft_utils.logger_test_results(self.project_name, - self.case_name, - self.criteria, - self.details) +class BashFeature(Feature): + """Class designed to run any bash command.""" + + __logger = logging.getLogger(__name__) + + def execute(self, **kwargs): + """Execute the cmd passed as arg + + Args: + kwargs: Arbitrary keyword arguments. + + Returns: + 0 if cmd returns 0, + -1 otherwise. + """ + ret = -1 + try: + 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) + except Exception: # pylint: disable=broad-except + 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 4f777628..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): - super(PyTestSuiteRunner, self).__init__() + + 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.criteria = '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.criteria = '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 b540cfb5..43161525 100644 --- a/functest/core/testcase.py +++ b/functest/core/testcase.py @@ -7,88 +7,149 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 -"""Define the parent class of Functest TestCase.""" +"""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>" class TestCase(object): - """Parent class of Functest TestCase.""" + """Base model for single test case.""" EX_OK = os.EX_OK + """everything is OK""" + EX_RUN_ERROR = os.EX_SOFTWARE + """run() failed""" + EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1 + """push_to_db() failed""" + EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2 + """results are false""" - logger = ft_logger.Logger(__name__).getLogger() + __logger = logging.getLogger(__name__) - def __init__(self): + def __init__(self, **kwargs): self.details = {} - self.project_name = "functest" - self.case_name = "" - self.criteria = "" - self.start_time = "" - self.stop_time = "" + self.project_name = kwargs.get('project_name', 'functest') + self.case_name = kwargs.get('case_name', '') + self.criteria = kwargs.get('criteria', 100) + 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 check_criteria(self): - """Interpret the results of TestCase. + def get_duration(self): + """Return the duration of the test case. - It allows getting the results of TestCase. It completes run() + 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 is_successful(self): + """Interpret the result of the test case. + + It allows getting the result of TestCase. It completes run() which only returns the execution status. - It can be overriden if checking criteria is not suitable. + It can be overriden if checking result is not suitable. Returns: - TestCase.EX_OK if criteria is 'PASS'. + TestCase.EX_OK if result is 'PASS'. TestCase.EX_TESTCASE_FAILED otherwise. """ try: assert self.criteria - if self.criteria == 'PASS': - return TestCase.EX_OK + 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: + # Backward compatibility + # It must be removed as soon as TestCase subclasses + # stop setting result = 'PASS' or 'FAIL'. + # In this case criteria is unread. + 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): - """Run TestCase. + """Run the test case. It allows running TestCase and getting its execution status. The subclasses must override the default implementation which - is false on purpose. The only prerequisite is to set the - following attributes to push the results to DB: - * case_name, - * criteria, + is false on purpose. + + The new implementation must set the following attributes to + push the results to DB: + + * result, * start_time, * stop_time. Args: - **kwargs: Arbitrary keyword arguments. + kwargs: Arbitrary keyword arguments. Returns: 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): - """Push the results of TestCase to the DB. + """Push the results of the test case to the DB. It allows publishing the results and to check the status. It could be overriden if the common implementation is not suitable. The following attributes must be set before pushing the results to DB: + + * project_name, * case_name, - * criteria, + * result, * start_time, * stop_time. @@ -99,17 +160,19 @@ class TestCase(object): try: assert self.project_name assert self.case_name - assert self.criteria assert self.start_time assert self.stop_time + 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, self.criteria, self.details): - self.logger.info("The results were successfully pushed to DB") + self.stop_time, pub_result, self.details): + 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 3f0adcc6..5667b299 100644 --- a/functest/core/vnf_base.py +++ b/functest/core/vnf.py @@ -8,27 +8,31 @@ # 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, project='functest', case='', repo='', cmd=''): - super(VnfOnBoardingBase, self).__init__() - self.repo = repo - self.project_name = project - self.case_name = case - self.cmd = cmd + def __init__(self, **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') + self.details['prepare_env'] = {} self.details['orchestrator'] = {} self.details['vnf'] = {} self.details['test_vnf'] = {} @@ -40,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 @@ -73,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'] @@ -85,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'] @@ -98,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 @@ -111,59 +115,57 @@ class VnfOnBoardingBase(base.TestCase): # prepare state could consist in the creation of the resources # a dedicated user - # a dedictaed tenant + # a dedicated tenant # dedicated images def prepare(self): 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 admin_user_id == '': - self.step_failure("Failed to get id of " + - self.creds['username']) + if not admin_user_id: + self.step_failure("Failed to get id of {0}".format( + self.creds['username'])) - tenant_id = os_utils.create_tenant( - self.keystone_client, self.tenant_name, self.tenant_description) + tenant_id = os_utils.get_tenant_id(self.keystone_client, + self.tenant_name) if not tenant_id: - self.step_failure("Failed to create " + - self.tenant_name + " tenant") - - roles_name = ["admin", "Admin"] - role_id = '' - for role_name in roles_name: - if role_id == '': - role_id = os_utils.get_role_id(self.keystone_client, role_name) - - if role_id == '': - self.logger.error("Failed to get id for %s role" % role_name) - self.step_failure("Failed to get role id of " + role_name) - - if not os_utils.add_role_user(self.keystone_client, admin_user_id, - role_id, tenant_id): - self.logger.error("Failed to add %s on tenant" % - self.creds['username']) - self.step_failure("Failed to add %s on tenant" % - self.creds['username']) - - user_id = os_utils.create_user(self.keystone_client, - self.tenant_name, - self.tenant_name, - None, - tenant_id) + tenant_id = os_utils.create_tenant(self.keystone_client, + self.tenant_name, + self.tenant_description) + if not tenant_id: + self.step_failure("Failed to get or create {0} tenant".format( + self.tenant_name)) + roles_name = ["admin", "Admin"] + role_id = '' + for role_name in roles_name: + if not role_id: + role_id = os_utils.get_role_id(self.keystone_client, + role_name) + + if not role_id: + self.step_failure("Failed to get id for {0} role".format( + 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'])) + + user_id = os_utils.get_or_create_user(self.keystone_client, + self.tenant_name, + self.tenant_name, + tenant_id) if not user_id: - self.logger.error("Failed to create %s user" % self.tenant_name) - self.step_failure("Failed to create user ") + self.step_failure("Failed to get or create {0} user".format( + self.tenant_name)) - if not os_utils.add_role_user(self.keystone_client, user_id, - role_id, tenant_id): - self.logger.error("Failed to add %s on tenant" % - self.tenant_name) - self.step_failure("Failed to add %s on tenant" % - self.tenant_name) + 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 @@ -176,66 +178,46 @@ class VnfOnBoardingBase(base.TestCase): "password": self.tenant_name, }) - # orchestrator is not mandatory to dpeloy and test VNF + # orchestrator is not mandatory to deploy and test VNF def deploy_orchestrator(self, **kwargs): pass # 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("Removing %s tenant .." % self.tenant_name) - tenant_id = os_utils.get_tenant_id(self.keystone_client, - self.tenant_name) - if tenant_id == '': - self.logger.error("Error : Failed to get id of %s tenant" % - self.tenant_name) - else: - if not os_utils.delete_tenant(self.keystone_client, tenant_id): - self.logger.error("Error : Failed to remove %s tenant" % - self.tenant_name) - - self.logger.info("Removing %s user .." % self.tenant_name) - user_id = os_utils.get_user_id( - self.keystone_client, self.tenant_name) - if user_id == '': - self.logger.error("Error : Failed to get id of %s user" % - self.tenant_name) - else: - if not os_utils.delete_user(self.keystone_client, user_id): - self.logger.error("Error : Failed to remove %s user" % - self.tenant_name) + self.__logger.info("test cleaning") def parse_results(self): exit_code = self.EX_OK - self.criteria = "PASS" - self.logger.info(self.details) + self.result = "PASS" + 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"): exit_code = self.EX_RUN_ERROR - self.criteria = "FAIL" + self.result = "FAIL" return exit_code def log_results(self): ft_utils.logger_test_results(self.project_name, self.case_name, - self.criteria, + self.result, self.details) def step_failure(self, error_msg): part = inspect.stack()[1][3] - self.logger.error("Step '%s' failed: %s", part, error_msg) + self.__logger.error("Step {0} failed: {1}".format(part, error_msg)) try: - part_info = self.details[part] + step_name = self.details_step_mapping[part] + part_info = self.details[step_name] except KeyError: self.details[part] = {} part_info = self.details[part] |