cloudbridge/tests/helpers/__init__.py

import functools
import operator
import os
import sys
import unittest
import uuid

import tenacity

from cloudbridge.base import helpers as cb_helpers
from cloudbridge.factory import CloudProviderFactory
from cloudbridge.interfaces import CloudProvider
from cloudbridge.interfaces import InstanceState
from cloudbridge.interfaces import TestMockHelperMixin
from cloudbridge.interfaces.resources import FloatingIpState
from cloudbridge.interfaces.resources import NetworkState
from cloudbridge.interfaces.resources import SubnetState


def parse_bool(val):
    if val:
        return str(val).upper() in ['TRUE', 'YES']
    else:
        return False


def skipIfNoService(services):
    """
    A decorator for skipping tests if the provider
    does not implement a given service.
    """
    def wrap(func):
        """
        The actual wrapper
        """
        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            provider = getattr(self, 'provider')
            if provider:
                for service in services:
                    if not provider.has_service(service):
                        self.skipTest("Skipping test because '%s' service is"
                                      " not implemented" % (service,))
            func(self, *args, **kwargs)
        return wrapper
    return wrap


def skipIfPython(op, major, minor):
    """
    A decorator for skipping tests if the python
    version doesn't match
    """
    def stringToOperator(op):
        op_map = {
            "=": operator.eq,
            "==": operator.eq,
            "<": operator.lt,
            "<=": operator.le,
            ">": operator.gt,
            ">=": operator.ge,
        }
        return op_map.get(op)

    def wrap(func):
        """
        The actual wrapper
        """
        @functools.wraps(func)
        def wrapper(self, *args, **kwargs):
            op_func = stringToOperator(op)
            if op_func(sys.version_info, (major, minor)):
                self.skipTest(
                    "Skipping test because python version {0} is {1} expected"
                    " version {2}".format(sys.version_info[:2],
                                          op, (major, minor)))
            func(self, *args, **kwargs)
        return wrapper
    return wrap


def env_or(varname, default_value):
    """
    Environment variable ``varname``, treating unset and empty alike.

    The cloud workflow sets every CB_* variable in every matrix cell, as
    ``${{ matrix.cloud-provider == '<x>' && secrets.<VAR> || '' }}``, so a
    variable belonging to another cell - or one whose secret is simply not
    configured - arrives as an empty string rather than absent. A plain
    ``os.environ.get`` hands that empty string straight back, silently
    blanking the test data instead of falling back to the default.
    """
    return cb_helpers.get_env(varname) or default_value


TEST_DATA_CONFIG = {
    "AWSCloudProvider": {
        # Canonical's Ubuntu 16.04, built 2017 - a Xen-era HVM image, and an
        # AMI id is per-region, so anything but us-east-1 must set
        # CB_IMAGE_AWS. Kept deliberately, not by inertia:
        #
        # The comment that used to sit here said moto needed this value to
        # match an entry in tests/fixtures/custom_amis.json. That has not been
        # true since 3fbcba2 removed MOTO_AMIS_PATH from tox.ini - nothing
        # loads that fixture, and moto does not validate instance-launch AMI
        # ids, so the mock provider is indifferent to this value.
        #
        # With that constraint gone the obvious move was a current image, on
        # the theory that the suite's wall time - dominated by launching an
        # instance, snapshotting it into an AMI, launching a second instance
        # from it, and stop/start cycling - is paying a Xen-era penalty.
        # Measured against Ubuntu 24.04 (ami-052355af2a014bd2c), it got
        # worse: test_create_and_list_image went 26.2 -> 33.8 min and
        # test_instance_start_stop_methods 20.3 -> 29.1 min. Plausibly the
        # larger image costs more to snapshot while CB_VM_TYPE_AWS stays on a
        # pre-Nitro type, so there is no launch saving to offset it. Worth
        # revisiting only alongside a Nitro instance type, and note the same
        # test has measured 26.2, 60.6 and 33.8 min across three runs, so a
        # single run cannot resolve a difference this size either way.
        "image": env_or('CB_IMAGE_AWS', 'ami-aa2ea6d0'),
        "vm_type": env_or('CB_VM_TYPE_AWS', 't2.nano'),
        "placement": env_or('CB_PLACEMENT_AWS', 'us-east-1a'),
        "placement_cfg_key": "aws_zone_name"
    },
    'OpenStackCloudProvider': {
        'image': env_or('CB_IMAGE_OS',
                        'c66bdfa1-62b1-43be-8964-e9ce208ac6a5'),
        "vm_type": env_or('CB_VM_TYPE_OS', 'm1.tiny'),
        "placement": env_or('CB_PLACEMENT_OS', 'nova'),
        "placement_cfg_key": "os_zone_name"
    },
    'GCPCloudProvider': {
        'image': env_or(
            'CB_IMAGE_GCP',
            'https://www.googleapis.com/compute/v1/projects/ubuntu-os-cloud/'
            'global/images/ubuntu-1804-bionic-v20200908'),
        'vm_type': env_or('CB_VM_TYPE_GCP', 'f1-micro'),
        'placement': env_or('GCP_ZONE_NAME', 'us-central1-a'),
        "placement_cfg_key": "gcp_zone_name"
    },
    "AzureCloudProvider": {
        "image":
            env_or('CB_IMAGE_AZURE',
                   'Canonical:0001-com-ubuntu-minimal-jammy:minimal-22_04-lts-gen2:latest'),
        "vm_type": env_or('CB_VM_TYPE_AZURE', 'Standard_DC1ds_v3'),
        "placement": env_or('CB_PLACEMENT_AZURE', 'eastus'),
        "placement_cfg_key": "azure_zone_name"
    }
}


