Refactor the vyos_vrouter to adopt VNF abstraction

JIRA: FUNCTEST-788

1.Modifying code of vyos_vrouter to inherit vnf abstraction class.
2.Adding vyos_vrouter code from our repo to functest.
3.Adding unit test of vyos_vrouter.
4.Doing test of modified vyos_vrouter codes on our labs.

Change-Id: I77e4be8b2a140ea0176c607f2be736599f893ace
Signed-off-by: Shuya Nakama <shuya.nakama@okinawaopenlabs.org>

1 files changed, 345 insertions, 0 deletions

diff --git a/functest/opnfv_tests/vnf/router/utilvnf.py b/functest/opnfv_tests/vnf/router/utilvnf.py
new file mode 100644
index 000000000..084af3312
--- /dev/null
+++ b/functest/opnfv_tests/vnf/router/utilvnf.py
@@ -0,0 +1,345 @@
+#!/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
+
+""" Utility module of vrouter testcase """
+
+import json
+import logging
+import os
+import pkg_resources
+import requests
+import yaml
+
+from functest.utils.constants import CONST
+from git import Repo
+from novaclient import client as novaclient
+from keystoneauth1 import session
+from keystoneauth1 import loading
+from requests.auth import HTTPBasicAuth
+
+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):
+    """ Utility class of vrouter testcase """
+
+    logger = logging.getLogger(__name__)
+
+    def __init__(self):
+        self.username = ""
+        self.password = ""
+        self.auth_url = ""
+        self.tenant_name = ""
+        self.region_name = ""
+
+        data_dir = data_dir = CONST.__getattribute__('dir_router_data')
+
+        self.vnf_data_dir = data_dir
+        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 = CONST.__getattribute__(
+            '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):
+        creds = self.get_nova_credentials()
+        loader = loading.get_plugin_loader('password')
+        auth = loader.load_from_options(**creds)
+        sess = session.Session(auth=auth)
+        nova_client = novaclient.Client(NOVA_CLIENT_API_VERSION, session=sess)
+
+        return nova_client
+
+    def set_credentials(self, username, password, auth_url,
+                        tenant_name, region_name="RegionOne"):
+        self.username = username
+        self.password = password
+        self.auth_url = auth_url
+        self.tenant_name = tenant_name
+        self.region_name = region_name
+
+    def get_nova_credentials(self):
+        creds = {}
+        creds['username'] = self.username
+        creds['password'] = self.password
+        creds['auth_url'] = self.auth_url
+        creds['tenant_name'] = self.tenant_name
+        return 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"]
+
+            if vnf_name == target_vnf_name:
+                vnf["target_vnf_flag"] = True
+            else:
+                vnf["target_vnf_flag"] = False
+
+            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
+
+    def get_target_vnf(self, vnf_info_list):
+        for vnf in vnf_info_list:
+            if vnf["target_vnf_flag"]:
+                return vnf
+
+        return None
+
+    def get_reference_vnf_list(self, 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
+
+    def get_vnf_info(self, vnf_info_list, vnf_name):
+        for vnf in vnf_info_list:
+            if vnf["vnf_name"] == vnf_name:
+                return vnf
+
+        return None
+
+    def convert_functional_test_result(self, 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)
+
+    def get_test_scenario(self, 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"]