# 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 functest.core import testcase
+
+__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."""
+
-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(Exception):
+ """Raise when VNF cannot be deployed."""
-class VnfOnBoarding(base.TestCase):
+class VnfTestException(Exception):
+ """Raise when VNF cannot be tested."""
+
+
+class VnfOnBoarding(testcase.TestCase):
+ # pylint: disable=too-many-instance-attributes
+ """Base model for VNF test cases."""
__logger = logging.getLogger(__name__)
+ env_file = "/home/opnfv/functest/conf/env_file"
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.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)
+
+ 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)
- return base.TestCase.EX_TESTCASE_FAILED
+ 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 testcase.TestCase.EX_OK
+ self.result = 0
+ self.stop_time = time.time()
+ return testcase.TestCase.EX_TESTCASE_FAILED
+ except Exception: # pylint: disable=broad-except
+ self.stop_time = time.time()
+ self.__logger.exception("Exception on VNF testing")
+ return testcase.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=self.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)
+
+ return testcase.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 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)
Code Review / functest.git / blobdiff