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 time
  11
  12 import inspect
  13
  14 import functest.utils.functest_logger as ft_logger
  15 import functest.utils.openstack_utils as os_utils
  16 import functest.utils.functest_utils as ft_utils
  17 import testcase_base as base
  18 from functest.utils.constants import CONST
  19
  20
  21 class VnfOnBoardingBase(base.TestcaseBase):
  22
  23     logger = ft_logger.Logger(__name__).getLogger()
  24
  25     def __init__(self, project='functest', case='', repo='', cmd=''):
  26         super(VnfOnBoardingBase, self).__init__()
  27         self.repo = repo
  28         self.project_name = project
  29         self.case_name = case
  30         self.cmd = cmd
  31         self.details = {}
  32         self.data_dir = CONST.dir_functest_data
  33         self.details['orchestrator'] = {}
  34         self.details['vnf'] = {}
  35         self.details['test_vnf'] = {}
  36         self.images = {}
  37         try:
  38             self.tenant_name = CONST.__getattribute__(
  39                 'vnf_{}_tenant_name'.format(self.case_name))
  40             self.tenant_description = CONST.__getattribute__(
  41                 'vnf_{}_tenant_description'.format(self.case_name))
  42         except:
  43             # raise Exception("Unknown VNF case=" + self.case_name)
  44             self.logger.error("Unknown VNF case={}".format(self.case_name))
  45
  46         try:
  47             self.images = CONST.__getattribute__(
  48                 'vnf_{}_tenant_images'.format(self.case_name))
  49         except:
  50             self.logger.warn("No tenant image defined for this VNF")
  51
  52     def execute(self):
  53         self.start_time = time.time()
  54         # Prepare the test (Create Tenant, User, ...)
  55         try:
  56             self.logger.info("Create VNF Onboarding environment")
  57             self.prepare()
  58         except Exception:
  59             self.logger.error("Error during VNF Onboarding environment" +
  60                               "creation", exc_info=True)
  61             return base.TestcaseBase.EX_TESTCASE_FAILED
  62
  63         # Deploy orchestrator
  64         try:
  65             self.logger.info("Deploy orchestrator (if necessary)")
  66             orchestrator_ready_time = time.time()
  67             res_orchestrator = self.deploy_orchestrator()
  68             # orchestrator is not mandatory
  69             if res_orchestrator is not None:
  70                 self.details['orchestrator']['status'] = (
  71                     res_orchestrator['status'])
  72                 self.details['orchestrator']['result'] = (
  73                     res_orchestrator['result'])
  74                 self.details['orchestrator']['duration'] = round(
  75                     orchestrator_ready_time - self.start_time, 1)
  76         except Exception:
  77             self.logger.warn("Problem with the Orchestrator", exc_info=True)
  78
  79         # Deploy VNF
  80         try:
  81             self.logger.info("Deploy VNF " + self.case_name)
  82             res_deploy_vnf = self.deploy_vnf()
  83             vnf_ready_time = time.time()
  84             self.details['vnf']['status'] = res_deploy_vnf['status']
  85             self.details['vnf']['result'] = res_deploy_vnf['result']
  86             self.details['vnf']['duration'] = round(
  87                 vnf_ready_time - orchestrator_ready_time, 1)
  88         except Exception:
  89             self.logger.error("Error during VNF deployment", exc_info=True)
  90             return base.TestcaseBase.EX_TESTCASE_FAILED
  91
  92         # Test VNF
  93         try:
  94             self.logger.info("Test VNF")
  95             res_test_vnf = self.test_vnf()
  96             test_vnf_done_time = time.time()
  97             self.details['test_vnf']['status'] = res_test_vnf['status']
  98             self.details['test_vnf']['result'] = res_test_vnf['result']
  99             self.details['test_vnf']['duration'] = round(
 100                 test_vnf_done_time - vnf_ready_time, 1)
 101         except Exception:
 102             self.logger.error("Error when running VNF tests", exc_info=True)
 103             return base.TestcaseBase.EX_TESTCASE_FAILED
 104
 105         # Clean the system
 106         self.clean()
 107         self.stop_time = time.time()
 108
 109         exit_code = self.parse_results()
 110         self.log_results()
 111         return exit_code
 112
 113     # prepare state could consist in the creation of the resources
 114     # a dedicated user
 115     # a dedictaed tenant
 116     # dedicated images
 117     def prepare(self):
 118         self.creds = os_utils.get_credentials()
 119         self.keystone_client = os_utils.get_keystone_client()
 120
 121         self.logger.info("Prepare OpenStack plateform(create tenant and user)")
 122         admin_user_id = os_utils.get_user_id(self.keystone_client,
 123                                              self.creds['username'])
 124         if admin_user_id == '':
 125             self.step_failure("Failed to get id of " +
 126                               self.creds['username'])
 127
 128         tenant_id = os_utils.create_tenant(
 129             self.keystone_client, self.tenant_name, self.tenant_description)
 130         if not tenant_id:
 131             self.step_failure("Failed to create " +
 132                               self.tenant_name + " tenant")
 133
 134         roles_name = ["admin", "Admin"]
 135         role_id = ''
 136         for role_name in roles_name:
 137             if role_id == '':
 138                 role_id = os_utils.get_role_id(self.keystone_client, role_name)
 139
 140         if role_id == '':
 141             self.logger.error("Failed to get id for %s role" % role_name)
 142             self.step_failure("Failed to get role id of " + role_name)
 143
 144         if not os_utils.add_role_user(self.keystone_client, admin_user_id,
 145                                       role_id, tenant_id):
 146             self.logger.error("Failed to add %s on tenant" %
 147                               self.creds['username'])
 148             self.step_failure("Failed to add %s on tenant" %
 149                               self.creds['username'])
 150
 151         user_id = os_utils.create_user(self.keystone_client,
 152                                        self.tenant_name,
 153                                        self.tenant_name,
 154                                        None,
 155                                        tenant_id)
 156         if not user_id:
 157             self.logger.error("Failed to create %s user" % self.tenant_name)
 158             self.step_failure("Failed to create user ")
 159
 160         if not os_utils.add_role_user(self.keystone_client, user_id,
 161                                       role_id, tenant_id):
 162             self.logger.error("Failed to add %s on tenant" %
 163                               self.tenant_name)
 164             self.step_failure("Failed to add %s on tenant" %
 165                               self.tenant_name)
 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 dpeloy 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     def clean(self):
 194         self.logger.info("test cleaning")
 195
 196         self.logger.info("Removing %s tenant .." % self.tenant_name)
 197         tenant_id = os_utils.get_tenant_id(self.keystone_client,
 198                                            self.tenant_name)
 199         if tenant_id == '':
 200             self.logger.error("Error : Failed to get id of %s tenant" %
 201                               self.tenant_name)
 202         else:
 203             if not os_utils.delete_tenant(self.keystone_client, tenant_id):
 204                 self.logger.error("Error : Failed to remove %s tenant" %
 205                                   self.tenant_name)
 206
 207         self.logger.info("Removing %s user .." % self.tenant_name)
 208         user_id = os_utils.get_user_id(
 209             self.keystone_client, self.tenant_name)
 210         if user_id == '':
 211             self.logger.error("Error : Failed to get id of %s user" %
 212                               self.tenant_name)
 213         else:
 214             if not os_utils.delete_user(self.keystone_client, user_id):
 215                 self.logger.error("Error : Failed to remove %s user" %
 216                                   self.tenant_name)
 217
 218     def parse_results(self):
 219         exit_code = self.EX_OK
 220         self.criteria = "PASS"
 221         self.logger.info(self.details)
 222         # The 2 VNF steps must be OK to get a PASS result
 223         if (self.details['vnf']['status'] is not "PASS" or
 224                 self.details['test_vnf']['status'] is not "PASS"):
 225             exit_code = self.EX_RUN_ERROR
 226             self.criteria = "FAIL"
 227         return exit_code
 228
 229     def log_results(self):
 230         ft_utils.logger_test_results(self.project_name,
 231                                      self.case_name,
 232                                      self.criteria,
 233                                      self.details)
 234
 235     def step_failure(self, error_msg):
 236         part = inspect.stack()[1][3]
 237         self.details[part]['status'] = 'FAIL'
 238         self.details[part]['result'] = error_msg
 239         self.logger.error("Step failure:{}".format(error_msg))
 240         raise Exception(error_msg)