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