aboutsummaryrefslogtreecommitdiffstats
path: root/functest/core/vnf.py
diff options
context:
space:
mode:
Diffstat (limited to 'functest/core/vnf.py')
-rw-r--r--functest/core/vnf.py334
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)