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) 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 = "http://%s/deployments/%s/outputs" % (
 102             cfy_manager_ip, deployment_name)
 103
 104         response = requests.get(
 105             url,
 106             auth=requests.auth.HTTPBasicAuth('admin', 'admin'),
 107             headers={'Tenant': 'default_tenant'})
 108
 109         resp_data = response.json()
 110         self.logger.debug(resp_data)
 111         data = resp_data["outputs"]
 112         return data
 113
 114     def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
 115         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 116                                              deployment_name)
 117         vnfs = outputs["vnfs"]
 118         vnf_list = []
 119         for vnf_name in vnfs:
 120             vnf_list.append(vnfs[vnf_name])
 121         return vnf_list
 122
 123     def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
 124         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 125                                              deployment_name)
 126         networks = outputs["networks"]
 127         network_list = []
 128         for network_name in networks:
 129             network_list.append(networks[network_name])
 130         return network_list
 131
 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"])
 136
 137     def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
 138                           target_vnf_name):
 139         network_list = self.get_blueprint_outputs_networks(
 140             cfy_manager_ip,
 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"]
 149
 150             vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
 151
 152             self.logger.debug("vnf name : %s", vnf_name)
 153             self.logger.debug(vnf_name + " floating ip address : " +
 154                               vnf["floating_ip"])
 155
 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
 164
 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"])
 169
 170         return vnf_info_list
 171
 172     @staticmethod
 173     def get_target_vnf(vnf_info_list):
 174         for vnf in vnf_info_list:
 175             if vnf["target_vnf_flag"]:
 176                 return vnf
 177
 178         return None
 179
 180     @staticmethod
 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)
 186
 187         return reference_vnf_list
 188
 189     @staticmethod
 190     def get_vnf_info(vnf_info_list, vnf_name):
 191         for vnf in vnf_info_list:
 192             if vnf["vnf_name"] == vnf_name:
 193                 return vnf
 194
 195         return None
 196
 197     @staticmethod
 198     def convert_functional_test_result(result_data_list):
 199         result = {}
 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"]
 204
 205             if test_kind not in result:
 206                 result[test_kind] = []
 207
 208             result[test_kind].append({protocol: test_result_data})
 209
 210         return {"Functional_test": result}
 211
 212     def write_result_data(self, result_data):
 213         test_result = []
 214         if not os.path.isfile(self.test_result_json_file):
 215             with open(self.test_result_json_file, "w") as file_fd:
 216                 pass
 217         else:
 218             with open(self.test_result_json_file, "r") as file_fd:
 219                 test_result = json.load(file_fd)
 220
 221         test_result.append(result_data)
 222
 223         with open(self.test_result_json_file, "w") as file_fd:
 224             json.dump(test_result, file_fd)
 225
 226     def output_test_result_json(self):
 227         if os.path.isfile(self.test_result_json_file):
 228             with open(self.test_result_json_file, "r") as file_fd:
 229                 test_result = json.load(file_fd)
 230             output_json_data = json.dumps(test_result,
 231                                           sort_keys=True,
 232                                           indent=4)
 233             self.logger.debug("test_result %s", output_json_data)
 234         else:
 235             self.logger.debug("Not found %s", self.test_result_json_file)
 236
 237     @staticmethod
 238     def get_test_scenario(file_path):
 239         with open(file_path, 'r') as test_scenario_file:
 240             test_scenario_yaml = yaml.safe_load(test_scenario_file)
 241         return test_scenario_yaml["test_scenario_list"]