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 """
23 from functest.utils import config
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): # pylint: disable=too-many-instance-attributes
50 """ Utility class of vrouter testcase """
52 logger = logging.getLogger(__name__)
55 self.vnf_data_dir = getattr(config.CONF, 'dir_router_data')
56 self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
57 self.command_template_dir = "command_template/"
58 self.test_scenario_yaml = "test_scenario.yaml"
59 test_env_config_yaml_file = "test_env_config.yaml"
60 self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
61 self.test_env_config_yaml = os.path.join(
63 self.opnfv_vnf_data_dir,
64 test_env_config_yaml_file)
66 self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
67 self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
69 if not os.path.exists(self.vnf_data_dir):
70 os.makedirs(self.vnf_data_dir)
72 case_dir = pkg_resources.resource_filename(
73 'functest', 'opnfv_tests/vnf/router')
75 config_file_name = getattr(
76 config.CONF, 'vnf_{}_config'.format("vyos_vrouter"))
78 config_file = os.path.join(case_dir, config_file_name)
80 with open(config_file) as file_fd:
81 vrouter_config_yaml = yaml.safe_load(file_fd)
84 test_data = vrouter_config_yaml.get("test_data")
86 self.logger.debug("Downloading the test data.")
87 vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
89 if not os.path.exists(vrouter_data_path):
90 Repo.clone_from(test_data['url'],
92 branch=test_data['branch'])
94 with open(self.test_env_config_yaml) as file_fd:
95 test_env_config_yaml = yaml.safe_load(file_fd)
98 self.image = test_env_config_yaml.get(
99 "general").get("images").get("vyos")
100 self.tester_image = test_env_config_yaml.get(
101 "general").get("images").get("tester_vm_os")
103 self.test_result_json_file = "test_result.json"
104 if os.path.isfile(self.test_result_json_file):
105 os.remove(self.test_result_json_file)
106 self.logger.debug("removed %s", self.test_result_json_file)
110 def set_credentials(self, cloud):
113 def get_address(self, server_name, network_name):
114 server = self.cloud.get_server(server_name)
115 address = server.addresses[
116 network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
120 def get_mac_address(self, server_name, network_name):
121 server = self.cloud.get_server(server_name)
122 mac_address = server.addresses[network_name][
123 NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
127 def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
128 url = "http://%s/deployments/%s/outputs" % (
129 cfy_manager_ip, deployment_name)
131 response = requests.get(
133 auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
134 headers={'Tenant': 'default_tenant'})
136 resp_data = response.json()
137 self.logger.debug(resp_data)
138 data = resp_data["outputs"]
141 def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
142 outputs = self.get_blueprint_outputs(cfy_manager_ip,
144 vnfs = outputs["vnfs"]
146 for vnf_name in vnfs:
147 vnf_list.append(vnfs[vnf_name])
150 def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
151 outputs = self.get_blueprint_outputs(cfy_manager_ip,
153 networks = outputs["networks"]
155 for network_name in networks:
156 network_list.append(networks[network_name])
159 def request_vm_delete(self, vnf_info_list):
160 for vnf in vnf_info_list:
161 self.logger.debug("delete the %s", vnf["vnf_name"])
162 self.cloud.delete_server(vnf["vnf_name"])
164 def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
166 network_list = self.get_blueprint_outputs_networks(
168 topology_deploy_name)
169 vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
170 topology_deploy_name)
171 for vnf in vnf_info_list:
172 vnf_name = vnf["vnf_name"]
173 vnf["os_type"] = self.image["os_type"]
174 vnf["user"] = self.image["user"]
175 vnf["pass"] = self.image["pass"]
177 vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
179 self.logger.debug("vnf name : %s", vnf_name)
180 self.logger.debug(vnf_name + " floating ip address : " +
183 for network in network_list:
184 network_name = network["network_name"]
185 ip_address = self.get_address(vnf["vnf_name"],
186 network["network_name"])
187 vnf[network_name + "_ip"] = ip_address
188 mac = self.get_mac_address(vnf["vnf_name"],
189 network["network_name"])
190 vnf[network_name + "_mac"] = mac
192 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
193 " : " + vnf[network_name + "_ip"])
194 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
195 " : " + vnf[network_name + "_mac"])
200 def get_target_vnf(vnf_info_list):
201 for vnf in vnf_info_list:
202 if vnf["target_vnf_flag"]:
208 def get_reference_vnf_list(vnf_info_list):
209 reference_vnf_list = []
210 for vnf in vnf_info_list:
211 if not vnf["target_vnf_flag"]:
212 reference_vnf_list.append(vnf)
214 return reference_vnf_list
217 def get_vnf_info(vnf_info_list, vnf_name):
218 for vnf in vnf_info_list:
219 if vnf["vnf_name"] == vnf_name:
225 def convert_functional_test_result(result_data_list):
227 for result_data in result_data_list:
228 test_kind = result_data["test_kind"]
229 protocol = result_data["protocol"]
230 test_result_data = result_data["result"]
232 if test_kind not in result:
233 result[test_kind] = []
235 result[test_kind].append({protocol: test_result_data})
237 return {"Functional_test": result}
239 def write_result_data(self, result_data):
241 if not os.path.isfile(self.test_result_json_file):
242 file_fd = open(self.test_result_json_file, "w")
245 file_fd = open(self.test_result_json_file, "r")
246 test_result = json.load(file_fd)
249 test_result.append(result_data)
251 file_fd = open(self.test_result_json_file, "w")
252 json.dump(test_result, file_fd)
255 def output_test_result_json(self):
256 if os.path.isfile(self.test_result_json_file):
257 file_fd = open(self.test_result_json_file, "r")
258 test_result = json.load(file_fd)
260 output_json_data = json.dumps(test_result,
263 self.logger.debug("test_result %s", output_json_data)
265 self.logger.debug("Not found %s", self.test_result_json_file)
268 def get_test_scenario(file_path):
269 test_scenario_file = open(file_path,
271 test_scenario_yaml = yaml.safe_load(test_scenario_file)
272 test_scenario_file.close()
273 return test_scenario_yaml["test_scenario_list"]