diff options
Diffstat (limited to 'functest/core/vnf.py')
| -rw-r--r-- | functest/core/vnf.py | 334 |
1 files changed, 148 insertions, 186 deletions
diff --git a/functest/core/vnf.py b/functest/core/vnf.py index 5667b299..0589b5d2 100644 --- a/functest/core/vnf.py +++ b/functest/core/vnf.py @@ -7,220 +7,182 @@ # which accompanies this distribution, and is available at # http://www.apache.org/licenses/LICENSE-2.0 -import inspect +"""Define the parent class of all VNF TestCases.""" + import logging import time import functest.core.testcase as base from functest.utils.constants import CONST -import functest.utils.functest_utils as ft_utils import functest.utils.openstack_utils as os_utils +__author__ = ("Morgan Richomme <morgan.richomme@orange.com>, " + "Valentin Boucher <valentin.boucher@orange.com>") + + +class VnfPreparationException(Exception): + """Raise when VNF preparation cannot be executed.""" + + +class OrchestratorDeploymentException(Exception): + """Raise when orchestrator cannot be deployed.""" + + +class VnfDeploymentException(Exception): + """Raise when VNF cannot be deployed.""" + + +class VnfTestException(Exception): + """Raise when VNF cannot be tested.""" + class VnfOnBoarding(base.TestCase): + """Base model for VNF test cases.""" __logger = logging.getLogger(__name__) 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.__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'] = {} - self.images = {} - try: - self.tenant_name = CONST.__getattribute__( - 'vnf_{}_tenant_name'.format(self.case_name)) - self.tenant_description = CONST.__getattribute__( - '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)) - - try: - self.images = CONST.__getattribute__( - 'vnf_{}_tenant_images'.format(self.case_name)) - except Exception: - self.__logger.warn("No tenant image defined for this VNF") - - def execute(self): + self.tenant_created = False + self.user_created = False + self.tenant_name = CONST.__getattribute__( + 'vnf_{}_tenant_name'.format(self.case_name)) + self.tenant_description = CONST.__getattribute__( + 'vnf_{}_tenant_description'.format(self.case_name)) + + def run(self, **kwargs): + """ + Run of the VNF test case: + + * Deploy an orchestrator if needed (e.g. heat, cloudify, ONAP), + * Deploy the VNF, + * Perform tests on the VNF + + A VNF test case is successfull when the 3 steps are PASS + If one of the step is FAIL, the test case is FAIL + + Returns: + TestCase.EX_OK if result is 'PASS'. + TestCase.EX_TESTCASE_FAILED otherwise. + """ self.start_time = time.time() - # Prepare the test (Create Tenant, User, ...) + try: - self.__logger.info("Create VNF Onboarding environment") self.prepare() - except Exception: - self.__logger.error("Error during VNF Onboarding environment" - "creation", exc_info=True) + if (self.deploy_orchestrator() and + self.deploy_vnf() and + self.test_vnf()): + self.stop_time = time.time() + # Calculation with different weight depending on the steps TODO + self.result = 100 + return base.TestCase.EX_OK + else: + self.result = 0 + return base.TestCase.EX_TESTCASE_FAILED + except Exception: # pylint: disable=broad-except + self.__logger.exception("Exception on VNF testing") return base.TestCase.EX_TESTCASE_FAILED - # Deploy orchestrator - try: - self.__logger.info("Deploy orchestrator (if necessary)") - orchestrator_ready_time = time.time() - res_orchestrator = self.deploy_orchestrator() - # orchestrator is not mandatory - if res_orchestrator is not None: - self.details['orchestrator']['status'] = ( - res_orchestrator['status']) - self.details['orchestrator']['result'] = ( - res_orchestrator['result']) - self.details['orchestrator']['duration'] = round( - orchestrator_ready_time - self.start_time, 1) - except Exception: - self.__logger.warn("Problem with the Orchestrator", exc_info=True) - - # Deploy VNF - try: - 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'] - self.details['vnf']['result'] = res_deploy_vnf['result'] - self.details['vnf']['duration'] = round( - vnf_ready_time - orchestrator_ready_time, 1) - except Exception: - self.__logger.error("Error during VNF deployment", exc_info=True) - return base.TestCase.EX_TESTCASE_FAILED + def prepare(self): + """ + Prepare the environment for VNF testing: - # Test VNF - try: - 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'] - self.details['test_vnf']['result'] = res_test_vnf['result'] - 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) - return base.TestCase.EX_TESTCASE_FAILED + * Creation of a user, + * Creation of a tenant, + * Allocation admin role to the user on this tenant - # Clean the system - self.clean() - self.stop_time = time.time() + Returns base.TestCase.EX_OK if preparation is successfull - exit_code = self.parse_results() - self.log_results() - return exit_code + Raise VnfPreparationException in case of problem + """ + try: + self.__logger.info("Prepare VNF: %s, description: %s", + self.tenant_name, self.tenant_description) + admin_creds = os_utils.get_credentials() + keystone_client = os_utils.get_keystone_client() + self.tenant_created = os_utils.get_or_create_tenant_for_vnf( + keystone_client, self.tenant_name, self.tenant_description) + self.user_created = os_utils.get_or_create_user_for_vnf( + keystone_client, self.tenant_name) + creds = admin_creds.copy() + creds.update({ + "tenant": self.tenant_name, + "username": self.tenant_name, + "password": self.tenant_name + }) + return base.TestCase.EX_OK + except Exception: # pylint: disable=broad-except + self.__logger.exception("Exception raised during VNF preparation") + raise VnfPreparationException + + def deploy_orchestrator(self): + """ + Deploy an orchestrator (optional). + + If function overwritten + raise orchestratorDeploymentException if error during orchestrator + deployment + """ + self.__logger.info("Deploy orchestrator (if necessary)") + return True - # prepare state could consist in the creation of the resources - # a dedicated user - # 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)") - 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'])) - - tenant_id = os_utils.get_tenant_id(self.keystone_client, - self.tenant_name) - if not 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.step_failure("Failed to get or create {0} user".format( - 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.admin_creds = self.creds.copy() - self.admin_creds.update({ - "tenant": self.tenant_name - }) - self.neutron_client = os_utils.get_neutron_client(self.admin_creds) - self.nova_client = os_utils.get_nova_client(self.admin_creds) - self.creds.update({ - "tenant": self.tenant_name, - "username": self.tenant_name, - "password": self.tenant_name, - }) - - # 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): + """ + Deploy the VNF + + This function MUST be implemented by vnf test cases. + The details section MAY be updated in the vnf test cases. + + The deployment can be executed via a specific orchestrator + or using nuild-in orchestrators such as: + + * heat, openbaton, cloudify (available on all scenario), + * open-o (on open-o scenarios) + + Returns: + True if the VNF is properly deployed + False if the VNF is not deployed + + Raise VnfDeploymentException if error during VNF deployment + """ self.__logger.error("VNF must be deployed") - raise Exception("VNF not deployed") + raise VnfDeploymentException def test_vnf(self): + """ + Test the VNF + + This function MUST be implemented by vnf test cases. + The details section MAY be updated in the vnf test cases. + + Once a VNF is deployed, it is assumed that specific test suite can be + run to validate the VNF. + Please note that the same test suite can be used on several test case + (e.g. clearwater test suite can be used whatever the orchestrator used + for the deployment) + + Returns: + True if VNF tests are PASS + False if test suite is FAIL + + Raise VnfTestException if error during VNF test + """ self.__logger.error("VNF must be tested") - raise Exception("VNF not tested") + raise VnfTestException - # clean before openstack clean run def clean(self): - self.__logger.info("test cleaning") + """ + Clean VNF test case. - def parse_results(self): - exit_code = self.EX_OK - 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.result = "FAIL" - return exit_code - - def log_results(self): - ft_utils.logger_test_results(self.project_name, - self.case_name, - self.result, - self.details) - - def step_failure(self, error_msg): - part = inspect.stack()[1][3] - 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] - except KeyError: - self.details[part] = {} - part_info = self.details[part] - part_info['status'] = 'FAIL' - part_info['result'] = error_msg - raise Exception(error_msg) + It is up to the test providers to delete resources used for the tests. + By default we clean: + + * the user, + * the tenant + """ + self.__logger.info("test cleaning") + keystone_client = os_utils.get_keystone_client() + if self.tenant_created: + os_utils.delete_tenant(keystone_client, self.tenant_name) + if self.user_created: + os_utils.delete_user(keystone_client, self.tenant_name) |