fixtures/base.py

import os
import pytest
import utils
import random
import time
import logging

from utils import helpers
from utils.k8s_client import K8SClientManager
from utils import os_client


logger = logging.getLogger(__name__)

nodes = utils.get_pairs()
hw_nodes = utils.get_hw_pairs()


@pytest.fixture(scope='session', params=list(nodes.values()),
                ids=list(nodes.keys()))
def pair(request):
    return request.param


@pytest.fixture(scope='session', params=list(hw_nodes.values()),
                ids=list(hw_nodes.keys()))
def hw_pair(request):
    return request.param


@pytest.fixture(scope='session')
def openstack_clients():
    return 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'],
        endpoint_type=os.environ["OS_ENDPOINT_TYPE"],
        cert=False,
        domain=os.environ['OS_PROJECT_DOMAIN_NAME'],
        interface=os.environ["OS_ENDPOINT_TYPE"],
    )


@pytest.fixture(scope='session')
def os_resources(openstack_clients):
    logger.info("Setting up resources in admin project...")
    os_actions = os_client.OSCliActions(openstack_clients)
    os_resource = {}
    config = utils.get_configuration()
    image_name = config.get('image_name', 'cvp.ubuntu.2004')
    flavor_name = config.get('flavor_name', 'spt-test')
    flavor_ram = config.get('flavor_ram', 1536)
    flavor_vcpus = config.get('flavor_vcpus', 1)
    flavor_disk = config.get('flavor_disk', 3)
    add_hw_vif_multiqueue_enabled_to_image = config.get('add_hw_vif_multiqueue_enabled_to_image', "true")

    os_images_list = [image.id for image in
                      openstack_clients.image.images.list(
                          filters={'name': image_name})]

    if os_images_list.__len__() == 0:
        pytest.skip("No images with name {}. This name can be redefined "
                    "with 'image_name' env var ".format(image_name))

    if add_hw_vif_multiqueue_enabled_to_image.lower() == "true":
        logger.info("Adding hw_vif_multiqueue_enabled=true property to image...")
        openstack_clients.image.images.update(os_images_list[0],
            hw_vif_multiqueue_enabled = "true")

    os_resource['image_id'] = str(os_images_list[0])
    flavor_id = os_actions.get_flavor_id_by_name(flavor_name)
    if flavor_id is None:
        flavor_is_created = True
        os_resource['flavor_id'] = os_actions.create_flavor(
            flavor_name, flavor_ram, flavor_vcpus, flavor_disk).id
    else:
        flavor_is_created = False
        logger.info("Flavor {} already exists".format(flavor_name))
        os_resource['flavor_id'] = flavor_id

    os_resource['sec_group'] = os_actions.create_sec_group()
    os_resource['keypair'] = openstack_clients.compute.keypairs.create(
        '{}-{}'.format('spt-key', random.randrange(100, 999))
    )
    os_resource['net1'] = os_actions.create_network_resources()
    os_resource['ext_net'] = os_actions.get_external_network()
    if not os_resource['ext_net']:
        pytest.skip("External network not found, skipping tests. Please set "
                    "the existing floating network at the 'external_network' "
                    "option in the global_config.yaml file.")
    adm_project = os_actions.get_project_by_name("admin")
    os_resource['router'] = os_actions.create_router(
        os_resource['ext_net'], adm_project.id)
    os_resource['net2'] = os_actions.create_network(adm_project.id)
    os_resource['subnet2'] = os_actions.create_subnet(
        os_resource['net2'], adm_project.id, '10.2.7.0/24')
    for subnet in openstack_clients.network.list_subnets()['subnets']:
        if subnet['network_id'] == os_resource['net1']['id']:
            os_resource['subnet1'] = subnet['id']

    openstack_clients.network.add_interface_router(
        os_resource['router']['id'], {'subnet_id': os_resource['subnet1']})
    openstack_clients.network.add_interface_router(
        os_resource['router']['id'],
        {'subnet_id': os_resource['subnet2']['id']})
    yield os_resource

    # cleanup created resources
    logger.info("Deleting routers, networks, SG, key pair, flavor in {}..."
                "".format(adm_project.name))
    openstack_clients.network.remove_interface_router(
        os_resource['router']['id'], {'subnet_id': os_resource['subnet1']})
    openstack_clients.network.remove_interface_router(
        os_resource['router']['id'],
        {'subnet_id': os_resource['subnet2']['id']})
    openstack_clients.network.remove_gateway_router(
        os_resource['router']['id'])
    time.sleep(5)
    openstack_clients.network.delete_router(os_resource['router']['id'])
    time.sleep(10)  # extended to 10 since fails at TF sometimes
    openstack_clients.network.delete_network(os_resource['net1']['id'])
    openstack_clients.network.delete_network(os_resource['net2']['id'])

    openstack_clients.network.delete_security_group(
        os_resource['sec_group']['id'])
    openstack_clients.compute.keypairs.delete(os_resource['keypair'].name)
    if flavor_is_created:
        openstack_clients.compute.flavors.delete(os_resource['flavor_id'])


