X-Git-Url: https://gerrit.opnfv.org/gerrit/gitweb?a=blobdiff_plain;f=functest%2Fcore%2Fvnf.py;h=a6afd4e6b0e690c7576c6399ec5a1fe7446ed58a;hb=383ed86699816f633f007c515338d0fa022bf4fc;hp=5667b2997165d92b7c3fe08763d2efb89aefae7a;hpb=8e6892b8fd6a2e28915c903baf1544c455dc881a;p=functest.git diff --git a/functest/core/vnf.py b/functest/core/vnf.py index 5667b2997..a6afd4e6b 100644 --- a/functest/core/vnf.py +++ b/functest/core/vnf.py @@ -7,220 +7,181 @@ # 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 uuid + +from snaps.config.user import UserConfig +from snaps.config.project import ProjectConfig +from snaps.openstack.create_user import OpenStackUser +from snaps.openstack.create_project import OpenStackProject +from snaps.openstack.utils import keystone_utils +from snaps.openstack.tests import openstack_tests + +from xtesting.core import vnf +from functest.utils import constants + +__author__ = ("Morgan Richomme , " + "Valentin Boucher ") + + +class VnfPreparationException(vnf.VnfPreparationException): + """Raise when VNF preparation cannot be executed.""" + + +class OrchestratorDeploymentException(vnf.OrchestratorDeploymentException): + """Raise when orchestrator cannot be deployed.""" -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 +class VnfDeploymentException(vnf.VnfDeploymentException): + """Raise when VNF cannot be deployed.""" -class VnfOnBoarding(base.TestCase): + +class VnfTestException(vnf.VnfTestException): + """Raise when VNF cannot be tested.""" + + +class VnfOnBoarding(vnf.VnfOnBoarding): + # pylint: disable=too-many-instance-attributes + """Base model for OpenStack 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)) + self.uuid = uuid.uuid4() + self.user_name = "{}-{}".format(self.case_name, self.uuid) + self.tenant_name = "{}-{}".format(self.case_name, self.uuid) + self.snaps_creds = {} + self.created_object = [] + self.os_project = None + self.tenant_description = "Created by OPNFV Functest: {}".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.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) - 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 - - # 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 - - # Clean the system - self.clean() - self.stop_time = time.time() - - exit_code = self.parse_results() - self.log_results() - return exit_code - - # 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 + """ + Prepare the environment for VNF testing: + + * Creation of a user, + * Creation of a tenant, + * Allocation admin role to the user on this tenant + + Returns base.TestCase.EX_OK if preparation is successfull + + Raise VnfPreparationException in case of problem + """ + try: + self.__logger.info( + "Prepare VNF: %s, description: %s", self.case_name, + self.tenant_description) + snaps_creds = openstack_tests.get_credentials( + os_env_file=constants.ENV_FILE) + + self.os_project = OpenStackProject( + snaps_creds, + ProjectConfig( + name=self.tenant_name, + description=self.tenant_description, + domain=snaps_creds.project_domain_name + )) + self.os_project.create() + self.created_object.append(self.os_project) + + snaps_creds.project_domain_id = \ + self.os_project.get_project().domain_id + snaps_creds.user_domain_id = \ + self.os_project.get_project().domain_id + + for role in ['admin', 'Admin']: + if keystone_utils.get_role_by_name( + keystone_utils.keystone_client(snaps_creds), role): + admin_role = role + break + + user_creator = OpenStackUser( + snaps_creds, + UserConfig( + name=self.user_name, + password=str(uuid.uuid4()), + project_name=self.tenant_name, + domain_name=snaps_creds.user_domain_name, + roles={admin_role: self.tenant_name})) + user_creator.create() + self.created_object.append(user_creator) + self.snaps_creds = user_creator.get_os_creds(self.tenant_name) + self.__logger.debug("snaps creds: %s", self.snaps_creds) + + return vnf.VnfOnBoarding.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 this method is overriden then raise orchestratorDeploymentException + if error during orchestrator deployment + """ + self.__logger.info("Deploy orchestrator (if necessary)") + return True + 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 build-in orchestrators such as heat, OpenBaton, cloudify, + juju, onap, ... + + 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") - - 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) + """ + Clean VNF test case. + + 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('Removing the VNF resources ..') + for creator in reversed(self.created_object): + try: + creator.clean() + except Exception as exc: # pylint: disable=broad-except + self.__logger.error('Unexpected error cleaning - %s', exc)