Merge "Refactor core VNF class"

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)