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 requests.auth import HTTPBasicAuth
22 from snaps.openstack.utils import nova_utils
25 RESULT_SPRIT_INDEX = {
35 NOVA_CLIENT_API_VERSION = '2'
36 NOVA_CILENT_NETWORK_INFO_INDEX = 0
37 CFY_INFO_OUTPUT_FILE = "output.txt"
39 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
40 PACKET_LOST_INFO_INDEX = 0
41 PACKET_TOTAL_INFO_INDEX = 1
43 NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
44 NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
45 NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
46 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
49 class Utilvnf(object):
50 """ Utility class of vrouter testcase """
52 logger = logging.getLogger(__name__)
57 data_dir = data_dir = CONST.__getattribute__('dir_router_data')
59 self.vnf_data_dir = data_dir
60 self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
61 self.command_template_dir = "command_template/"
62 self.test_scenario_yaml = "test_scenario.yaml"
63 test_env_config_yaml_file = "test_env_config.yaml"
64 self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
65 self.test_env_config_yaml = os.path.join(
67 self.opnfv_vnf_data_dir,
68 test_env_config_yaml_file)
70 self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
71 self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
73 if not os.path.exists(self.vnf_data_dir):
74 os.makedirs(self.vnf_data_dir)
76 case_dir = pkg_resources.resource_filename(
77 'functest', 'opnfv_tests/vnf/router')
79 config_file_name = CONST.__getattribute__(
80 'vnf_{}_config'.format("vyos_vrouter"))
82 config_file = os.path.join(case_dir, config_file_name)
84 with open(config_file) as file_fd:
85 vrouter_config_yaml = yaml.safe_load(file_fd)
88 test_data = vrouter_config_yaml.get("test_data")
90 self.logger.debug("Downloading the test data.")
91 vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
93 if not os.path.exists(vrouter_data_path):
94 Repo.clone_from(test_data['url'],
96 branch=test_data['branch'])
98 with open(self.test_env_config_yaml) as file_fd:
99 test_env_config_yaml = yaml.safe_load(file_fd)
102 self.image = test_env_config_yaml.get(
103 "general").get("images").get("vyos")
104 self.tester_image = test_env_config_yaml.get(
105 "general").get("images").get("tester_vm_os")
107 self.test_result_json_file = "test_result.json"
108 if os.path.isfile(self.test_result_json_file):
109 os.remove(self.test_result_json_file)
110 self.logger.debug("removed %s" % self.test_result_json_file)
112 def get_nova_client(self):
113 nova_client = nova_utils.nova_client(self.snaps_creds)
117 def set_credentials(self, snaps_creds):
118 self.snaps_creds = snaps_creds
120 def get_address(self, server_name, network_name):
121 nova_client = self.get_nova_client()
122 servers_list = nova_client.servers.list()
125 for server in servers_list:
126 if server.name == server_name:
129 address = server.addresses[
130 network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
134 def get_mac_address(self, server_name, network_name):
135 nova_client = self.get_nova_client()
136 servers_list = nova_client.servers.list()
139 for server in servers_list:
140 if server.name == server_name:
143 mac_address = server.addresses[network_name][
144 NOVA_CILENT_NETWORK_INFO_INDEX][
145 "OS-EXT-IPS-MAC:mac_addr"]
149 def reboot_vm(self, server_name):
150 nova_client = self.get_nova_client()
151 servers_list = nova_client.servers.list()
154 for server in servers_list:
155 if server.name == server_name:
162 def delete_vm(self, server_name):
163 nova_client = self.get_nova_client()
164 servers_list = nova_client.servers.list()
167 for server in servers_list:
168 if server.name == server_name:
169 nova_client.servers.delete(server)
174 def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
175 url = "http://%s/deployments/%s/outputs" % (
176 cfy_manager_ip, deployment_name)
178 response = requests.get(
180 auth=HTTPBasicAuth('admin', 'admin'),
181 headers={'Tenant': 'default_tenant'})
183 resp_data = response.json()
184 self.logger.debug(resp_data)
185 data = resp_data["outputs"]
188 def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
189 outputs = self.get_blueprint_outputs(cfy_manager_ip,
191 vnfs = outputs["vnfs"]
193 for vnf_name in vnfs:
194 vnf_list.append(vnfs[vnf_name])
197 def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
198 outputs = self.get_blueprint_outputs(cfy_manager_ip,
200 networks = outputs["networks"]
202 for network_name in networks:
203 network_list.append(networks[network_name])
206 def request_vnf_reboot(self, vnf_info_list):
207 for vnf in vnf_info_list:
208 self.logger.debug("reboot the " + vnf["vnf_name"])
209 self.reboot_vm(vnf["vnf_name"])
211 def request_vm_delete(self, vnf_info_list):
212 for vnf in vnf_info_list:
213 self.logger.debug("delete the " + vnf["vnf_name"])
214 self.delete_vm(vnf["vnf_name"])
216 def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
218 network_list = self.get_blueprint_outputs_networks(
220 topology_deploy_name)
221 vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
222 topology_deploy_name)
223 for vnf in vnf_info_list:
224 vnf_name = vnf["vnf_name"]
225 vnf["os_type"] = self.image["os_type"]
226 vnf["user"] = self.image["user"]
227 vnf["pass"] = self.image["pass"]
229 if vnf_name == target_vnf_name:
230 vnf["target_vnf_flag"] = True
232 vnf["target_vnf_flag"] = False
234 self.logger.debug("vnf name : " + vnf_name)
235 self.logger.debug(vnf_name + " floating ip address : " +
238 for network in network_list:
239 network_name = network["network_name"]
240 ip_address = self.get_address(vnf["vnf_name"],
241 network["network_name"])
242 vnf[network_name + "_ip"] = ip_address
243 mac = self.get_mac_address(vnf["vnf_name"],
244 network["network_name"])
245 vnf[network_name + "_mac"] = mac
247 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
248 " : " + vnf[network_name + "_ip"])
249 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
250 " : " + vnf[network_name + "_mac"])
254 def get_target_vnf(self, vnf_info_list):
255 for vnf in vnf_info_list:
256 if vnf["target_vnf_flag"]:
261 def get_reference_vnf_list(self, vnf_info_list):
262 reference_vnf_list = []
263 for vnf in vnf_info_list:
264 if not vnf["target_vnf_flag"]:
265 reference_vnf_list.append(vnf)
267 return reference_vnf_list
269 def get_vnf_info(self, vnf_info_list, vnf_name):
270 for vnf in vnf_info_list:
271 if vnf["vnf_name"] == vnf_name:
276 def convert_functional_test_result(self, result_data_list):
278 for result_data in result_data_list:
279 test_kind = result_data["test_kind"]
280 protocol = result_data["protocol"]
281 test_result_data = result_data["result"]
283 if test_kind not in result:
284 result[test_kind] = []
286 result[test_kind].append({protocol: test_result_data})
288 return {"Functional_test": result}
290 def write_result_data(self, result_data):
292 if not os.path.isfile(self.test_result_json_file):
293 file_fd = open(self.test_result_json_file, "w")
296 file_fd = open(self.test_result_json_file, "r")
297 test_result = json.load(file_fd)
300 test_result.append(result_data)
302 file_fd = open(self.test_result_json_file, "w")
303 json.dump(test_result, file_fd)
306 def output_test_result_json(self):
307 if os.path.isfile(self.test_result_json_file):
308 file_fd = open(self.test_result_json_file, "r")
309 test_result = json.load(file_fd)
311 output_json_data = json.dumps(test_result,
314 self.logger.debug("test_result %s" % output_json_data)
316 self.logger.debug("Not found %s" % self.test_result_json_file)
318 def get_test_scenario(self, file_path):
319 test_scenario_file = open(file_path,
321 test_scenario_yaml = yaml.safe_load(test_scenario_file)
322 test_scenario_file.close()
323 return test_scenario_yaml["test_scenario_list"]