ansible/test/lib/ansible_test/_internal/cloud/tower.py

"""Tower plugin for integration tests."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type

import os
import time

from ..util import (
    display,
    ApplicationError,
    is_shippable,
    SubprocessError,
    ConfigParser,
)

from ..util_common import (
    run_command,
)

from . import (
    CloudProvider,
    CloudEnvironment,
    CloudEnvironmentConfig,
)

from ..core_ci import (
    AnsibleCoreCI,
)


class TowerCloudProvider(CloudProvider):
    """Tower cloud provider plugin. Sets up cloud resources before delegation."""
    def __init__(self, args):
        """
        :type args: TestConfig
        """
        super(TowerCloudProvider, self).__init__(args)

        self.aci = None
        self.version = ''

    def filter(self, targets, exclude):
        """Filter out the cloud tests when the necessary config and resources are not available.
        :type targets: tuple[TestTarget]
        :type exclude: list[str]
        """
        if os.path.isfile(self.config_static_path):
            return

        aci = get_tower_aci(self.args)

        if os.path.isfile(aci.ci_key):
            return

        if is_shippable():
            return

        super(TowerCloudProvider, self).filter(targets, exclude)

    def setup(self):
        """Setup the cloud resource before delegation and register a cleanup callback."""
        super(TowerCloudProvider, self).setup()

        if self._use_static_config():
            self._setup_static()
        else:
            self._setup_dynamic()

    def check_tower_version(self, fallback=None):
        """Check the Tower version being tested and determine the correct CLI version to use.
        :type fallback: str | None
        """
        tower_cli_version_map = {
            '3.1.5': '3.1.8',
            '3.2.3': '3.3.0',
            '3.3.5': '3.3.3',
            '3.4.3': '3.3.3',
        }

        cli_version = tower_cli_version_map.get(self.version, fallback)

        if not cli_version:
            raise ApplicationError('Mapping to ansible-tower-cli version required for Tower version: %s' % self.version)

        self._set_cloud_config('tower_cli_version', cli_version)

    def cleanup(self):
        """Clean up the cloud resource and any temporary configuration files after tests complete."""
        # cleanup on success or failure is not yet supported due to how cleanup is called
        if self.aci and self.args.remote_terminate == 'always':
            self.aci.stop()

        super(TowerCloudProvider, self).cleanup()

    def _setup_static(self):
        config = TowerConfig.parse(self.config_static_path)

        self.version = config.version
        self.check_tower_version()

    def _setup_dynamic(self):
        """Request Tower credentials through the Ansible Core CI service."""
        display.info('Provisioning %s cloud environment.' % self.platform, verbosity=1)

        # temporary solution to allow version selection
        self.version = os.environ.get('TOWER_VERSION', '3.2.3')
        self.check_tower_version(os.environ.get('TOWER_CLI_VERSION'))

        aci = get_tower_aci(self.args, self.version)
        aci.start()
        aci.wait()

        connection = aci.get()

        config = self._read_config_template()

        if not self.args.explain:
            self.aci = aci

            values = dict(
                VERSION=self.version,
                HOST=connection.hostname,
                USERNAME=connection.username,
                PASSWORD=connection.password,
            )

            config = self._populate_config_template(config, values)

        self._write_config(config)


class TowerCloudEnvironment(CloudEnvironment):
    """Tower cloud environment plugin. Updates integration test environment after delegation."""
    def setup(self):
        """Setup which should be done once per environment instead of once per test target."""
        self.setup_cli()
        self.disable_pendo()

    def setup_cli(self):
        """Install the correct Tower CLI for the version of Tower being tested."""
        tower_cli_version = self._get_cloud_config('tower_cli_version')

        display.info('Installing Tower CLI version: %s' % tower_cli_version)

        cmd = self.args.pip_command + ['install', '--disable-pip-version-check', 'ansible-tower-cli==%s' % tower_cli_version]

        run_command(self.args, cmd)

    def disable_pendo(self):
        """Disable Pendo tracking."""
        display.info('Disable Pendo tracking')

        config = TowerConfig.parse(self.config_path)

        # tower-cli does not recognize TOWER_ environment variables
        cmd = ['tower-cli', 'setting', 'modify', 'PENDO_TRACKING_STATE', 'off',
               '-h', config.host, '-u', config.username, '-p', config.password]

        attempts = 60

        while True:
            attempts -= 1

            try:
                run_command(self.args, cmd, capture=True)
                return
            except SubprocessError as ex:
                if not attempts:
                    raise ApplicationError('Timed out trying to disable Pendo tracking:\n%s' % ex)

            time.sleep(5)

    def get_environment_config(self):
        """
        :rtype: CloudEnvironmentConfig
        """
        config = TowerConfig.parse(self.config_path)

        env_vars = config.environment

        ansible_vars = dict((key.lower(), value) for key, value in env_vars.items())

        return CloudEnvironmentConfig(
            env_vars=env_vars,
            ansible_vars=ansible_vars,
        )


class TowerConfig:
    """Tower settings."""
    def __init__(self, values):
        self.version = values.get('version')
        self.host = values.get('host')
        self.username = values.get('username')
        self.password = values.get('password')

        if self.password:
            display.sensitive.add(self.password)

    @property
    def environment(self):
        """Tower settings as environment variables.
        :rtype: dict[str, str]
        """
        env = dict(
            TOWER_VERSION=self.version,
            TOWER_HOST=self.host,
            TOWER_USERNAME=self.username,
            TOWER_PASSWORD=self.password,
        )

        return env

    @staticmethod
    def parse(path):
        """
        :type path: str
        :rtype: TowerConfig
        """
        parser = ConfigParser()
        parser.read(path)

        keys = (
            'version',
            'host',
            'username',
            'password',
        )

        values = dict((k, parser.get('default', k)) for k in keys)
        config = TowerConfig(values)

        missing = [k for k in keys if not values.get(k)]

        if missing:
            raise ApplicationError('Missing or empty Tower configuration value(s): %s' % ', '.join(missing))

        return config


def get_tower_aci(args, version=None):
    """
    :type args: EnvironmentConfig
    :type version: str | None
    :rtype: AnsibleCoreCI
    """
    if version:
        persist = True
    else:
        version = ''
        persist = False

    return AnsibleCoreCI(args, 'tower', version, persist=persist, stage=args.remote_stage, provider=args.remote_provider)