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['orchestrator'] = {}
  31         self.details['vnf'] = {}
  32         self.details['test_vnf'] = {}
  33         self.images = {}
  34         try:
  35             self.tenant_name = CONST.__getattribute__(
  36                 'vnf_{}_tenant_name'.format(self.case_name))
  37             self.tenant_description = CONST.__getattribute__(
  38                 'vnf_{}_tenant_description'.format(self.case_name))
  39         except Exception:
  40             # raise Exception("Unknown VNF case=" + self.case_name)
  41             self.logger.error("Unknown VNF case={}".format(self.case_name))
  42
  43         try:
  44             self.images = CONST.__getattribute__(
  45                 'vnf_{}_tenant_images'.format(self.case_name))
  46         except Exception:
  47             self.logger.warn("No tenant image defined for this VNF")
  48
  49     def execute(self):
  50         self.start_time = time.time()
  51         # Prepare the test (Create Tenant, User, ...)
  52         try:
  53             self.logger.info("Create VNF Onboarding environment")
  54             self.prepare()
  55         except Exception:
  56             self.logger.error("Error during VNF Onboarding environment" +
  57                               "creation", exc_info=True)
  58             return base.TestCase.EX_TESTCASE_FAILED
  59
  60         # Deploy orchestrator
  61         try:
  62             self.logger.info("Deploy orchestrator (if necessary)")
  63             orchestrator_ready_time = time.time()
  64             res_orchestrator = self.deploy_orchestrator()
  65             # orchestrator is not mandatory
  66             if res_orchestrator is not None:
  67                 self.details['orchestrator']['status'] = (
  68                     res_orchestrator['status'])
  69                 self.details['orchestrator']['result'] = (
  70                     res_orchestrator['result'])
  71                 self.details['orchestrator']['duration'] = round(
  72                     orchestrator_ready_time - self.start_time, 1)
  73         except Exception:
  74             self.logger.warn("Problem with the Orchestrator", exc_info=True)
  75
  76         # Deploy VNF
  77         try:
  78             self.logger.info("Deploy VNF " + self.case_name)
  79             res_deploy_vnf = self.deploy_vnf()
  80             vnf_ready_time = time.time()
  81             self.details['vnf']['status'] = res_deploy_vnf['status']
  82             self.details['vnf']['result'] = res_deploy_vnf['result']
  83             self.details['vnf']['duration'] = round(
  84                 vnf_ready_time - orchestrator_ready_time, 1)
  85         except Exception:
  86             self.logger.error("Error during VNF deployment", exc_info=True)
  87             return base.TestCase.EX_TESTCASE_FAILED
  88
  89         # Test VNF
  90         try:
  91             self.logger.info("Test VNF")
  92             res_test_vnf = self.test_vnf()
  93             test_vnf_done_time = time.time()
  94             self.details['test_vnf']['status'] = res_test_vnf['status']
  95             self.details['test_vnf']['result'] = res_test_vnf['result']
  96             self.details['test_vnf']['duration'] = round(
  97                 test_vnf_done_time - vnf_ready_time, 1)
  98         except Exception:
  99             self.logger.error("Error when running VNF tests", exc_info=True)
 100             return base.TestCase.EX_TESTCASE_FAILED
 101
 102         # Clean the system
 103         self.clean()
 104         self.stop_time = time.time()
 105
 106         exit_code = self.parse_results()
 107         self.log_results()
 108         return exit_code
 109
 110     # prepare state could consist in the creation of the resources
 111     # a dedicated user
 112     # a dedictaed tenant
 113     # dedicated images
 114     def prepare(self):
 115         self.creds = os_utils.get_credentials()
 116         self.keystone_client = os_utils.get_keystone_client()
 117
 118         self.logger.info("Prepare OpenStack plateform(create tenant and user)")
 119         admin_user_id = os_utils.get_user_id(self.keystone_client,
 120                                              self.creds['username'])
 121         if admin_user_id == '':
 122             self.step_failure("Failed to get id of " +
 123                               self.creds['username'])
 124
 125         tenant_id = os_utils.create_tenant(
 126             self.keystone_client, self.tenant_name, self.tenant_description)
 127         if not tenant_id:
 128             self.step_failure("Failed to create " +
 129                               self.tenant_name + " tenant")
 130
 131         roles_name = ["admin", "Admin"]
 132         role_id = ''
 133         for role_name in roles_name:
 134             if role_id == '':
 135                 role_id = os_utils.get_role_id(self.keystone_client, role_name)
 136
 137         if role_id == '':
 138             self.logger.error("Failed to get id for %s role" % role_name)
 139             self.step_failure("Failed to get role id of " + role_name)
 140
 141         if not os_utils.add_role_user(self.keystone_client, admin_user_id,
 142                                       role_id, tenant_id):
 143             self.logger.error("Failed to add %s on tenant" %
 144                               self.creds['username'])
 145             self.step_failure("Failed to add %s on tenant" %
 146                               self.creds['username'])
 147
 148         user_id = os_utils.create_user(self.keystone_client,
 149                                        self.tenant_name,
 150                                        self.tenant_name,
 151                                        None,
 152                                        tenant_id)
 153         if not user_id:
 154             self.logger.error("Failed to create %s user" % self.tenant_name)
 155             self.step_failure("Failed to create user ")
 156
 157         if not os_utils.add_role_user(self.keystone_client, user_id,
 158                                       role_id, tenant_id):
 159             self.logger.error("Failed to add %s on tenant" %
 160                               self.tenant_name)
 161             self.step_failure("Failed to add %s on tenant" %
 162                               self.tenant_name)
 163
 164         self.logger.info("Update OpenStack creds informations")
 165         self.admin_creds = self.creds.copy()
 166         self.admin_creds.update({
 167             "tenant": self.tenant_name
 168         })
 169         self.neutron_client = os_utils.get_neutron_client(self.admin_creds)
 170         self.nova_client = os_utils.get_nova_client(self.admin_creds)
 171         self.creds.update({
 172             "tenant": self.tenant_name,
 173             "username": self.tenant_name,
 174             "password": self.tenant_name,
 175         })
 176
 177     # orchestrator is not mandatory to dpeloy and test VNF
 178     def deploy_orchestrator(self, **kwargs):
 179         pass
 180
 181     # TODO see how to use built-in exception from releng module
 182     def deploy_vnf(self):
 183         self.logger.error("VNF must be deployed")
 184         raise Exception("VNF not deployed")
 185
 186     def test_vnf(self):
 187         self.logger.error("VNF must be tested")
 188         raise Exception("VNF not tested")
 189
 190     def clean(self):
 191         self.logger.info("test cleaning")
 192
 193         self.logger.info("Removing %s tenant .." % self.tenant_name)
 194         tenant_id = os_utils.get_tenant_id(self.keystone_client,
 195                                            self.tenant_name)
 196         if tenant_id == '':
 197             self.logger.error("Error : Failed to get id of %s tenant" %
 198                               self.tenant_name)
 199         else:
 200             if not os_utils.delete_tenant(self.keystone_client, tenant_id):
 201                 self.logger.error("Error : Failed to remove %s tenant" %
 202                                   self.tenant_name)
 203
 204         self.logger.info("Removing %s user .." % self.tenant_name)
 205         user_id = os_utils.get_user_id(
 206             self.keystone_client, self.tenant_name)
 207         if user_id == '':
 208             self.logger.error("Error : Failed to get id of %s user" %
 209                               self.tenant_name)
 210         else:
 211             if not os_utils.delete_user(self.keystone_client, user_id):
 212                 self.logger.error("Error : Failed to remove %s user" %
 213                                   self.tenant_name)
 214
 215     def parse_results(self):
 216         exit_code = self.EX_OK
 217         self.criteria = "PASS"
 218         self.logger.info(self.details)
 219         # The 2 VNF steps must be OK to get a PASS result
 220         if (self.details['vnf']['status'] is not "PASS" or
 221                 self.details['test_vnf']['status'] is not "PASS"):
 222             exit_code = self.EX_RUN_ERROR
 223             self.criteria = "FAIL"
 224         return exit_code
 225
 226     def log_results(self):
 227         ft_utils.logger_test_results(self.project_name,
 228                                      self.case_name,
 229                                      self.criteria,
 230                                      self.details)
 231
 232     def step_failure(self, error_msg):
 233         part = inspect.stack()[1][3]
 234         self.logger.error("Step '%s' failed: %s", part, error_msg)
 235         try:
 236             part_info = self.details[part]
 237         except KeyError:
 238             self.details[part] = {}
 239             part_info = self.details[part]
 240         part_info['status'] = 'FAIL'
 241         part_info['result'] = error_msg
 242         raise Exception(error_msg)