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