Clean all OpenStack related modules

Xtesting is only focused on the framework and entry points.

Change-Id: I1a4146ed8519438b13810a20ddf1140c35bb6ecd
Signed-off-by: Cédric Ollivier <cedric.ollivier@orange.com>

1 files changed, 0 insertions, 323 deletions

diff --git a/functest/opnfv_tests/vnf/router/utilvnf.py b/functest/opnfv_tests/vnf/router/utilvnf.py
deleted file mode 100644
index 6861b386..00000000
--- a/functest/opnfv_tests/vnf/router/utilvnf.py
+++ /dev/null
@@ -1,323 +0,0 @@
-#!/usr/bin/env python
-
-# Copyright (c) 2017 Okinawa Open Laboratory and others.
-#
-# All rights reserved. This program and the accompanying materials
-# are made available under the terms of the Apache License, Version 2.0
-# which accompanies this distribution, and is available at
-# http://www.apache.org/licenses/LICENSE-2.0
-
-# pylint: disable=missing-docstring
-
-""" Utility module of vrouter testcase """
-
-import json
-import logging
-import os
-import pkg_resources
-import requests
-import yaml
-
-from functest.utils import config
-from git import Repo
-from requests.auth import HTTPBasicAuth
-from snaps.openstack.utils import nova_utils
-
-
-RESULT_SPRIT_INDEX = {
-    "transfer": 8,
-    "bandwidth": 6,
-    "jitter": 4,
-    "los_total": 2,
-    "pkt_loss": 1
-}
-
-BIT_PER_BYTE = 8
-
-NOVA_CLIENT_API_VERSION = '2'
-NOVA_CILENT_NETWORK_INFO_INDEX = 0
-CFY_INFO_OUTPUT_FILE = "output.txt"
-
-CIDR_NETWORK_SEGMENT_INFO_INDEX = 0
-PACKET_LOST_INFO_INDEX = 0
-PACKET_TOTAL_INFO_INDEX = 1
-
-NUMBER_OF_DIGITS_FOR_AVG_TRANSFER = 0
-NUMBER_OF_DIGITS_FOR_AVG_BANDWIDTH = 0
-NUMBER_OF_DIGITS_FOR_AVG_JITTER = 3
-NUMBER_OF_DIGITS_FOR_AVG_PKT_LOSS = 1
-
-
-class Utilvnf(object):  # pylint: disable=too-many-instance-attributes
-    """ Utility class of vrouter testcase """
-
-    logger = logging.getLogger(__name__)
-
-    def __init__(self):
-        self.snaps_creds = ""
-        self.vnf_data_dir = getattr(config.CONF, 'dir_router_data')
-        self.opnfv_vnf_data_dir = "opnfv-vnf-data/"
-        self.command_template_dir = "command_template/"
-        self.test_scenario_yaml = "test_scenario.yaml"
-        test_env_config_yaml_file = "test_env_config.yaml"
-        self.test_cmd_map_yaml_file = "test_cmd_map.yaml"
-        self.test_env_config_yaml = os.path.join(
-            self.vnf_data_dir,
-            self.opnfv_vnf_data_dir,
-            test_env_config_yaml_file)
-
-        self.blueprint_dir = "opnfv-vnf-vyos-blueprint/"
-        self.blueprint_file_name = "function-test-openstack-blueprint.yaml"
-
-        if not os.path.exists(self.vnf_data_dir):
-            os.makedirs(self.vnf_data_dir)
-
-        case_dir = pkg_resources.resource_filename(
-            'functest', 'opnfv_tests/vnf/router')
-
-        config_file_name = getattr(
-            config.CONF, 'vnf_{}_config'.format("vyos_vrouter"))
-
-        config_file = os.path.join(case_dir, config_file_name)
-
-        with open(config_file) as file_fd:
-            vrouter_config_yaml = yaml.safe_load(file_fd)
-        file_fd.close()
-
-        test_data = vrouter_config_yaml.get("test_data")
-
-        self.logger.debug("Downloading the test data.")
-        vrouter_data_path = self.vnf_data_dir + self.opnfv_vnf_data_dir
-
-        if not os.path.exists(vrouter_data_path):
-            Repo.clone_from(test_data['url'],
-                            vrouter_data_path,
-                            branch=test_data['branch'])
-
-        with open(self.test_env_config_yaml) as file_fd:
-            test_env_config_yaml = yaml.safe_load(file_fd)
-        file_fd.close()
-
-        self.image = test_env_config_yaml.get(
-            "general").get("images").get("vyos")
-        self.tester_image = test_env_config_yaml.get(
-            "general").get("images").get("tester_vm_os")
-
-        self.test_result_json_file = "test_result.json"
-        if os.path.isfile(self.test_result_json_file):
-            os.remove(self.test_result_json_file)
-            self.logger.debug("removed %s", self.test_result_json_file)
-
-    def get_nova_client(self):
-        nova_client = nova_utils.nova_client(self.snaps_creds)
-
-        return nova_client
-
-    def set_credentials(self, snaps_creds):
-        self.snaps_creds = snaps_creds
-
-    def get_address(self, server_name, network_name):
-        nova_client = self.get_nova_client()
-        servers_list = nova_client.servers.list()
-        server = None
-
-        for server in servers_list:
-            if server.name == server_name:
-                break
-
-        address = server.addresses[
-            network_name][NOVA_CILENT_NETWORK_INFO_INDEX]["addr"]
-
-        return address
-
-    def get_mac_address(self, server_name, network_name):
-        nova_client = self.get_nova_client()
-        servers_list = nova_client.servers.list()
-        server = None
-
-        for server in servers_list:
-            if server.name == server_name:
-                break
-
-        mac_address = server.addresses[network_name][
-            NOVA_CILENT_NETWORK_INFO_INDEX]["OS-EXT-IPS-MAC:mac_addr"]
-
-        return mac_address
-
-    def reboot_vm(self, server_name):
-        nova_client = self.get_nova_client()
-        servers_list = nova_client.servers.list()
-        server = None
-
-        for server in servers_list:
-            if server.name == server_name:
-                break
-
-        server.reboot()
-
-        return
-
-    def delete_vm(self, server_name):
-        nova_client = self.get_nova_client()
-        servers_list = nova_client.servers.list()
-        server = None
-
-        for server in servers_list:
-            if server.name == server_name:
-                nova_client.servers.delete(server)
-                break
-
-        return
-
-    def get_blueprint_outputs(self, cfy_manager_ip, deployment_name):
-        url = "http://%s/deployments/%s/outputs" % (
-            cfy_manager_ip, deployment_name)
-
-        response = requests.get(
-            url,
-            auth=HTTPBasicAuth('admin', 'admin'),
-            headers={'Tenant': 'default_tenant'})
-
-        resp_data = response.json()
-        self.logger.debug(resp_data)
-        data = resp_data["outputs"]
-        return data
-
-    def get_blueprint_outputs_vnfs(self, cfy_manager_ip, deployment_name):
-        outputs = self.get_blueprint_outputs(cfy_manager_ip,
-                                             deployment_name)
-        vnfs = outputs["vnfs"]
-        vnf_list = []
-        for vnf_name in vnfs:
-            vnf_list.append(vnfs[vnf_name])
-        return vnf_list
-
-    def get_blueprint_outputs_networks(self, cfy_manager_ip, deployment_name):
-        outputs = self.get_blueprint_outputs(cfy_manager_ip,
-                                             deployment_name)
-        networks = outputs["networks"]
-        network_list = []
-        for network_name in networks:
-            network_list.append(networks[network_name])
-        return network_list
-
-    def request_vnf_reboot(self, vnf_info_list):
-        for vnf in vnf_info_list:
-            self.logger.debug("reboot the " + vnf["vnf_name"])
-            self.reboot_vm(vnf["vnf_name"])
-
-    def request_vm_delete(self, vnf_info_list):
-        for vnf in vnf_info_list:
-            self.logger.debug("delete the " + vnf["vnf_name"])
-            self.delete_vm(vnf["vnf_name"])
-
-    def get_vnf_info_list(self, cfy_manager_ip, topology_deploy_name,
-                          target_vnf_name):
-        network_list = self.get_blueprint_outputs_networks(
-            cfy_manager_ip,
-            topology_deploy_name)
-        vnf_info_list = self.get_blueprint_outputs_vnfs(cfy_manager_ip,
-                                                        topology_deploy_name)
-        for vnf in vnf_info_list:
-            vnf_name = vnf["vnf_name"]
-            vnf["os_type"] = self.image["os_type"]
-            vnf["user"] = self.image["user"]
-            vnf["pass"] = self.image["pass"]
-
-            vnf["target_vnf_flag"] = bool(vnf_name == target_vnf_name)
-
-            self.logger.debug("vnf name : " + vnf_name)
-            self.logger.debug(vnf_name + " floating ip address : " +
-                              vnf["floating_ip"])
-
-            for network in network_list:
-                network_name = network["network_name"]
-                ip_address = self.get_address(vnf["vnf_name"],
-                                              network["network_name"])
-                vnf[network_name + "_ip"] = ip_address
-                mac = self.get_mac_address(vnf["vnf_name"],
-                                           network["network_name"])
-                vnf[network_name + "_mac"] = mac
-
-                self.logger.debug(network_name + "_ip of " + vnf["vnf_name"] +
-                                  " : " + vnf[network_name + "_ip"])
-                self.logger.debug(network_name + "_mac of " + vnf["vnf_name"] +
-                                  " : " + vnf[network_name + "_mac"])
-
-        return vnf_info_list
-
-    @staticmethod
-    def get_target_vnf(vnf_info_list):
-        for vnf in vnf_info_list:
-            if vnf["target_vnf_flag"]:
-                return vnf
-
-        return None
-
-    @staticmethod
-    def get_reference_vnf_list(vnf_info_list):
-        reference_vnf_list = []
-        for vnf in vnf_info_list:
-            if not vnf["target_vnf_flag"]:
-                reference_vnf_list.append(vnf)
-
-        return reference_vnf_list
-
-    @staticmethod
-    def get_vnf_info(vnf_info_list, vnf_name):
-        for vnf in vnf_info_list:
-            if vnf["vnf_name"] == vnf_name:
-                return vnf
-
-        return None
-
-    @staticmethod
-    def convert_functional_test_result(result_data_list):
-        result = {}
-        for result_data in result_data_list:
-            test_kind = result_data["test_kind"]
-            protocol = result_data["protocol"]
-            test_result_data = result_data["result"]
-
-            if test_kind not in result:
-                result[test_kind] = []
-
-            result[test_kind].append({protocol: test_result_data})
-
-        return {"Functional_test": result}
-
-    def write_result_data(self, result_data):
-        test_result = []
-        if not os.path.isfile(self.test_result_json_file):
-            file_fd = open(self.test_result_json_file, "w")
-            file_fd.close()
-        else:
-            file_fd = open(self.test_result_json_file, "r")
-            test_result = json.load(file_fd)
-            file_fd.close()
-
-        test_result.append(result_data)
-
-        file_fd = open(self.test_result_json_file, "w")
-        json.dump(test_result, file_fd)
-        file_fd.close()
-
-    def output_test_result_json(self):
-        if os.path.isfile(self.test_result_json_file):
-            file_fd = open(self.test_result_json_file, "r")
-            test_result = json.load(file_fd)
-            file_fd.close()
-            output_json_data = json.dumps(test_result,
-                                          sort_keys=True,
-                                          indent=4)
-            self.logger.debug("test_result %s", output_json_data)
-        else:
-            self.logger.debug("Not found %s", self.test_result_json_file)
-
-    @staticmethod
-    def get_test_scenario(file_path):
-        test_scenario_file = open(file_path,
-                                  'r')
-        test_scenario_yaml = yaml.safe_load(test_scenario_file)
-        test_scenario_file.close()
-        return test_scenario_yaml["test_scenario_list"]