functest/opnfv_tests/vnf/router/utilvnf.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2017 Okinawa Open Laboratory and others.
   4 #
   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
   9
  10 # pylint: disable=missing-docstring
  11
  12 """ Utility module of vrouter testcase """
  13
  14 import json
  15 import logging
  16 import os
  17 import requests
  18 import yaml
  19
  20 from functest.utils import config
  21
  22 RESULT_SPRIT_INDEX = {
  23     "transfer": 8,
  24     "bandwidth": 6,
  25     "jitter": 4,
  26     "los_total": 2,
  27     "pkt_loss": 1
  28 }
  29
  30 BIT_PER_BYTE = 8
  31
  32 NOVA_CLIENT_API_VERSION = '2'
  33 NOVA_CILENT_NETWORK_INFO_INDEX = 0
  34 CFY_INFO_OUTPUT_FILE = "output.txt"
  35
  36 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
  37 PACKET_LOST_INFO_INDEX = 0
  38 PACKET_TOTAL_INFO_INDEX = 1
  39
  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
  44
  45
  46 class Utilvnf():  # pylint: disable=too-many-instance-attributes
  47     """ Utility class of vrouter testcase """
  48
  49     logger = logging.getLogger(__name__)
  50
  51     def __init__(self):
  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(
  58             self.vnf_data_dir,
  59             test_env_config_yaml_file)
  60
  61         self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
  62         self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
  63
  64         if not os.path.exists(self.vnf_data_dir):
  65             os.makedirs(self.vnf_data_dir)
  66
  67         with open(self.test_env_config_yaml, encoding='utf-8') as file_fd:
  68             test_env_config_yaml = yaml.safe_load(file_fd)
  69         file_fd.close()
  70
  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")
  75
  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)
  80
  81         self.cloud = None
  82
  83     def set_credentials(self, cloud):
  84         self.cloud = cloud
  85
  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"]
  90
  91         return address
  92
  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"]
  97
  98         return mac_address
  99
 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(
 103             url,
 104             auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
 105             headers={'Tenant': 'default_tenant'})
 106
 107         resp_data = response.json()
 108         self.logger.debug(resp_data)
 109         data = resp_data["outputs"]
 110         return data
 111
 112     def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
 113         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 114                                              deployment_name)
 115         vnfs = outputs["vnfs"]
 116         vnf_list = []
 117         for vnf_name in vnfs:
 118             vnf_list.append(vnfs[vnf_name])
 119         return vnf_list
 120
 121     def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
 122         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 123                                              deployment_name)
 124         networks = outputs["networks"]
 125         network_list = []
 126         for network_name in networks:
 127             network_list.append(networks[network_name])
 128         return network_list
 129
 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"])
 134
 135     def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
 136                           target_vnf_name):
 137         network_list = self.get_blueprint_outputs_networks(
 138             cfy_manager_ip,
 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"]
 147
 148             vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
 149
 150             self.logger.debug("vnf name : %s", vnf_name)
 151             self.logger.debug(vnf_name + " floating ip address : " +
 152                               vnf["floating_ip"])
 153
 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
 162
 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"])
 167
 168         return vnf_info_list
 169
 170     @staticmethod
 171     def get_target_vnf(vnf_info_list):
 172         for vnf in vnf_info_list:
 173             if vnf["target_vnf_flag"]:
 174                 return vnf
 175
 176         return None
 177
 178     @staticmethod
 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)
 184
 185         return reference_vnf_list
 186
 187     @staticmethod
 188     def get_vnf_info(vnf_info_list, vnf_name):
 189         for vnf in vnf_info_list:
 190             if vnf["vnf_name"] == vnf_name:
 191                 return vnf
 192
 193         return None
 194
 195     @staticmethod
 196     def convert_functional_test_result(result_data_list):
 197         result = {}
 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"]
 202
 203             if test_kind not in result:
 204                 result[test_kind] = []
 205
 206             result[test_kind].append({protocol: test_result_data})
 207
 208         return {"Functional_test": result}
 209
 210     def write_result_data(self, result_data):
 211         test_result = []
 212         if not os.path.isfile(self.test_result_json_file):
 213             with open(
 214                     self.test_result_json_file, "w",
 215                     encoding="utf-8") as file_fd:
 216                 pass
 217         else:
 218             with open(
 219                     self.test_result_json_file, "r",
 220                     encoding="utf-8") as file_fd:
 221                 test_result = json.load(file_fd)
 222
 223         test_result.append(result_data)
 224
 225         with open(
 226                 self.test_result_json_file, "w",
 227                 encoding="utf-8") as file_fd:
 228             json.dump(test_result, file_fd)
 229
 230     def output_test_result_json(self):
 231         if os.path.isfile(self.test_result_json_file):
 232             with open(
 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,
 237                                           sort_keys=True,
 238                                           indent=4)
 239             self.logger.debug("test_result %s", output_json_data)
 240         else:
 241             self.logger.debug("Not found %s", self.test_result_json_file)
 242
 243     @staticmethod
 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"]