utils/__init__.py

import logging
import os
import sys
import yaml

from kubernetes import client as kclient, config as kconfig

from utils import exceptions
from utils import os_client
from utils import k8s_client

logger = logging.getLogger(__name__)


def compile_pairs(nodes):
    result = {}
    if len(nodes) % 2 != 0:
        nodes.pop(1)
    pairs = list(zip(*[iter(nodes)] * 2))
    for pair in pairs:
        result[pair[0] + '<>' + pair[1]] = pair
    return result


def get_pairs():
    config = get_configuration()
    cmp_hosts = config.get('CMP_HOSTS') or []
    skipped_nodes = config.get('skipped_nodes') or []
    if skipped_nodes:
        sys.stdout.write(("\nNotice: {} node(s) will be skipped for vm2vm "
                          "test.\n".format(", ".join(skipped_nodes))))
        logger.info("Skipping nodes {}".format(",".join(skipped_nodes)))
    if not cmp_hosts:
        openstack_clients = os_client.OfficialClientManager(
            username=os.environ['OS_USERNAME'],
            password=os.environ['OS_PASSWORD'],
            project_name=os.environ['OS_PROJECT_NAME'],
            auth_url=os.environ['OS_AUTH_URL'],
            cert=False,
            domain=os.environ['OS_PROJECT_DOMAIN_NAME']
        )
        os_actions = os_client.OSCliActions(openstack_clients)
        nova_computes = os_actions.list_nova_computes()
        if len(nova_computes) < 2:
            raise exceptions.NotEnoughNodes(
                "At least 2 compute hosts are needed for VM2VM test, "
                "now: {}.".format(len(nova_computes)))
        cmp_hosts = [n.host_name for n in nova_computes
                     if n.host_name not in skipped_nodes]
        if len(cmp_hosts) < 2:
            raise exceptions.NotEnoughNodes(
                "At least 2 compute hosts are needed for VM2VM test. "
                "Cannot create a pair from {}. Please check skip list, at "
                "least 2 computes should be tested.".format(cmp_hosts))
        logger.info("CMP_HOSTS option is not set, using host pair from "
                    "Nova compute list. Pair generated: {}".format(cmp_hosts))

    return compile_pairs(cmp_hosts)


def get_hw_pairs():
    # get the K8S config, check whether the HW nodes list is set
    config = get_configuration()
    logger.info("Getting the K8S config path from the global_config.yaml file.")
    k8s_config_path = config.get("mos_kubeconfig_path", "")
    hw_nodes_list = config.get("hw_nodes_list", [])

    # if the specific HW nodes list is not set in the config, get from K8S
    hw_nodes = None
    if not hw_nodes_list:
        # fetch only compute nodes
        label_selector = "openstack-compute-node=enabled," \
                         "openvswitch=enabled"
        k8s_api = k8s_client.K8SClientManager(k8s_config_path=k8s_config_path)
        k8s_actions = k8s_client.K8SCliActions(k8s_api.k8s_v1)
        hw_nodes_list = k8s_actions.list_nodes_names(
            label_selector=label_selector)

    # remove some skipped nodes if any
    skipped_nodes = config.get('skipped_nodes', [])
    if skipped_nodes:
        print(f"Notice: {', '.join(skipped_nodes)} node(s) will be skipped for"
              f" hw2hw test.\n")
    hw_nodes = [node for node in hw_nodes_list
                if node not in skipped_nodes]
    if len(hw_nodes) < 2:
        raise exceptions.NotEnoughNodes(
            f"At least 2 HW nodes are required to run hw2hw test. Cannot "
            f"create a pair from {hw_nodes}. Check whether the cluster has at"
            f" least 2 compute nodes, or the nodes are not in the skip list.")
    return compile_pairs(hw_nodes)


def get_configuration():
    """function returns configuration for environment
    and for test if it's specified"""

    global_config_file = os.path.join(
        os.path.dirname(os.path.abspath(__file__)), "../global_config.yaml")
    with open(global_config_file, 'r') as file:
        global_config = yaml.load(file, Loader=yaml.SafeLoader)
    for param in list(global_config.keys()):
        if param in list(os.environ.keys()):
            if ',' in os.environ[param]:
                global_config[param] = []
                for item in os.environ[param].split(','):
                    global_config[param].append(item)
            else:
                global_config[param] = os.environ[param]

    return global_config


def check_iperf_utility(actual_iperf_utility):
    valid_values = ["iperf", "iperf3"]
    if actual_iperf_utility not in valid_values:
        raise exceptions.InvalidConfigException(
            "The iperf utility for multiple threads test case is not correct. "
            "Valid value is one of {}. Actual value is {}. Please set the "
            "correct value in global_config.yaml:"
            "multiple_threads_iperf_utility".format(
                valid_values, actual_iperf_utility))