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 pkg_resources
  18 import requests
  19 import yaml
  20
  21 from functest.utils.constants import CONST
  22 from git import Repo
  23 from requests.auth import HTTPBasicAuth
  24 from snaps.openstack.utils import nova_utils
  25
  26
  27 RESULT_SPRIT_INDEX = {
  28     "transfer": 8,
  29     "bandwidth": 6,
  30     "jitter": 4,
  31     "los_total": 2,
  32     "pkt_loss": 1
  33 }
  34
  35 BIT_PER_BYTE = 8
  36
  37 NOVA_CLIENT_API_VERSION = '2'
  38 NOVA_CILENT_NETWORK_INFO_INDEX = 0
  39 CFY_INFO_OUTPUT_FILE = "output.txt"
  40
  41 CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
  42 PACKET_LOST_INFO_INDEX = 0
  43 PACKET_TOTAL_INFO_INDEX = 1
  44
  45 NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
  46 NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
  47 NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
  48 NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
  49
  50
  51 class Utilvnf(object):  # pylint: disable=too-many-instance-attributes
  52     """ Utility class of vrouter testcase """
  53
  54     logger = logging.getLogger(__name__)
  55
  56     def __init__(self):
  57         self.snaps_creds = ""
  58
  59         data_dir = data_dir = CONST.__getattribute__('dir_router_data')
  60
  61         self.vnf_data_dir = data_dir
  62         self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
  63         self.command_template_dir = "command_template/"
  64         self.test_scenario_yaml = "test_scenario.yaml"
  65         test_env_config_yaml_file = "test_env_config.yaml"
  66         self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
  67         self.test_env_config_yaml = os.path.join(
  68             self.vnf_data_dir,
  69             self.opnfv_vnf_data_dir,
  70             test_env_config_yaml_file)
  71
  72         self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
  73         self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
  74
  75         if not os.path.exists(self.vnf_data_dir):
  76             os.makedirs(self.vnf_data_dir)
  77
  78         case_dir = pkg_resources.resource_filename(
  79             'functest', 'opnfv_tests/vnf/router')
  80
  81         config_file_name = CONST.__getattribute__(
  82             'vnf_{}_config'.format("vyos_vrouter"))
  83
  84         config_file = os.path.join(case_dir, config_file_name)
  85
  86         with open(config_file) as file_fd:
  87             vrouter_config_yaml = yaml.safe_load(file_fd)
  88         file_fd.close()
  89
  90         test_data = vrouter_config_yaml.get("test_data")
  91
  92         self.logger.debug("Downloading the test data.")
  93         vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
  94
  95         if not os.path.exists(vrouter_data_path):
  96             Repo.clone_from(test_data['url'],
  97                             vrouter_data_path,
  98                             branch=test_data['branch'])
  99
 100         with open(self.test_env_config_yaml) as file_fd:
 101             test_env_config_yaml = yaml.safe_load(file_fd)
 102         file_fd.close()
 103
 104         self.image = test_env_config_yaml.get(
 105             "general").get("images").get("vyos")
 106         self.tester_image = test_env_config_yaml.get(
 107             "general").get("images").get("tester_vm_os")
 108
 109         self.test_result_json_file = "test_result.json"
 110         if os.path.isfile(self.test_result_json_file):
 111             os.remove(self.test_result_json_file)
 112             self.logger.debug("removed %s", self.test_result_json_file)
 113
 114     def get_nova_client(self):
 115         nova_client = nova_utils.nova_client(self.snaps_creds)
 116
 117         return nova_client
 118
 119     def set_credentials(self, snaps_creds):
 120         self.snaps_creds = snaps_creds
 121
 122     def get_address(self, server_name, network_name):
 123         nova_client = self.get_nova_client()
 124         servers_list = nova_client.servers.list()
 125         server = None
 126
 127         for server in servers_list:
 128             if server.name == server_name:
 129                 break
 130
 131         address = server.addresses[
 132             network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
 133
 134         return address
 135
 136     def get_mac_address(self, server_name, network_name):
 137         nova_client = self.get_nova_client()
 138         servers_list = nova_client.servers.list()
 139         server = None
 140
 141         for server in servers_list:
 142             if server.name == server_name:
 143                 break
 144
 145         mac_address = server.addresses[network_name][
 146             NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
 147
 148         return mac_address
 149
 150     def reboot_vm(self, server_name):
 151         nova_client = self.get_nova_client()
 152         servers_list = nova_client.servers.list()
 153         server = None
 154
 155         for server in servers_list:
 156             if server.name == server_name:
 157                 break
 158
 159         server.reboot()
 160
 161         return
 162
 163     def delete_vm(self, server_name):
 164         nova_client = self.get_nova_client()
 165         servers_list = nova_client.servers.list()
 166         server = None
 167
 168         for server in servers_list:
 169             if server.name == server_name:
 170                 nova_client.servers.delete(server)
 171                 break
 172
 173         return
 174
 175     def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
 176         url = "http://%s/deployments/%s/outputs" % (
 177             cfy_manager_ip, deployment_name)
 178
 179         response = requests.get(
 180             url,
 181             auth=HTTPBasicAuth('admin', 'admin'),
 182             headers={'Tenant': 'default_tenant'})
 183
 184         resp_data = response.json()
 185         self.logger.debug(resp_data)
 186         data = resp_data["outputs"]
 187         return data
 188
 189     def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
 190         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 191                                              deployment_name)
 192         vnfs = outputs["vnfs"]
 193         vnf_list = []
 194         for vnf_name in vnfs:
 195             vnf_list.append(vnfs[vnf_name])
 196         return vnf_list
 197
 198     def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
 199         outputs = self.get_blueprint_outputs(cfy_manager_ip,
 200                                              deployment_name)
 201         networks = outputs["networks"]
 202         network_list = []
 203         for network_name in networks:
 204             network_list.append(networks[network_name])
 205         return network_list
 206
 207     def request_vnf_reboot(self, vnf_info_list):
 208         for vnf in vnf_info_list:
 209             self.logger.debug("reboot the " + vnf["vnf_name"])
 210             self.reboot_vm(vnf["vnf_name"])
 211
 212     def request_vm_delete(self, vnf_info_list):
 213         for vnf in vnf_info_list:
 214             self.logger.debug("delete the " + vnf["vnf_name"])
 215             self.delete_vm(vnf["vnf_name"])
 216
 217     def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
 218                           target_vnf_name):
 219         network_list = self.get_blueprint_outputs_networks(
 220             cfy_manager_ip,
 221             topology_deploy_name)
 222         vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
 223                                                         topology_deploy_name)
 224         for vnf in vnf_info_list:
 225             vnf_name = vnf["vnf_name"]
 226             vnf["os_type"] = self.image["os_type"]
 227             vnf["user"] = self.image["user"]
 228             vnf["pass"] = self.image["pass"]
 229
 230             vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
 231
 232             self.logger.debug("vnf name : " + vnf_name)
 233             self.logger.debug(vnf_name + " floating ip address : " +
 234                               vnf["floating_ip"])
 235
 236             for network in network_list:
 237                 network_name = network["network_name"]
 238                 ip_address = self.get_address(vnf["vnf_name"],
 239                                               network["network_name"])
 240                 vnf[network_name + "_ip"] = ip_address
 241                 mac = self.get_mac_address(vnf["vnf_name"],
 242                                            network["network_name"])
 243                 vnf[network_name + "_mac"] = mac
 244
 245                 self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
 246                                   " : " + vnf[network_name + "_ip"])
 247                 self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
 248                                   " : " + vnf[network_name + "_mac"])
 249
 250         return vnf_info_list
 251
 252     @staticmethod
 253     def get_target_vnf(vnf_info_list):
 254         for vnf in vnf_info_list:
 255             if vnf["target_vnf_flag"]:
 256                 return vnf
 257
 258         return None
 259
 260     @staticmethod
 261     def get_reference_vnf_list(vnf_info_list):
 262         reference_vnf_list = []
 263         for vnf in vnf_info_list:
 264             if not vnf["target_vnf_flag"]:
 265                 reference_vnf_list.append(vnf)
 266
 267         return reference_vnf_list
 268
 269     @staticmethod
 270     def get_vnf_info(vnf_info_list, vnf_name):
 271         for vnf in vnf_info_list:
 272             if vnf["vnf_name"] == vnf_name:
 273                 return vnf
 274
 275         return None
 276
 277     @staticmethod
 278     def convert_functional_test_result(result_data_list):
 279         result = {}
 280         for result_data in result_data_list:
 281             test_kind = result_data["test_kind"]
 282             protocol = result_data["protocol"]
 283             test_result_data = result_data["result"]
 284
 285             if test_kind not in result:
 286                 result[test_kind] = []
 287
 288             result[test_kind].append({protocol: test_result_data})
 289
 290         return {"Functional_test": result}
 291
 292     def write_result_data(self, result_data):
 293         test_result = []
 294         if not os.path.isfile(self.test_result_json_file):
 295             file_fd = open(self.test_result_json_file, "w")
 296             file_fd.close()
 297         else:
 298             file_fd = open(self.test_result_json_file, "r")
 299             test_result = json.load(file_fd)
 300             file_fd.close()
 301
 302         test_result.append(result_data)
 303
 304         file_fd = open(self.test_result_json_file, "w")
 305         json.dump(test_result, file_fd)
 306         file_fd.close()
 307
 308     def output_test_result_json(self):
 309         if os.path.isfile(self.test_result_json_file):
 310             file_fd = open(self.test_result_json_file, "r")
 311             test_result = json.load(file_fd)
 312             file_fd.close()
 313             output_json_data = json.dumps(test_result,
 314                                           sort_keys=True,
 315                                           indent=4)
 316             self.logger.debug("test_result %s", output_json_data)
 317         else:
 318             self.logger.debug("Not found %s", self.test_result_json_file)
 319
 320     @staticmethod
 321     def get_test_scenario(file_path):
 322         test_scenario_file = open(file_path,
 323                                   'r')
 324         test_scenario_yaml = yaml.safe_load(test_scenario_file)
 325         test_scenario_file.close()
 326         return test_scenario_yaml["test_scenario_list"]