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(): # 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, encoding='utf-8') 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 = f"http://{cfy_manager_ip}/deployments/{deployment_name}/outputs"
102 response = requests.get(
104 auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
105 headers={'Tenant': 'default_tenant'})
107 resp_data = response.json()
108 self.logger.debug(resp_data)
109 data = resp_data["outputs"]
112 def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
113 outputs = self.get_blueprint_outputs(cfy_manager_ip,
115 vnfs = outputs["vnfs"]
117 for vnf_name in vnfs:
118 vnf_list.append(vnfs[vnf_name])
121 def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
122 outputs = self.get_blueprint_outputs(cfy_manager_ip,
124 networks = outputs["networks"]
126 for network_name in networks:
127 network_list.append(networks[network_name])
130 def request_vm_delete(self, vnf_info_list):
131 for vnf in vnf_info_list:
132 self.logger.debug("delete the %s", vnf["vnf_name"])
133 self.cloud.delete_server(vnf["vnf_name"])
135 def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
137 network_list = self.get_blueprint_outputs_networks(
139 topology_deploy_name)
140 vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
141 topology_deploy_name)
142 for vnf in vnf_info_list:
143 vnf_name = vnf["vnf_name"]
144 vnf["os_type"] = self.image["os_type"]
145 vnf["user"] = self.image["user"]
146 vnf["pass"] = self.image["pass"]
148 vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
150 self.logger.debug("vnf name : %s", vnf_name)
151 self.logger.debug(vnf_name + " floating ip address : " +
154 for network in network_list:
155 network_name = network["network_name"]
156 ip_address = self.get_address(vnf["vnf_name"],
157 network["network_name"])
158 vnf[network_name + "_ip"] = ip_address
159 mac = self.get_mac_address(vnf["vnf_name"],
160 network["network_name"])
161 vnf[network_name + "_mac"] = mac
163 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
164 " : " + vnf[network_name + "_ip"])
165 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
166 " : " + vnf[network_name + "_mac"])
171 def get_target_vnf(vnf_info_list):
172 for vnf in vnf_info_list:
173 if vnf["target_vnf_flag"]:
179 def get_reference_vnf_list(vnf_info_list):
180 reference_vnf_list = []
181 for vnf in vnf_info_list:
182 if not vnf["target_vnf_flag"]:
183 reference_vnf_list.append(vnf)
185 return reference_vnf_list
188 def get_vnf_info(vnf_info_list, vnf_name):
189 for vnf in vnf_info_list:
190 if vnf["vnf_name"] == vnf_name:
196 def convert_functional_test_result(result_data_list):
198 for result_data in result_data_list:
199 test_kind = result_data["test_kind"]
200 protocol = result_data["protocol"]
201 test_result_data = result_data["result"]
203 if test_kind not in result:
204 result[test_kind] = []
206 result[test_kind].append({protocol: test_result_data})
208 return {"Functional_test": result}
210 def write_result_data(self, result_data):
212 if not os.path.isfile(self.test_result_json_file):
214 self.test_result_json_file, "w",
215 encoding="utf-8") as file_fd:
219 self.test_result_json_file, "r",
220 encoding="utf-8") as file_fd:
221 test_result = json.load(file_fd)
223 test_result.append(result_data)
226 self.test_result_json_file, "w",
227 encoding="utf-8") as file_fd:
228 json.dump(test_result, file_fd)
230 def output_test_result_json(self):
231 if os.path.isfile(self.test_result_json_file):
233 self.test_result_json_file, "r",
234 encoding="utf-8") as file_fd:
235 test_result = json.load(file_fd)
236 output_json_data = json.dumps(test_result,
239 self.logger.debug("test_result %s", output_json_data)
241 self.logger.debug("Not found %s", self.test_result_json_file)
244 def get_test_scenario(file_path):
245 with open(file_path, "r", encoding="utf-8") as test_scenario_file:
246 test_scenario_yaml = yaml.safe_load(test_scenario_file)
247 return test_scenario_yaml["test_scenario_list"]