# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
-import os
-import time
-
import inspect
+import logging
+import time
-
-import functest.utils.functest_logger as ft_logger
-import functest.utils.openstack_utils as os_utils
-import functest.utils.functest_utils as ft_utils
-import testcase_base as base
+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
-class VnfOnBoardingBase(base.TestcaseBase):
+class VnfOnBoardingBase(base.TestCase):
- logger = ft_logger.Logger(__name__).getLogger()
+ logger = logging.getLogger(__name__)
- def __init__(self, project='functest', case='', repo='', cmd=''):
- super(VnfOnBoardingBase, self).__init__()
- self.repo = repo
- self.project_name = project
- self.case_name = case
- self.cmd = cmd
+ def __init__(self, **kwargs):
+ super(VnfOnBoardingBase, self).__init__(**kwargs)
+ self.repo = kwargs.get('repo', '')
+ self.cmd = kwargs.get('cmd', '')
self.details = {}
- self.data_dir = CONST.dir_functest_data
+ self.result_dir = CONST.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'] = {}
'vnf_{}_tenant_name'.format(self.case_name))
self.tenant_description = CONST.__getattribute__(
'vnf_{}_tenant_description'.format(self.case_name))
- except:
- raise Exception("Unknown VNF case=" + 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:
+ except Exception:
self.logger.warn("No tenant image defined for this VNF")
def execute(self):
self.start_time = time.time()
# Prepare the test (Create Tenant, User, ...)
- self.logger.info("Create VNF Onboarding environment")
- self.prepare()
+ try:
+ self.logger.info("Create VNF Onboarding environment")
+ self.prepare()
+ except Exception:
+ self.logger.error("Error during VNF Onboarding environment" +
+ "creation", exc_info=True)
+ return base.TestCase.EX_TESTCASE_FAILED
# Deploy orchestrator
try:
res_orchestrator['result'])
self.details['orchestrator']['duration'] = round(
orchestrator_ready_time - self.start_time, 1)
- except:
- self.logger.warn("Problem with the Orchestrator")
+ except Exception:
+ self.logger.warn("Problem with the Orchestrator", exc_info=True)
# Deploy VNF
try:
self.details['vnf']['result'] = res_deploy_vnf['result']
self.details['vnf']['duration'] = round(
vnf_ready_time - orchestrator_ready_time, 1)
- except:
- raise Exception("Error during VNF deployment")
+ except Exception:
+ self.logger.error("Error during VNF deployment", exc_info=True)
+ return base.TestCase.EX_TESTCASE_FAILED
# Test VNF
try:
self.details['test_vnf']['result'] = res_test_vnf['result']
self.details['test_vnf']['duration'] = round(
test_vnf_done_time - vnf_ready_time, 1)
- except:
- raise Exception("Error when running VNF tests")
+ except Exception:
+ self.logger.error("Error when running VNF tests", exc_info=True)
+ return base.TestCase.EX_TESTCASE_FAILED
# Clean the system
self.clean()
# prepare state could consist in the creation of the resources
# a dedicated user
- # a dedictaed tenant
+ # 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)")
- user_id = os_utils.get_user_id(self.keystone_client,
- self.creds['username'])
- if user_id == '':
- self.step_failure("Failed to get id of " +
- self.creds['username'])
-
- tenant_id = os_utils.create_tenant(
- self.keystone_client, self.tenant_name, self.tenant_description)
+ 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:
- self.step_failure("Failed to create " +
- self.tenant_name + " tenant")
-
- roles_name = ["admin", "Admin"]
- role_id = ''
- for role_name in roles_name:
- if role_id == '':
- role_id = os_utils.get_role_id(self.keystone_client, role_name)
-
- if role_id == '':
- self.logger.error("Failed to get id for %s role" % role_name)
- self.step_failure("Failed to get role id of " + role_name)
-
- if not os_utils.add_role_user(self.keystone_client, user_id,
- role_id, tenant_id):
- self.logger.error("Failed to add %s on tenant" %
- self.creds['username'])
- self.step_failure("Failed to add %s on tenant" %
- self.creds['username'])
-
- user_id = os_utils.create_user(self.keystone_client,
- self.tenant_name,
- self.tenant_name,
- None,
- 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.logger.error("Failed to create %s user" % self.tenant_name)
- self.step_failure("Failed to create user ")
+ 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.creds.update({
- "tenant": self.tenant_name,
+ self.admin_creds = self.creds.copy()
+ self.admin_creds.update({
+ "tenant": self.tenant_name
})
- self.neutron_client = os_utils.get_neutron_client(self.creds)
- self.nova_client = os_utils.get_nova_client(self.creds)
+ 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,
})
- self.glance_client = os_utils.get_glance_client(self.creds)
- self.logger.info("Upload some OS images if it doesn't exist")
-
- temp_dir = os.path.join(self.data_dir, "tmp/")
- for image_name, image_url in self.images.iteritems():
- image_id = os_utils.get_image_id(self.glance_client, image_name)
-
- if image_id == '':
- self.logger.info("""%s image doesn't exist on glance repository. Try
- downloading this image and upload on glance !""" % image_name)
- image_id = os_utils.download_and_add_image_on_glance(
- self.glance_client, image_name, image_url, temp_dir)
-
- if image_id == '':
- self.step_failure(
- "Failed to find or upload required OS "
- "image for this deployment")
- self.logger.info("Update security group quota for this tenant")
-
- if not os_utils.update_sg_quota(self.neutron_client,
- tenant_id, 50, 100):
- self.step_failure("Failed to update security group quota" +
- " for tenant " + self.tenant_name)
-
- # orchestrator is not mandatory to dpeloy and test VNF
+ # orchestrator is not mandatory to deploy and test VNF
def deploy_orchestrator(self, **kwargs):
pass
self.logger.error("VNF must be tested")
raise Exception("VNF not tested")
+ # clean before openstack clean run
def clean(self):
self.logger.info("test cleaning")
- self.logger.info("Removing %s tenant .." % self.tenant_name)
- tenant_id = os_utils.get_tenant_id(self.keystone_client,
- self.tenant_name)
- if tenant_id == '':
- self.logger.error("Error : Failed to get id of %s tenant" %
- self.tenant_name)
- else:
- if not os_utils.delete_tenant(self.keystone_client, tenant_id):
- self.logger.error("Error : Failed to remove %s tenant" %
- self.tenant_name)
-
- self.logger.info("Removing %s user .." % self.tenant_name)
- user_id = os_utils.get_user_id(
- self.keystone_client, self.tenant_name)
- if user_id == '':
- self.logger.error("Error : Failed to get id of %s user" %
- self.tenant_name)
- else:
- if not os_utils.delete_user(self.keystone_client, user_id):
- self.logger.error("Error : Failed to remove %s user" %
- self.tenant_name)
-
def parse_results(self):
exit_code = self.EX_OK
- self.criteria = "PASS"
+ 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.criteria = "FAIL"
+ self.result = "FAIL"
return exit_code
def log_results(self):
ft_utils.logger_test_results(self.project_name,
self.case_name,
- self.criteria,
+ self.result,
self.details)
def step_failure(self, error_msg):
part = inspect.stack()[1][3]
- self.details[part]['status'] = 'FAIL'
- self.details[part]['result'] = error_msg
+ 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)
Code Review / functest.git / blobdiff