def get_provider_test_data(provider, key):
    provider_id = (provider.PROVIDER_ID if isinstance(provider, CloudProvider)
                   else provider)
    if "aws" == provider_id:
        return TEST_DATA_CONFIG.get("AWSCloudProvider").get(key)
    if "mock" == provider_id:
        return TEST_DATA_CONFIG.get("AWSCloudProvider").get(key)
    elif "openstack" == provider_id:
        return TEST_DATA_CONFIG.get("OpenStackCloudProvider").get(key)
    elif "gcp" == provider_id:
        return TEST_DATA_CONFIG.get("GCPCloudProvider").get(key)
    elif "azure" == provider_id:
        return TEST_DATA_CONFIG.get("AzureCloudProvider").get(key)
    return None


@tenacity.retry(
    stop=tenacity.stop_after_attempt(10),
    wait=tenacity.wait_fixed(5),
    reraise=True,
)
def get_or_create_default_subnet(provider):
    """
    Return the default subnet to be used for tests.

    Multiple pytest-xdist workers can call this concurrently at cold-start
    and race on the singleton default vnet/subnet (on Azure ARM that shows
    up as AnotherOperationInProgress / cancelled-and-superseded / subnet
    IP-range overlap). Retry with a short backoff so the loser re-runs the
    find() and inherits whatever the winner just created.
    """
    return provider.networking.subnets.get_or_create_default()


def cleanup_subnet(subnet):
    if subnet:
        subnet.delete()
        subnet.wait_for([SubnetState.UNKNOWN],
                        terminal_states=[SubnetState.ERROR])


def cleanup_network(network):
    """
    Delete the supplied network, first deleting any contained subnets.
    """
    if network:
        try:
            for sn in network.subnets:
                with cb_helpers.cleanup_action(lambda: cleanup_subnet(sn)):
                    pass
        finally:
            network.delete()
            network.wait_for([NetworkState.UNKNOWN],
                             terminal_states=[NetworkState.ERROR])


def cleanup_fip(fip):
    if fip:
        fip.delete()
        fip.wait_for([FloatingIpState.UNKNOWN],
                     terminal_states=[FloatingIpState.ERROR])


def get_test_gateway(provider):
    """
    Get an internet gateway for testing.

    This includes creating a network for the gateway, which is also returned.
    """
    sn = get_or_create_default_subnet(provider)
    net = sn.network
    return net.gateways.get_or_create()


def cleanup_gateway(gateway):
    """
    Delete the supplied network and gateway.
    """
    with cb_helpers.cleanup_action(lambda: gateway.delete()):
        pass


def create_test_instance(
        provider, instance_label, subnet, launch_config=None,
        key_pair=None, vm_firewalls=None, user_data=None):

    instance = provider.compute.instances.create(
        instance_label, get_provider_test_data(provider, 'image'),
        get_provider_test_data(provider, 'vm_type'),
        subnet=subnet,
        key_pair=key_pair,
        vm_firewalls=vm_firewalls,
        launch_config=launch_config,
        user_data=user_data)

    return instance


def get_test_instance(provider, label, key_pair=None, vm_firewalls=None,
                      subnet=None, user_data=None):
    launch_config = None
    instance = create_test_instance(
        provider,
        label,
        subnet=subnet,
        key_pair=key_pair,
        vm_firewalls=vm_firewalls,
        launch_config=launch_config,
        user_data=user_data)
    instance.wait_till_ready()
    return instance


def get_test_fixtures_folder():
    return os.path.join(os.path.dirname(__file__), '../fixtures/')


def delete_instance(instance):
    if instance:
        instance.delete()
        instance.wait_for([InstanceState.DELETED, InstanceState.UNKNOWN],
                          terminal_states=[InstanceState.ERROR])


def cleanup_test_resources(instance=None, vm_firewall=None,
                           key_pair=None, network=None):
    """Clean up any combination of supplied resources."""
    with cb_helpers.cleanup_action(
            lambda: cleanup_network(network) if network else None):
        with cb_helpers.cleanup_action(
                lambda: key_pair.delete() if key_pair else None):
            with cb_helpers.cleanup_action(
                    lambda: vm_firewall.delete() if vm_firewall else None):
                delete_instance(instance)


def get_uuid():
    return str(uuid.uuid4())[:6]


class ProviderTestBase(unittest.TestCase):

    _provider = None

    def setUp(self):
        if isinstance(self.provider, TestMockHelperMixin):
            self.provider.setUpMock()

    def tearDown(self):
        if isinstance(self.provider, TestMockHelperMixin):
            self.provider.tearDownMock()
        self._provider = None

    def get_provider_wait_interval(self, provider_class):
        if issubclass(provider_class, TestMockHelperMixin):
            return 0
        else:
            return 1

    def create_provider_instance(self):
        provider_name = cb_helpers.get_env("CB_TEST_PROVIDER", "aws")
        zone_cfg_key = get_provider_test_data(provider_name,
                                              'placement_cfg_key')
        factory = CloudProviderFactory()
        provider_class = factory.get_provider_class(provider_name)
        config = {
            'default_wait_interval': self.get_provider_wait_interval(
                provider_class),
            'default_result_limit': 5,
            zone_cfg_key: get_provider_test_data(provider_name, 'placement')
        }
        return provider_class(config)

    @property
    def provider(self):
        if not self._provider:
            self._provider = self.create_provider_instance()
        return self._provider