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 # pylint: disable=missing-docstring
12 """ Utility module of vrouter testcase """
20 from functest.utils import config
22 RESULT_SPRIT_INDEX = {
32 NOVA_CLIENT_API_VERSION = '2'
33 NOVA_CILENT_NETWORK_INFO_INDEX = 0
34 CFY_INFO_OUTPUT_FILE = "output.txt"
36 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
37 PACKET_LOST_INFO_INDEX = 0
38 PACKET_TOTAL_INFO_INDEX = 1
40 NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
41 NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
42 NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
43 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
46 class Utilvnf(object): # pylint: disable=too-many-instance-attributes
47 """ Utility class of vrouter testcase """
49 logger = logging.getLogger(__name__)
52 self.vnf_data_dir = getattr(config.CONF, 'dir_router_data')
53 self.command_template_dir = "command_template/"
54 self.test_scenario_yaml = "test_scenario.yaml"
55 test_env_config_yaml_file = "test_env_config.yaml"
56 self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
57 self.test_env_config_yaml = os.path.join(
59 test_env_config_yaml_file)
61 self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
62 self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
64 if not os.path.exists(self.vnf_data_dir):
65 os.makedirs(self.vnf_data_dir)
67 with open(self.test_env_config_yaml) as file_fd:
68 test_env_config_yaml = yaml.safe_load(file_fd)
71 self.image = test_env_config_yaml.get(
72 "general").get("images").get("vyos")
73 self.tester_image = test_env_config_yaml.get(
74 "general").get("images").get("tester_vm_os")
76 self.test_result_json_file = "test_result.json"
77 if os.path.isfile(self.test_result_json_file):
78 os.remove(self.test_result_json_file)
79 self.logger.debug("removed %s", self.test_result_json_file)
83 def set_credentials(self, cloud):
86 def get_address(self, server_name, network_name):
87 server = self.cloud.get_server(server_name)
88 address = server.addresses[
89 network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
93 def get_mac_address(self, server_name, network_name):
94 server = self.cloud.get_server(server_name)
95 mac_address = server.addresses[network_name][
96 NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
100 def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
101 url = "http://%s/deployments/%s/outputs" % (
102 cfy_manager_ip, deployment_name)
104 response = requests.get(
106 auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
107 headers={'Tenant': 'default_tenant'})
109 resp_data = response.json()
110 self.logger.debug(resp_data)
111 data = resp_data["outputs"]
114 def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
115 outputs = self.get_blueprint_outputs(cfy_manager_ip,
117 vnfs = outputs["vnfs"]
119 for vnf_name in vnfs:
120 vnf_list.append(vnfs[vnf_name])
123 def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
124 outputs = self.get_blueprint_outputs(cfy_manager_ip,
126 networks = outputs["networks"]
128 for network_name in networks:
129 network_list.append(networks[network_name])
132 def request_vm_delete(self, vnf_info_list):
133 for vnf in vnf_info_list:
134 self.logger.debug("delete the %s", vnf["vnf_name"])
135 self.cloud.delete_server(vnf["vnf_name"])
137 def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
139 network_list = self.get_blueprint_outputs_networks(
141 topology_deploy_name)
142 vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
143 topology_deploy_name)
144 for vnf in vnf_info_list:
145 vnf_name = vnf["vnf_name"]
146 vnf["os_type"] = self.image["os_type"]
147 vnf["user"] = self.image["user"]
148 vnf["pass"] = self.image["pass"]
150 vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
152 self.logger.debug("vnf name : %s", vnf_name)
153 self.logger.debug(vnf_name + " floating ip address : " +
156 for network in network_list:
157 network_name = network["network_name"]
158 ip_address = self.get_address(vnf["vnf_name"],
159 network["network_name"])
160 vnf[network_name + "_ip"] = ip_address
161 mac = self.get_mac_address(vnf["vnf_name"],
162 network["network_name"])
163 vnf[network_name + "_mac"] = mac
165 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
166 " : " + vnf[network_name + "_ip"])
167 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
168 " : " + vnf[network_name + "_mac"])
173 def get_target_vnf(vnf_info_list):
174 for vnf in vnf_info_list:
175 if vnf["target_vnf_flag"]:
181 def get_reference_vnf_list(vnf_info_list):
182 reference_vnf_list = []
183 for vnf in vnf_info_list:
184 if not vnf["target_vnf_flag"]:
185 reference_vnf_list.append(vnf)
187 return reference_vnf_list
190 def get_vnf_info(vnf_info_list, vnf_name):
191 for vnf in vnf_info_list:
192 if vnf["vnf_name"] == vnf_name:
198 def convert_functional_test_result(result_data_list):
200 for result_data in result_data_list:
201 test_kind = result_data["test_kind"]
202 protocol = result_data["protocol"]
203 test_result_data = result_data["result"]
205 if test_kind not in result:
206 result[test_kind] = []
208 result[test_kind].append({protocol: test_result_data})
210 return {"Functional_test": result}
212 def write_result_data(self, result_data):
214 if not os.path.isfile(self.test_result_json_file):
215 file_fd = open(self.test_result_json_file, "w")
218 file_fd = open(self.test_result_json_file, "r")
219 test_result = json.load(file_fd)
222 test_result.append(result_data)
224 file_fd = open(self.test_result_json_file, "w")
225 json.dump(test_result, file_fd)
228 def output_test_result_json(self):
229 if os.path.isfile(self.test_result_json_file):
230 file_fd = open(self.test_result_json_file, "r")
231 test_result = json.load(file_fd)
233 output_json_data = json.dumps(test_result,
236 self.logger.debug("test_result %s", output_json_data)
238 self.logger.debug("Not found %s", self.test_result_json_file)
241 def get_test_scenario(file_path):
242 test_scenario_file = open(file_path,
244 test_scenario_yaml = yaml.safe_load(test_scenario_file)
245 test_scenario_file.close()
246 return test_scenario_yaml["test_scenario_list"]