@pytest.fixture(scope='session')
def openstack_alt_clients(openstack_clients):
    # create alt project with regular admin user
    tmp_os_actions = os_client.OSCliActions(openstack_clients)
    alt_project = tmp_os_actions.create_project()
    tmp_os_actions.add_roles_to_user_in_project(
        alt_project.id, username=os.environ['OS_USERNAME'],
        domain=os.environ['OS_PROJECT_DOMAIN_NAME']
    )
    # create a client using alt project with admin user in it
    return os_client.OfficialClientManager(
        username=os.environ['OS_USERNAME'],
        password=os.environ['OS_PASSWORD'],
        project_name=alt_project.name,
        auth_url=os.environ['OS_AUTH_URL'],
        cert=False,
        domain=os.environ['OS_PROJECT_DOMAIN_NAME'],
        endpoint_type=os.environ['OS_ENDPOINT_TYPE'],
        interface=os.environ['OS_ENDPOINT_TYPE']
    )


@pytest.fixture(scope='session')
def os_resources_alt_project(openstack_alt_clients):
    logger.info("Setting up resources in the project {}..."
                "".format(openstack_alt_clients.project_name))
    alt_os_actions = os_client.OSCliActions(openstack_alt_clients)
    os_resource_alt_project = {}
    config = utils.get_configuration()
    image_name = config.get('image_name', 'Ubuntu-18.04')
    flavor_name = config.get('flavor_name', 'spt-test')
    flavor_ram = config.get('flavor_ram', 1536)
    flavor_vcpus = config.get('flavor_vcpus', 1)
    flavor_disk = config.get('flavor_disk', 3)
    add_hw_vif_multiqueue_enabled_to_image = config.get('add_hw_vif_multiqueue_enabled_to_image', "true")
    os_images_list = [image.id for image in
                      openstack_alt_clients.image.images.list(
                          filters={'name': image_name})]

    if os_images_list.__len__() == 0:
        pytest.skip("No images with name {}. This name can be redefined "
                    "with 'image_name' env var ".format(image_name))
    if add_hw_vif_multiqueue_enabled_to_image.lower() == "true":
        logger.info("Adding hw_vif_multiqueue_enabled=true property to image...")
        openstack_alt_clients.image.images.update(os_images_list[0],
            hw_vif_multiqueue_enabled = "true")
    os_resource_alt_project['image_id'] = str(os_images_list[0])

    flavor_id = alt_os_actions.get_flavor_id_by_name(flavor_name)
    if flavor_id is None:
        flavor_is_created = True
        os_resource_alt_project['flavor_id'] = alt_os_actions.create_flavor(
            flavor_name, flavor_ram, flavor_vcpus, flavor_disk).id
    else:
        flavor_is_created = False
        logger.info("Flavor {} already exists".format(flavor_name))
        os_resource_alt_project['flavor_id'] = flavor_id

    os_resource_alt_project['sec_group'] = alt_os_actions.create_sec_group()

    alt_project = alt_os_actions.get_project_by_name(
        openstack_alt_clients.project_name)
    os_resource_alt_project['net1'] = alt_os_actions.create_network_resources(
        project=alt_project.name, cidr='10.3.7.0/24')
    os_resource_alt_project['ext_net'] = alt_os_actions.get_external_network()
    if not os_resource_alt_project['ext_net']:
        pytest.skip("External network not found, skipping tests. Please set "
                    "the existing floating network at the 'external_network' "
                    "option in the global_config.yaml file.")

    os_resource_alt_project['router'] = alt_os_actions.create_router(
        os_resource_alt_project['ext_net'], alt_project.id)

    os_resource_alt_project['subnet1'] = \
        openstack_alt_clients.network.show_network(
            os_resource_alt_project['net1']['id'])['network']['subnets'][0]
    openstack_alt_clients.network.add_interface_router(
        os_resource_alt_project['router']['id'],
        {'subnet_id': os_resource_alt_project['subnet1']})
    yield os_resource_alt_project

    # cleanup created resources
    logger.info("Deleting routers, networks, SG, flavor in {} "
                "project...".format(alt_project.name))

    openstack_alt_clients.network.remove_interface_router(
        os_resource_alt_project['router']['id'],
        {'subnet_id': os_resource_alt_project['subnet1']})
    openstack_alt_clients.network.remove_gateway_router(
        os_resource_alt_project['router']['id'])
    time.sleep(5)
    openstack_alt_clients.network.delete_router(
        os_resource_alt_project['router']['id'])
    time.sleep(5)
    openstack_alt_clients.network.delete_network(
        os_resource_alt_project['net1']['id'])

    openstack_alt_clients.network.delete_security_group(
        os_resource_alt_project['sec_group']['id'])
    if flavor_is_created:
        openstack_alt_clients.compute.flavors.delete(
            os_resource_alt_project['flavor_id'])

    if alt_os_actions.is_project_empty(alt_project.id):
        openstack_alt_clients.auth.projects.delete(alt_project.id)
        logger.info("Deleted project {}".format(alt_project.name))
    else:
        logger.info("Project {} is not empty, skip deleting".format(
            alt_project.name))


@pytest.fixture(scope='module')
def k8s_v1_client():
    config = utils.get_configuration()
    logger.info("Getting the K8S config path from the global_config.yaml file.")
    k8s_config_path = config.get('mos_kubeconfig_path') or ""
    k8s_manager = K8SClientManager(k8s_config_path=k8s_config_path)
    return k8s_manager.k8s_v1


@pytest.fixture(scope="session")
def html_report():
    yield
    helpers.convert_csvs_to_single_html_report()