ansible-collections/community.general
1
0
Fork 0
mirror of https://github.com/ansible-collections/community.general.git synced 2025-05-02 15:21:25 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Add support for cloud tests to ansible-test. (#24315)

Browse source 
* Split out ansible-test docker functions.
* Add cloud support to ansible-test.
This commit is contained in:
Matt Clay 2017-05-05 16:23:00 +08:00 • committed by GitHub
parent 986765312f
commit a07d42e16d
19 changed files with 1059 additions and 135 deletions
Download patch file Download diff file Expand all files Collapse all files

356
test/runner/lib/cloud/__init__.py Normal file
View file

@ -0,0 +1,356 @@
"""Plugin system for cloud providers and environments for use in integration tests."""
from __future__ import absolute_import, print_function
import abc
import atexit
import os
import platform
import random
import re
import tempfile
from lib.util import (
ApplicationError,
display,
is_shippable,
import_plugins,
load_plugins,
)
from lib.test import (
TestConfig,
)
from lib.target import (
TestTarget,
)
PROVIDERS = {}
ENVIRONMENTS = {}
def initialize_cloud_plugins():
"""Import cloud plugins and load them into the plugin dictionaries."""
import_plugins('cloud')
load_plugins(CloudProvider, PROVIDERS)
load_plugins(CloudEnvironment, ENVIRONMENTS)
def get_cloud_platforms(args, targets=None):
"""
:type args: TestConfig
:type targets: tuple[IntegrationTarget] | None
:rtype: list[str]
"""
if targets is None:
cloud_platforms = set(args.metadata.cloud_config or [])
else:
cloud_platforms = set(get_cloud_platform(t) for t in targets)
cloud_platforms.discard(None)
return sorted(cloud_platforms)
def get_cloud_platform(target):
"""
:type target: IntegrationTarget
:rtype: str | None
"""
cloud_platforms = set(a.split('/')[1] for a in target.aliases if a.startswith('cloud/') and a.endswith('/') and a != 'cloud/')
if not cloud_platforms:
return None
if len(cloud_platforms) == 1:
cloud_platform = cloud_platforms.pop()
if cloud_platform not in PROVIDERS:
raise ApplicationError('Target %s aliases contains unknown cloud platform: %s' % (target.name, cloud_platform))
return cloud_platform
raise ApplicationError('Target %s aliases contains multiple cloud platforms: %s' % (target.name, ', '.join(sorted(cloud_platforms))))
def get_cloud_providers(args, targets=None):
"""
:type args: TestConfig
:type targets: tuple[IntegrationTarget] | None
:rtype: list[CloudProvider]
"""
return [PROVIDERS[p](args) for p in get_cloud_platforms(args, targets)]
def get_cloud_environment(args, target):
"""
:type args: TestConfig
:type target: IntegrationTarget
:rtype: CloudEnvironment
"""
cloud_platform = get_cloud_platform(target)
if not cloud_platform:
return None
return ENVIRONMENTS[cloud_platform](args)
def cloud_filter(args, targets):
"""
:type args: TestConfig
:type targets: tuple[IntegrationTarget]
:return: list[str]
"""
if args.metadata.cloud_config is not None:
return [] # cloud filter already performed prior to delegation
exclude = []
for provider in get_cloud_providers(args, targets):
provider.filter(targets, exclude)
return exclude
def cloud_init(args, targets):
"""
:type args: TestConfig
:type targets: tuple[IntegrationTarget]
"""
if args.metadata.cloud_config is not None:
return # cloud configuration already established prior to delegation
args.metadata.cloud_config = {}
for provider in get_cloud_providers(args, targets):
args.metadata.cloud_config[provider.platform] = {}
provider.setup()
class CloudBase(object):
"""Base class for cloud plugins."""
__metaclass__ = abc.ABCMeta
_CONFIG_PATH = 'config_path'
_RESOURCE_PREFIX = 'resource_prefix'
_MANAGED = 'managed'
def __init__(self, args):
"""
:type args: TestConfig
"""
self.args = args
self.platform = self.__module__.split('.')[2]
@property
def config_path(self):
"""
:rtype: str
"""
return os.path.join(os.getcwd(), self._get_cloud_config(self._CONFIG_PATH))
@config_path.setter
def config_path(self, value):
"""
:type value: str
"""
self._set_cloud_config(self._CONFIG_PATH, value)
@property
def resource_prefix(self):
"""
:rtype: str
"""
return self._get_cloud_config(self._RESOURCE_PREFIX)
@resource_prefix.setter
def resource_prefix(self, value):
"""
:type value: str
"""
self._set_cloud_config(self._RESOURCE_PREFIX, value)
@property
def managed(self):
"""
:rtype: bool
"""
return self._get_cloud_config(self._MANAGED)
@managed.setter
def managed(self, value):
"""
:type value: bool
"""
self._set_cloud_config(self._MANAGED, value)
def _get_cloud_config(self, key):
"""
:type key: str
:rtype: str | int | bool
"""
return self.args.metadata.cloud_config[self.platform][key]
def _set_cloud_config(self, key, value):
"""
:type key: str
:type value: str | int | bool
"""
self.args.metadata.cloud_config[self.platform][key] = value
class CloudProvider(CloudBase):
"""Base class for cloud provider plugins. Sets up cloud resources before delegation."""
__metaclass__ = abc.ABCMeta
TEST_DIR = 'test/integration'
def __init__(self, args, config_extension='.yml'):
"""
:type args: TestConfig
:type config_extension: str
"""
super(CloudProvider, self).__init__(args)
self.remove_config = False
self.config_static_path = '%s/cloud-config-%s%s' % (self.TEST_DIR, self.platform, config_extension)
self.config_template_path = '%s.template' % self.config_static_path
self.config_extension = config_extension
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]
"""
skip = 'cloud/%s/' % self.platform
skipped = [target.name for target in targets if skip in target.aliases]
if skipped:
exclude.append(skip)
display.warning('Excluding tests marked "%s" which require config (see "%s"): %s'
% (skip.rstrip('/'), self.config_template_path, ', '.join(skipped)))
def setup(self):
"""Setup the cloud resource before delegation and register a cleanup callback."""
self.resource_prefix = self._generate_resource_prefix()
atexit.register(self.cleanup)
# pylint: disable=locally-disabled, no-self-use
def get_remote_ssh_options(self):
"""Get any additional options needed when delegating tests to a remote instance via SSH.
:rtype: list[str]
"""
return []
# pylint: disable=locally-disabled, no-self-use
def get_docker_run_options(self):
"""Get any additional options needed when delegating tests to a docker container.
:rtype: list[str]
"""
return []
def cleanup(self):
"""Clean up the cloud resource and any temporary configuration files after tests complete."""
if self.remove_config:
os.remove(self.config_path)
def _use_static_config(self):
"""
:rtype: bool
"""
if os.path.isfile(self.config_static_path):
display.info('Using existing %s cloud config: %s' % (self.platform, self.config_static_path), verbosity=1)
self.config_path = self.config_static_path
static = True
else:
static = False
self.managed = not static
return static
def _write_config(self, content):
"""
:type content: str
"""
prefix = '%s-' % os.path.splitext(os.path.basename(self.config_static_path))[0]
with tempfile.NamedTemporaryFile(dir=self.TEST_DIR, prefix=prefix, suffix=self.config_extension, delete=False) as config_fd:
filename = os.path.join(self.TEST_DIR, os.path.basename(config_fd.name))
self.config_path = config_fd.name
self.remove_config = True
self._set_cloud_config('config_path', filename)
display.info('>>> Config: %s\n%s' % (filename, content.strip()), verbosity=3)
config_fd.write(content.encode('utf-8'))
config_fd.flush()
def _read_config_template(self):
"""
:rtype: str
"""
with open(self.config_template_path, 'r') as template_fd:
lines = template_fd.read().splitlines()
lines = [l for l in lines if not l.startswith('#')]
config = '\n'.join(lines).strip() + '\n'
return config
@staticmethod
def _populate_config_template(template, values):
"""
:type template: str
:type values: dict[str, str]
:rtype: str
"""
for key in sorted(values):
value = values[key]
template = template.replace('@%s' % key, value)
return template
@staticmethod
def _generate_resource_prefix():
"""
:rtype: str
"""
if is_shippable():
return 'shippable-%d-%d' % (
os.environ['SHIPPABLE_BUILD_NUMBER'],
os.environ['SHIPPABLE_JOB_NUMBER'],
)
node = re.sub(r'[^a-zA-Z0-9]+', '-', platform.node().split('.')[0])
return 'ansible-test-%s-%d' % (node, random.randint(10000000, 99999999))
class CloudEnvironment(CloudBase):
"""Base class for cloud environment plugins. Updates integration test environment after delegation."""
__metaclass__ = abc.ABCMeta
@abc.abstractmethod
def configure_environment(self, env, cmd):
"""
:type env: dict[str, str]
:type cmd: list[str]
"""
pass
def on_failure(self, target, tries):
"""
:type target: TestTarget
:type tries: int
"""
pass
@property
def inventory_hosts(self):
"""
:rtype: str | None
"""
return None

111
test/runner/lib/cloud/aws.py Normal file
View file

@ -0,0 +1,111 @@
"""AWS plugin for integration tests."""
from __future__ import absolute_import, print_function
import os
from lib.util import (
ApplicationError,
display,
is_shippable,
)
from lib.cloud import (
CloudProvider,
CloudEnvironment,
)
from lib.core_ci import (
AnsibleCoreCI,
)
class AwsCloudProvider(CloudProvider):
"""AWS cloud provider plugin. Sets up cloud resources before delegation."""
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 = self._create_ansible_core_ci()
if os.path.isfile(aci.ci_key):
return
if is_shippable():
return
super(AwsCloudProvider, self).filter(targets, exclude)
def setup(self):
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(AwsCloudProvider, self).setup()
aws_config_path = os.path.expanduser('~/.aws')
if os.path.exists(aws_config_path) and not self.args.docker and not self.args.remote:
raise ApplicationError('Rename "%s" or use the --docker or --remote option to isolate tests.' % aws_config_path)
if not self._use_static_config():
self._setup_dynamic()
def _setup_dynamic(self):
"""Request AWS credentials through the Ansible Core CI service."""
display.info('Provisioning %s cloud environment.' % self.platform, verbosity=1)
config = self._read_config_template()
aci = self._create_ansible_core_ci()
response = aci.start_remote()
if not self.args.explain:
credentials = response['aws']['credentials']
values = dict(
ACCESS_KEY=credentials['access_key'],
SECRET_KEY=credentials['secret_key'],
SECURITY_TOKEN=credentials['session_token'],
)
config = self._populate_config_template(config, values)
self._write_config(config)
def _create_ansible_core_ci(self):
"""
:rtype: AnsibleCoreCI
"""
return AnsibleCoreCI(self.args, 'aws', 'sts', persist=False, stage=self.args.remote_stage)
class AwsCloudEnvironment(CloudEnvironment):
"""AWS cloud environment plugin. Updates integration test environment after delegation."""
def configure_environment(self, env, cmd):
"""
:type env: dict[str, str]
:type cmd: list[str]
"""
cmd.append('-e')
cmd.append('@%s' % self.config_path)
cmd.append('-e')
cmd.append('resource_prefix=%s' % self.resource_prefix)
def on_failure(self, target, tries):
"""
:type target: TestTarget
:type tries: int
"""
if not tries and self.managed:
display.notice('If %s failed due to permissions, the IAM test policy may need to be updated. '
'For help, consult @mattclay or @gundalow on GitHub or #ansible-devel on IRC.' % target.name)
@property
def inventory_hosts(self):
"""
:rtype: str | None
"""
return 'amazon'

240
test/runner/lib/cloud/cs.py Normal file
View file

@ -0,0 +1,240 @@
"""CloudStack plugin for integration tests."""
from __future__ import absolute_import, print_function
import os
import time
from lib.cloud import (
CloudProvider,
CloudEnvironment,
)
from lib.util import (
find_executable,
ApplicationError,
display,
SubprocessError,
is_shippable,
)
from lib.http import (
HttpClient,
urlparse,
)
from lib.docker_util import (
docker_run,
docker_rm,
docker_inspect,
docker_pull,
)
try:
# noinspection PyPep8Naming
import ConfigParser as configparser
except ImportError:
# noinspection PyUnresolvedReferences
import configparser
class CsCloudProvider(CloudProvider):
"""CloudStack cloud provider plugin. Sets up cloud resources before delegation."""
DOCKER_SIMULATOR_NAME = 'cloudstack-sim'
def __init__(self, args):
"""
:type args: TestConfig
"""
super(CsCloudProvider, self).__init__(args, config_extension='.ini')
self.image = 'resmo/cloudstack-sim'
self.container_name = ''
self.endpoint = ''
self.host = ''
self.port = 0
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
docker = find_executable('docker')
if docker:
return
super(CsCloudProvider, self).filter(targets, exclude)
def setup(self):
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(CsCloudProvider, self).setup()
if self._use_static_config():
self._setup_static()
else:
self._setup_dynamic()
def get_remote_ssh_options(self):
"""Get any additional options needed when delegating tests to a remote instance via SSH.
:rtype: list[str]
"""
if self.managed:
return ['-R', '8888:localhost:8888']
return []
def get_docker_run_options(self):
"""Get any additional options needed when delegating tests to a docker container.
:rtype: list[str]
"""
if self.managed:
return ['--link', self.DOCKER_SIMULATOR_NAME]
return []
def cleanup(self):
"""Clean up the cloud resource and any temporary configuration files after tests complete."""
if self.container_name:
if is_shippable():
docker_rm(self.args, self.container_name)
elif not self.args.explain:
display.notice('Remember to run `docker rm -f %s` when finished testing.' % self.container_name)
super(CsCloudProvider, self).cleanup()
def _setup_static(self):
"""Configure CloudStack tests for use with static configuration."""
parser = configparser.RawConfigParser()
parser.read(self.config_static_path)
self.endpoint = parser.get('cloudstack', 'endpoint')
parts = urlparse(self.endpoint)
self.host = parts.hostname
if not self.host:
raise ApplicationError('Could not determine host from endpoint: %s' % self.endpoint)
if parts.port:
self.port = parts.port
elif parts.scheme == 'http':
self.port = 80
elif parts.scheme == 'https':
self.port = 443
else:
raise ApplicationError('Could not determine port from endpoint: %s' % self.endpoint)
display.info('Read cs host "%s" and port %d from config: %s' % (self.host, self.port, self.config_static_path), verbosity=1)
self._wait_for_service()
def _setup_dynamic(self):
"""Create a CloudStack simulator using docker."""
config = self._read_config_template()
self.container_name = self.DOCKER_SIMULATOR_NAME
results = docker_inspect(self.args, self.container_name)
if results and not results[0]['State']['Running']:
docker_rm(self.args, self.container_name)
results = []
if results:
display.info('Using the existing CloudStack simulator docker container.', verbosity=1)
else:
display.info('Starting a new CloudStack simulator docker container.', verbosity=1)
docker_pull(self.args, self.image)
docker_run(self.args, self.image, ['-d', '-p', '8888:8888', '--name', self.container_name])
display.notice('The CloudStack simulator will probably be ready in 5 - 10 minutes.')
self.host = 'localhost'
self.port = 8888
self.endpoint = 'http://%s:%d' % (self.host, self.port)
self._wait_for_service()
if self.args.explain:
values = dict(
HOST=self.host,
PORT=str(self.port),
)
else:
credentials = self._get_credentials()
if self.args.docker:
host = self.DOCKER_SIMULATOR_NAME
else:
host = self.host
values = dict(
HOST=host,
PORT=str(self.port),
KEY=credentials['apikey'],
SECRET=credentials['secretkey'],
)
config = self._populate_config_template(config, values)
self._write_config(config)
def _wait_for_service(self):
"""Wait for the CloudStack service endpoint to accept connections."""
if self.args.explain:
return
client = HttpClient(self.args, always=True)
endpoint = self.endpoint
for _ in range(1, 90):
display.info('Waiting for CloudStack service: %s' % endpoint, verbosity=1)
try:
client.get(endpoint)
return
except SubprocessError:
pass
time.sleep(10)
raise ApplicationError('Timeout waiting for CloudStack service.')
def _get_credentials(self):
"""Wait for the CloudStack simulator to return credentials.
:rtype: dict[str, str]
"""
client = HttpClient(self.args, always=True)
endpoint = '%s/admin.json' % self.endpoint
for _ in range(1, 90):
display.info('Waiting for CloudStack credentials: %s' % endpoint, verbosity=1)
response = client.get(endpoint)
if response.status_code == 200:
return response.json()
time.sleep(10)
raise ApplicationError('Timeout waiting for CloudStack credentials.')
class CsCloudEnvironment(CloudEnvironment):
"""CloudStack cloud environment plugin. Updates integration test environment after delegation."""
def configure_environment(self, env, cmd):
"""
:type env: dict[str, str]
:type cmd: list[str]
"""
changes = dict(
CLOUDSTACK_CONFIG=self.config_path,
)
env.update(changes)
cmd.append('-e')
cmd.append('cs_resource_prefix=%s' % self.resource_prefix)