3 # Copyright (c) 2017 Okinawa Open Laboratory 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
10 """ Utility module of vrouter testcase """
19 from functest.utils.constants import CONST
21 from novaclient import client as novaclient
22 from keystoneauth1.identity import v3
23 from keystoneauth1 import session
24 from requests.auth import HTTPBasicAuth
26 RESULT_SPRIT_INDEX = {
36 NOVA_CLIENT_API_VERSION = '2'
37 NOVA_CILENT_NETWORK_INFO_INDEX = 0
38 CFY_INFO_OUTPUT_FILE = "output.txt"
40 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
41 PACKET_LOST_INFO_INDEX = 0
42 PACKET_TOTAL_INFO_INDEX = 1
44 NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
45 NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
46 NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
47 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
50 class Utilvnf(object):
51 """ Utility class of vrouter testcase """
53 logger = logging.getLogger(__name__)
61 data_dir = data_dir = CONST.__getattribute__('dir_router_data')
63 self.vnf_data_dir = data_dir
64 self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
65 self.command_template_dir = "command_template/"
66 self.test_scenario_yaml = "test_scenario.yaml"
67 test_env_config_yaml_file = "test_env_config.yaml"
68 self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
69 self.test_env_config_yaml = os.path.join(
71 self.opnfv_vnf_data_dir,
72 test_env_config_yaml_file)
74 self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
75 self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
77 if not os.path.exists(self.vnf_data_dir):
78 os.makedirs(self.vnf_data_dir)
80 case_dir = pkg_resources.resource_filename(
81 'functest', 'opnfv_tests/vnf/router')
83 config_file_name = CONST.__getattribute__(
84 'vnf_{}_config'.format("vyos_vrouter"))
86 config_file = os.path.join(case_dir, config_file_name)
88 with open(config_file) as file_fd:
89 vrouter_config_yaml = yaml.safe_load(file_fd)
92 test_data = vrouter_config_yaml.get("test_data")
94 self.logger.debug("Downloading the test data.")
95 vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
97 if not os.path.exists(vrouter_data_path):
98 Repo.clone_from(test_data['url'],
100 branch=test_data['branch'])
102 with open(self.test_env_config_yaml) as file_fd:
103 test_env_config_yaml = yaml.safe_load(file_fd)
106 self.image = test_env_config_yaml.get(
107 "general").get("images").get("vyos")
108 self.tester_image = test_env_config_yaml.get(
109 "general").get("images").get("tester_vm_os")
111 self.test_result_json_file = "test_result.json"
112 if os.path.isfile(self.test_result_json_file):
113 os.remove(self.test_result_json_file)
114 self.logger.debug("removed %s" % self.test_result_json_file)
116 def get_nova_client(self):
117 creds = self.get_nova_credentials()
118 auth = v3.Password(auth_url=creds['auth_url'],
119 username=creds['username'],
120 password=creds['password'],
121 project_name=creds['tenant_name'],
122 user_domain_id='default',
123 project_domain_id='default')
124 sess = session.Session(auth=auth)
125 nova_client = novaclient.Client(NOVA_CLIENT_API_VERSION, session=sess)
129 def set_credentials(self, username, password, auth_url, tenant_name):
130 self.username = username
131 self.password = password
132 self.auth_url = auth_url
133 self.tenant_name = tenant_name
135 def get_nova_credentials(self):
137 creds['username'] = self.username
138 creds['password'] = self.password
139 creds['auth_url'] = self.auth_url
140 creds['tenant_name'] = self.tenant_name
143 def get_address(self, server_name, network_name):
144 nova_client = self.get_nova_client()
145 servers_list = nova_client.servers.list()
148 for server in servers_list:
149 if server.name == server_name:
152 address = server.addresses[
153 network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
157 def get_mac_address(self, server_name, network_name):
158 nova_client = self.get_nova_client()
159 servers_list = nova_client.servers.list()
162 for server in servers_list:
163 if server.name == server_name:
166 mac_address = server.addresses[network_name][
167 NOVA_CILENT_NETWORK_INFO_INDEX][
168 "OS-EXT-IPS-MAC:mac_addr"]
172 def reboot_vm(self, server_name):
173 nova_client = self.get_nova_client()
174 servers_list = nova_client.servers.list()
177 for server in servers_list:
178 if server.name == server_name:
185 def delete_vm(self, server_name):
186 nova_client = self.get_nova_client()
187 servers_list = nova_client.servers.list()
190 for server in servers_list:
191 if server.name == server_name:
192 nova_client.servers.delete(server)
197 def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
198 url = "http://%s/deployments/%s/outputs" % (
199 cfy_manager_ip, deployment_name)
201 response = requests.get(
203 auth=HTTPBasicAuth('admin', 'admin'),
204 headers={'Tenant': 'default_tenant'})
206 resp_data = response.json()
207 self.logger.debug(resp_data)
208 data = resp_data["outputs"]
211 def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
212 outputs = self.get_blueprint_outputs(cfy_manager_ip,
214 vnfs = outputs["vnfs"]
216 for vnf_name in vnfs:
217 vnf_list.append(vnfs[vnf_name])
220 def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
221 outputs = self.get_blueprint_outputs(cfy_manager_ip,
223 networks = outputs["networks"]
225 for network_name in networks:
226 network_list.append(networks[network_name])
229 def request_vnf_reboot(self, vnf_info_list):
230 for vnf in vnf_info_list:
231 self.logger.debug("reboot the " + vnf["vnf_name"])
232 self.reboot_vm(vnf["vnf_name"])
234 def request_vm_delete(self, vnf_info_list):
235 for vnf in vnf_info_list:
236 self.logger.debug("delete the " + vnf["vnf_name"])
237 self.delete_vm(vnf["vnf_name"])
239 def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
241 network_list = self.get_blueprint_outputs_networks(
243 topology_deploy_name)
244 vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
245 topology_deploy_name)
246 for vnf in vnf_info_list:
247 vnf_name = vnf["vnf_name"]
248 vnf["os_type"] = self.image["os_type"]
249 vnf["user"] = self.image["user"]
250 vnf["pass"] = self.image["pass"]
252 if vnf_name == target_vnf_name:
253 vnf["target_vnf_flag"] = True
255 vnf["target_vnf_flag"] = False
257 self.logger.debug("vnf name : " + vnf_name)
258 self.logger.debug(vnf_name + " floating ip address : " +
261 for network in network_list:
262 network_name = network["network_name"]
263 ip_address = self.get_address(vnf["vnf_name"],
264 network["network_name"])
265 vnf[network_name + "_ip"] = ip_address
266 mac = self.get_mac_address(vnf["vnf_name"],
267 network["network_name"])
268 vnf[network_name + "_mac"] = mac
270 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
271 " : " + vnf[network_name + "_ip"])
272 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
273 " : " + vnf[network_name + "_mac"])
277 def get_target_vnf(self, vnf_info_list):
278 for vnf in vnf_info_list:
279 if vnf["target_vnf_flag"]:
284 def get_reference_vnf_list(self, vnf_info_list):
285 reference_vnf_list = []
286 for vnf in vnf_info_list:
287 if not vnf["target_vnf_flag"]:
288 reference_vnf_list.append(vnf)
290 return reference_vnf_list
292 def get_vnf_info(self, vnf_info_list, vnf_name):
293 for vnf in vnf_info_list:
294 if vnf["vnf_name"] == vnf_name:
299 def convert_functional_test_result(self, result_data_list):
301 for result_data in result_data_list:
302 test_kind = result_data["test_kind"]
303 protocol = result_data["protocol"]
304 test_result_data = result_data["result"]
306 if test_kind not in result:
307 result[test_kind] = []
309 result[test_kind].append({protocol: test_result_data})
311 return {"Functional_test": result}
313 def write_result_data(self, result_data):
315 if not os.path.isfile(self.test_result_json_file):
316 file_fd = open(self.test_result_json_file, "w")
319 file_fd = open(self.test_result_json_file, "r")
320 test_result = json.load(file_fd)
323 test_result.append(result_data)
325 file_fd = open(self.test_result_json_file, "w")
326 json.dump(test_result, file_fd)
329 def output_test_result_json(self):
330 if os.path.isfile(self.test_result_json_file):
331 file_fd = open(self.test_result_json_file, "r")
332 test_result = json.load(file_fd)
334 output_json_data = json.dumps(test_result,
337 self.logger.debug("test_result %s" % output_json_data)
339 self.logger.debug("Not found %s" % self.test_result_json_file)
341 def get_test_scenario(self, file_path):
342 test_scenario_file = open(file_path,
344 test_scenario_yaml = yaml.safe_load(test_scenario_file)
345 test_scenario_file.close()
346 return test_scenario_yaml["test_scenario_list"]