3 # Copyright (c) 2016 Orange and others.
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
14 import functest.core.testcase as base
15 from functest.utils.constants import CONST
16 import functest.utils.functest_utils as ft_utils
17 import functest.utils.openstack_utils as os_utils
20 class VnfOnBoardingBase(base.TestCase):
22 __logger = logging.getLogger(__name__)
24 def __init__(self, **kwargs):
25 super(VnfOnBoardingBase, self).__init__(**kwargs)
26 self.repo = kwargs.get('repo', '')
27 self.cmd = kwargs.get('cmd', '')
29 self.result_dir = CONST.__getattribute__('dir_results')
30 self.details_step_mapping = dict(
31 deploy_orchestrator='orchestrator',
34 prepare='prepare_env')
35 self.details['prepare_env'] = {}
36 self.details['orchestrator'] = {}
37 self.details['vnf'] = {}
38 self.details['test_vnf'] = {}
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))
46 # raise Exception("Unknown VNF case=" + self.case_name)
47 self.__logger.error("Unknown VNF case={}".format(self.case_name))
50 self.images = CONST.__getattribute__(
51 'vnf_{}_tenant_images'.format(self.case_name))
53 self.__logger.warn("No tenant image defined for this VNF")
56 self.start_time = time.time()
57 # Prepare the test (Create Tenant, User, ...)
59 self.__logger.info("Create VNF Onboarding environment")
62 self.__logger.error("Error during VNF Onboarding environment"
63 "creation", exc_info=True)
64 return base.TestCase.EX_TESTCASE_FAILED
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)
80 self.__logger.warn("Problem with the Orchestrator", exc_info=True)
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)
92 self.__logger.error("Error during VNF deployment", exc_info=True)
93 return base.TestCase.EX_TESTCASE_FAILED
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)
105 self.__logger.error("Error when running VNF tests", exc_info=True)
106 return base.TestCase.EX_TESTCASE_FAILED
110 self.stop_time = time.time()
112 exit_code = self.parse_results()
116 # prepare state could consist in the creation of the resources
121 self.creds = os_utils.get_credentials()
122 self.keystone_client = os_utils.get_keystone_client()
125 "Prepare OpenStack plateform(create tenant and user)")
126 admin_user_id = os_utils.get_user_id(self.keystone_client,
127 self.creds['username'])
128 if not admin_user_id:
129 self.step_failure("Failed to get id of {0}".format(
130 self.creds['username']))
132 tenant_id = os_utils.get_tenant_id(self.keystone_client,
135 tenant_id = os_utils.create_tenant(self.keystone_client,
137 self.tenant_description)
139 self.step_failure("Failed to get or create {0} tenant".format(
141 roles_name = ["admin", "Admin"]
143 for role_name in roles_name:
145 role_id = os_utils.get_role_id(self.keystone_client,
149 self.step_failure("Failed to get id for {0} role".format(
152 if not os_utils.add_role_user(self.keystone_client, admin_user_id,
154 self.step_failure("Failed to add {0} on tenant".format(
155 self.creds['username']))
157 user_id = os_utils.get_or_create_user(self.keystone_client,
162 self.step_failure("Failed to get or create {0} user".format(
165 os_utils.add_role_user(self.keystone_client, user_id,
168 self.__logger.info("Update OpenStack creds informations")
169 self.admin_creds = self.creds.copy()
170 self.admin_creds.update({
171 "tenant": self.tenant_name
173 self.neutron_client = os_utils.get_neutron_client(self.admin_creds)
174 self.nova_client = os_utils.get_nova_client(self.admin_creds)
176 "tenant": self.tenant_name,
177 "username": self.tenant_name,
178 "password": self.tenant_name,
181 # orchestrator is not mandatory to deploy and test VNF
182 def deploy_orchestrator(self, **kwargs):
185 # TODO see how to use built-in exception from releng module
186 def deploy_vnf(self):
187 self.__logger.error("VNF must be deployed")
188 raise Exception("VNF not deployed")
191 self.__logger.error("VNF must be tested")
192 raise Exception("VNF not tested")
194 # clean before openstack clean run
196 self.__logger.info("test cleaning")
198 def parse_results(self):
199 exit_code = self.EX_OK
201 self.__logger.info(self.details)
202 # The 2 VNF steps must be OK to get a PASS result
203 if (self.details['vnf']['status'] is not "PASS" or
204 self.details['test_vnf']['status'] is not "PASS"):
205 exit_code = self.EX_RUN_ERROR
209 def log_results(self):
210 ft_utils.logger_test_results(self.project_name,
215 def step_failure(self, error_msg):
216 part = inspect.stack()[1][3]
217 self.__logger.error("Step {0} failed: {1}".format(part, error_msg))
219 step_name = self.details_step_mapping[part]
220 part_info = self.details[step_name]
222 self.details[part] = {}
223 part_info = self.details[part]
224 part_info['status'] = 'FAIL'
225 part_info['result'] = error_msg
226 raise Exception(error_msg)