functest/core/vnf_base.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Orange and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 import inspect
  11 import time
  12
  13 import functest.core.testcase as base
  14 from functest.utils.constants import CONST
  15 import functest.utils.functest_logger as ft_logger
  16 import functest.utils.functest_utils as ft_utils
  17 import functest.utils.openstack_utils as os_utils
  18
  19
  20 class VnfOnBoardingBase(base.TestCase):
  21
  22     logger = ft_logger.Logger(__name__).getLogger()
  23
  24     def __init__(self, **kwargs):
  25         super(VnfOnBoardingBase, self).__init__(**kwargs)
  26         self.repo = kwargs.get('repo', '')
  27         self.cmd = kwargs.get('cmd', '')
  28         self.details = {}
  29         self.result_dir = CONST.dir_results
  30         self.details_step_mapping = dict(
  31                     deploy_orchestrator='orchestrator',
  32                     deploy_vnf='vnf',
  33                     test_vnf='test_vnf',
  34                     prepare='prepare_env')
  35         self.details['prepare_env'] = {}
  36         self.details['orchestrator'] = {}
  37         self.details['vnf'] = {}
  38         self.details['test_vnf'] = {}
  39         self.images = {}
  40         try:
  41             self.tenant_name = CONST.__getattribute__(
  42                 'vnf_{}_tenant_name'.format(self.case_name))
  43             self.tenant_description = CONST.__getattribute__(
  44                 'vnf_{}_tenant_description'.format(self.case_name))
  45         except Exception:
  46             # raise Exception("Unknown VNF case=" + self.case_name)
  47             self.logger.error("Unknown VNF case={}".format(self.case_name))
  48
  49         try:
  50             self.images = CONST.__getattribute__(
  51                 'vnf_{}_tenant_images'.format(self.case_name))
  52         except Exception:
  53             self.logger.warn("No tenant image defined for this VNF")
  54
  55     def execute(self):
  56         self.start_time = time.time()
  57         # Prepare the test (Create Tenant, User, ...)
  58         try:
  59             self.logger.info("Create VNF Onboarding environment")
  60             self.prepare()
  61         except Exception:
  62             self.logger.error("Error during VNF Onboarding environment" +
  63                               "creation", exc_info=True)
  64             return base.TestCase.EX_TESTCASE_FAILED
  65
  66         # Deploy orchestrator
  67         try:
  68             self.logger.info("Deploy orchestrator (if necessary)")
  69             orchestrator_ready_time = time.time()
  70             res_orchestrator = self.deploy_orchestrator()
  71             # orchestrator is not mandatory
  72             if res_orchestrator is not None:
  73                 self.details['orchestrator']['status'] = (
  74                     res_orchestrator['status'])
  75                 self.details['orchestrator']['result'] = (
  76                     res_orchestrator['result'])
  77                 self.details['orchestrator']['duration'] = round(
  78                     orchestrator_ready_time - self.start_time, 1)
  79         except Exception:
  80             self.logger.warn("Problem with the Orchestrator", exc_info=True)
  81
  82         # Deploy VNF
  83         try:
  84             self.logger.info("Deploy VNF " + self.case_name)
  85             res_deploy_vnf = self.deploy_vnf()
  86             vnf_ready_time = time.time()
  87             self.details['vnf']['status'] = res_deploy_vnf['status']
  88             self.details['vnf']['result'] = res_deploy_vnf['result']
  89             self.details['vnf']['duration'] = round(
  90                 vnf_ready_time - orchestrator_ready_time, 1)
  91         except Exception:
  92             self.logger.error("Error during VNF deployment", exc_info=True)
  93             return base.TestCase.EX_TESTCASE_FAILED
  94
  95         # Test VNF
  96         try:
  97             self.logger.info("Test VNF")
  98             res_test_vnf = self.test_vnf()
  99             test_vnf_done_time = time.time()
 100             self.details['test_vnf']['status'] = res_test_vnf['status']
 101             self.details['test_vnf']['result'] = res_test_vnf['result']
 102             self.details['test_vnf']['duration'] = round(
 103                 test_vnf_done_time - vnf_ready_time, 1)
 104         except Exception:
 105             self.logger.error("Error when running VNF tests", exc_info=True)
 106             return base.TestCase.EX_TESTCASE_FAILED
 107
 108         # Clean the system
 109         self.clean()
 110         self.stop_time = time.time()
 111
 112         exit_code = self.parse_results()
 113         self.log_results()
 114         return exit_code
 115
 116     # prepare state could consist in the creation of the resources
 117     # a dedicated user
 118     # a dedicated tenant
 119     # dedicated images
 120     def prepare(self):
 121         self.creds = os_utils.get_credentials()
 122         self.keystone_client = os_utils.get_keystone_client()
 123
 124         self.logger.info("Prepare OpenStack plateform(create tenant and user)")
 125         admin_user_id = os_utils.get_user_id(self.keystone_client,
 126                                              self.creds['username'])
 127         if not admin_user_id:
 128             self.step_failure("Failed to get id of {0}".format(
 129                                                 self.creds['username']))
 130
 131         tenant_id = os_utils.get_tenant_id(self.keystone_client,
 132                                            self.tenant_name)
 133         if not tenant_id:
 134             tenant_id = os_utils.create_tenant(self.keystone_client,
 135                                                self.tenant_name,
 136                                                self.tenant_description)
 137             if not tenant_id:
 138                 self.step_failure("Failed to get or create {0} tenant".format(
 139                                                         self.tenant_name))
 140             roles_name = ["admin", "Admin"]
 141             role_id = ''
 142             for role_name in roles_name:
 143                 if not role_id:
 144                     role_id = os_utils.get_role_id(self.keystone_client,
 145                                                    role_name)
 146
 147             if not role_id:
 148                 self.step_failure("Failed to get id for {0} role".format(
 149                                                             role_name))
 150
 151             if not os_utils.add_role_user(self.keystone_client, admin_user_id,
 152                                           role_id, tenant_id):
 153                 self.step_failure("Failed to add {0} on tenant".format(
 154                                                 self.creds['username']))
 155
 156         user_id = os_utils.get_or_create_user(self.keystone_client,
 157                                               self.tenant_name,
 158                                               self.tenant_name,
 159                                               tenant_id)
 160         if not user_id:
 161             self.step_failure("Failed to get or create {0} user".format(
 162                               self.tenant_name))
 163
 164         os_utils.add_role_user(self.keystone_client, user_id,
 165                                role_id, tenant_id)
 166
 167         self.logger.info("Update OpenStack creds informations")
 168         self.admin_creds = self.creds.copy()
 169         self.admin_creds.update({
 170             "tenant": self.tenant_name
 171         })
 172         self.neutron_client = os_utils.get_neutron_client(self.admin_creds)
 173         self.nova_client = os_utils.get_nova_client(self.admin_creds)
 174         self.creds.update({
 175             "tenant": self.tenant_name,
 176             "username": self.tenant_name,
 177             "password": self.tenant_name,
 178         })
 179
 180     # orchestrator is not mandatory to deploy and test VNF
 181     def deploy_orchestrator(self, **kwargs):
 182         pass
 183
 184     # TODO see how to use built-in exception from releng module
 185     def deploy_vnf(self):
 186         self.logger.error("VNF must be deployed")
 187         raise Exception("VNF not deployed")
 188
 189     def test_vnf(self):
 190         self.logger.error("VNF must be tested")
 191         raise Exception("VNF not tested")
 192
 193     # clean before openstack clean run
 194     def clean(self):
 195         self.logger.info("test cleaning")
 196
 197     def parse_results(self):
 198         exit_code = self.EX_OK
 199         self.result = "PASS"
 200         self.logger.info(self.details)
 201         # The 2 VNF steps must be OK to get a PASS result
 202         if (self.details['vnf']['status'] is not "PASS" or
 203                 self.details['test_vnf']['status'] is not "PASS"):
 204             exit_code = self.EX_RUN_ERROR
 205             self.result = "FAIL"
 206         return exit_code
 207
 208     def log_results(self):
 209         ft_utils.logger_test_results(self.project_name,
 210                                      self.case_name,
 211                                      self.result,
 212                                      self.details)
 213
 214     def step_failure(self, error_msg):
 215         part = inspect.stack()[1][3]
 216         self.logger.error("Step {0} failed: {1}".format(part, error_msg))
 217         try:
 218             step_name = self.details_step_mapping[part]
 219             part_info = self.details[step_name]
 220         except KeyError:
 221             self.details[part] = {}
 222             part_info = self.details[part]
 223         part_info['status'] = 'FAIL'
 224         part_info['result'] = error_msg
 225         raise Exception(error_msg)