cmueller/ansible
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Clean up ansible-test cloud plugins. (#74322)

Browse source 
- Improve code reuse.
- Add missing type hints, fix existing ones and convert them to PEP 484 style.
- Add missing imports and clean up existing ones.
- Add missing docstrings and clean up existing ones.
This commit is contained in:
Matt Clay 2021-04-16 18:49:22 -07:00 committed by GitHub
parent 17237c1d88
commit 277a06f641
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
18 changed files with 363 additions and 509 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
changelogs/fragments/ansible-test-cloud-plugin-code-cleanup.yml Normal file
View file

@ -0,0 +1,2 @@
minor_changes:
- ansible-test - Clean up code in the cloud plugins.

200
test/lib/ansible_test/_internal/cloud/__init__.py
View file

@ -5,10 +5,10 @@ __metaclass__ = type
import abc
import atexit
import datetime
import time
import os
import re
import tempfile
import time
from .. import types as t
@ -21,25 +21,26 @@ from ..io import (
)
from ..util import (
ABC,
ANSIBLE_TEST_CONFIG_ROOT,
ApplicationError,
display,
import_plugins,
load_plugins,
ABC,
ANSIBLE_TEST_CONFIG_ROOT,
)
from ..util_common import (
write_json_test_results,
ResultType,
write_json_test_results,
)
from ..target import (
TestTarget,
IntegrationTarget,
)
from ..config import (
IntegrationConfig,
TestConfig,
)
from ..ci import (
@ -58,7 +59,7 @@ PROVIDERS = {}
ENVIRONMENTS = {}
def initialize_cloud_plugins():
def initialize_cloud_plugins(): # type: () -> None
"""Import cloud plugins and load them into the plugin dictionaries."""
import_plugins('cloud')
@ -66,12 +67,8 @@ def initialize_cloud_plugins():
load_plugins(CloudEnvironment, ENVIRONMENTS)
def get_cloud_platforms(args, targets=None):
"""
:type args: TestConfig
:type targets: tuple[IntegrationTarget] | None
:rtype: list[str]
"""
def get_cloud_platforms(args, targets=None): # type: (TestConfig, t.Optional[t.Tuple[IntegrationTarget, ...]]) -> t.List[str]
"""Return cloud platform names for the specified targets."""
if isinstance(args, IntegrationConfig):
if args.list_targets:
return []
@ -86,11 +83,8 @@ def get_cloud_platforms(args, targets=None):
return sorted(cloud_platforms)
def get_cloud_platform(target):
"""
:type target: IntegrationTarget
:rtype: str | None
"""
def get_cloud_platform(target): # type: (IntegrationTarget) -> t.Optional[str]
"""Return the name of the cloud platform used for the given target, or None if no cloud platform is used."""
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:
@ -107,21 +101,13 @@ def get_cloud_platform(target):
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: IntegrationConfig
:type targets: tuple[IntegrationTarget] | None
:rtype: list[CloudProvider]
"""
def get_cloud_providers(args, targets=None): # type: (IntegrationConfig, t.Optional[t.Tuple[IntegrationTarget, ...]]) -> t.List[CloudProvider]
"""Return a list of cloud providers for the given targets."""
return [PROVIDERS[p](args) for p in get_cloud_platforms(args, targets)]
def get_cloud_environment(args, target):
"""
:type args: IntegrationConfig
:type target: IntegrationTarget
:rtype: CloudEnvironment
"""
def get_cloud_environment(args, target): # type: (IntegrationConfig, IntegrationTarget) -> t.Optional[CloudEnvironment]
"""Return the cloud environment for the given target, or None if no cloud environment is used for the target."""
cloud_platform = get_cloud_platform(target)
if not cloud_platform:
@ -130,12 +116,8 @@ def get_cloud_environment(args, target):
return ENVIRONMENTS[cloud_platform](args)
def cloud_filter(args, targets):
"""
:type args: IntegrationConfig
:type targets: tuple[IntegrationTarget]
:return: list[str]
"""
def cloud_filter(args, targets): # type: (IntegrationConfig, t.Tuple[IntegrationTarget, ...]) -> t.List[str]
"""Return a list of target names to exclude based on the given targets."""
if args.metadata.cloud_config is not None:
return [] # cloud filter already performed prior to delegation
@ -147,11 +129,8 @@ def cloud_filter(args, targets):
return exclude
def cloud_init(args, targets):
"""
:type args: IntegrationConfig
:type targets: tuple[IntegrationTarget]
"""
def cloud_init(args, targets): # type: (IntegrationConfig, t.Tuple[IntegrationTarget, ...]) -> None
"""Initialize cloud plugins for the given targets."""
if args.metadata.cloud_config is not None:
return # cloud configuration already established prior to delegation
@ -192,10 +171,7 @@ class CloudBase(ABC):
_MANAGED = 'managed'
_SETUP_EXECUTED = 'setup_executed'
def __init__(self, args):
"""
:type args: IntegrationConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
self.args = args
self.platform = self.__module__.split('.')[-1]
@ -214,87 +190,60 @@ class CloudBase(ABC):
data_context().register_payload_callback(config_callback)
@property
def setup_executed(self):
"""
:rtype: bool
"""
def setup_executed(self): # type: () -> bool
"""True if setup has been executed, otherwise False."""
return self._get_cloud_config(self._SETUP_EXECUTED, False)
@setup_executed.setter
def setup_executed(self, value):
"""
:type value: bool
"""
def setup_executed(self, value): # type: (bool) -> None
"""True if setup has been executed, otherwise False."""
self._set_cloud_config(self._SETUP_EXECUTED, value)
@property
def config_path(self):
"""
:rtype: str
"""
def config_path(self): # type: () -> str
"""Path to the configuration file."""
return os.path.join(data_context().content.root, self._get_cloud_config(self._CONFIG_PATH))
@config_path.setter
def config_path(self, value):
"""
:type value: str
"""
def config_path(self, value): # type: (str) -> None
"""Path to the configuration file."""
self._set_cloud_config(self._CONFIG_PATH, value)
@property
def resource_prefix(self):
"""
:rtype: str
"""
def resource_prefix(self): # type: () -> str
"""Resource prefix."""
return self._get_cloud_config(self._RESOURCE_PREFIX)
@resource_prefix.setter
def resource_prefix(self, value):
"""
:type value: str
"""
def resource_prefix(self, value): # type: (str) -> None
"""Resource prefix."""
self._set_cloud_config(self._RESOURCE_PREFIX, value)
@property
def managed(self):
"""
:rtype: bool
"""
def managed(self): # type: () -> bool
"""True if resources are managed by ansible-test, otherwise False."""
return self._get_cloud_config(self._MANAGED)
@managed.setter
def managed(self, value):
"""
:type value: bool
"""
def managed(self, value): # type: (bool) -> None
"""True if resources are managed by ansible-test, otherwise False."""
self._set_cloud_config(self._MANAGED, value)
def _get_cloud_config(self, key, default=None):
"""
:type key: str
:type default: str | int | bool | None
:rtype: str | int | bool
"""
def _get_cloud_config(self, key, default=None): # type: (str, t.Optional[t.Union[str, int, bool]]) -> t.Union[str, int, bool]
"""Return the specified value from the internal configuration."""
if default is not None:
return self.args.metadata.cloud_config[self.platform].get(key, default)
return self.args.metadata.cloud_config[self.platform][key]
def _set_cloud_config(self, key, value):
"""
:type key: str
:type value: str | int | bool
"""
def _set_cloud_config(self, key, value): # type: (str, t.Union[str, int, bool]) -> None
"""Set the specified key and value in the internal configuration."""
self.args.metadata.cloud_config[self.platform][key] = value
class CloudProvider(CloudBase):
"""Base class for cloud provider plugins. Sets up cloud resources before delegation."""
def __init__(self, args, config_extension='.ini'):
"""
:type args: IntegrationConfig
:type config_extension: str
"""
def __init__(self, args, config_extension='.ini'): # type: (IntegrationConfig, str) -> None
super(CloudProvider, self).__init__(args)
self.ci_provider = get_ci_provider()
@ -307,11 +256,8 @@ class CloudProvider(CloudBase):
self.uses_config = False
self.uses_docker = False
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]
"""
def filter(self, targets, exclude): # type: (t.Tuple[IntegrationTarget, ...], t.List[str]) -> None
"""Filter out the cloud tests when the necessary config and resources are not available."""
if not self.uses_docker and not self.uses_config:
return
@ -337,22 +283,20 @@ class CloudProvider(CloudBase):
display.warning('Excluding tests marked "%s" which requires container support or config (see "%s"): %s'
% (skip.rstrip('/'), self.config_template_path, ', '.join(skipped)))
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
self.resource_prefix = self.ci_provider.generate_resource_prefix()
self.resource_prefix = re.sub(r'[^a-zA-Z0-9]+', '-', self.resource_prefix)[:63].lower().rstrip('-')
atexit.register(self.cleanup)
def cleanup(self):
def cleanup(self): # type: () -> None
"""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
"""
def _use_static_config(self): # type: () -> bool
"""Use a static config file if available. Returns True if static config is used, otherwise returns False."""
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
@ -364,10 +308,8 @@ class CloudProvider(CloudBase):
return static
def _write_config(self, content):
"""
:type content: str
"""
def _write_config(self, content): # type: (t.Text) -> None
"""Write the given content to the config file."""
prefix = '%s-' % os.path.splitext(os.path.basename(self.config_static_path))[0]
with tempfile.NamedTemporaryFile(dir=data_context().content.integration_path, prefix=prefix, suffix=self.config_extension, delete=False) as config_fd:
@ -381,22 +323,16 @@ class CloudProvider(CloudBase):
config_fd.write(to_bytes(content))
config_fd.flush()
def _read_config_template(self):
"""
:rtype: str
"""
def _read_config_template(self): # type: () -> t.Text
"""Read and return the configuration template."""
lines = read_text_file(self.config_template_path).splitlines()
lines = [line for line in lines if not line.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
"""
def _populate_config_template(template, values): # type: (t.Text, t.Dict[str, str]) -> t.Text
"""Populate and return the given template with the provided values."""
for key in sorted(values):
value = values[key]
template = template.replace('@%s' % key, value)
@ -406,7 +342,7 @@ class CloudProvider(CloudBase):
class CloudEnvironment(CloudBase):
"""Base class for cloud environment plugins. Updates integration test environment after delegation."""
def setup_once(self):
def setup_once(self): # type: () -> None
"""Run setup if it has not already been run."""
if self.setup_executed:
return
@ -414,31 +350,25 @@ class CloudEnvironment(CloudBase):
self.setup()
self.setup_executed = True
def setup(self):
def setup(self): # type: () -> None
"""Setup which should be done once per environment instead of once per test target."""
@abc.abstractmethod
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
def on_failure(self, target, tries):
"""
:type target: IntegrationTarget
:type tries: int
"""
def on_failure(self, target, tries): # type: (IntegrationTarget, int) -> None
"""Callback to run when an integration target fails."""
class CloudEnvironmentConfig:
"""Configuration for the environment."""
def __init__(self, env_vars=None, ansible_vars=None, module_defaults=None, callback_plugins=None):
"""
:type env_vars: dict[str, str] | None
:type ansible_vars: dict[str, any] | None
:type module_defaults: dict[str, dict[str, any]] | None
:type callback_plugins: list[str] | None
"""
def __init__(self,
env_vars=None, # type: t.Optional[t.Dict[str, str]]
ansible_vars=None, # type: t.Optional[t.Dict[str, t.Any]]
module_defaults=None, # type: t.Optional[t.Dict[str, t.Dict[str, t.Any]]]
callback_plugins=None, # type: t.Optional[t.List[str]]
):
self.env_vars = env_vars
self.ansible_vars = ansible_vars
self.module_defaults = module_defaults

29
test/lib/ansible_test/_internal/cloud/acme.py
View file

@ -4,25 +4,26 @@ __metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..containers import (
run_support_container,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class ACMEProvider(CloudProvider):
"""ACME plugin. Sets up cloud resources for tests."""
DOCKER_SIMULATOR_NAME = 'acme-simulator'
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(ACMEProvider, self).__init__(args)
# The simulator must be pinned to a specific version to guarantee CI passes with the version used.
@ -33,7 +34,7 @@ class ACMEProvider(CloudProvider):
self.uses_docker = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(ACMEProvider, self).setup()
@ -42,7 +43,7 @@ class ACMEProvider(CloudProvider):
else:
self._setup_dynamic()
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Create a ACME test container using docker."""
ports = [
5000, # control port for flask app in container
@ -63,16 +64,14 @@ class ACMEProvider(CloudProvider):
self._set_cloud_config('acme_host', self.DOCKER_SIMULATOR_NAME)
def _setup_static(self):
def _setup_static(self): # type: () -> None
raise NotImplementedError()
class ACMEEnvironment(CloudEnvironment):
"""ACME environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
ansible_vars = dict(
acme_host=self._get_cloud_config('acme_host'),
)

55
test/lib/ansible_test/_internal/cloud/aws.py
View file

@ -4,38 +4,42 @@ __metaclass__ = type
import os
from .. import types as t
from ..util import (
ApplicationError,
display,
ConfigParser,
display,
)
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..target import (
IntegrationTarget,
)
from ..core_ci import (
AnsibleCoreCI,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class AwsCloudProvider(CloudProvider):
"""AWS cloud provider plugin. Sets up cloud resources before delegation."""
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(AwsCloudProvider, self).__init__(args)
self.uses_config = True
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]
"""
def filter(self, targets, exclude): # type: (t.Tuple[IntegrationTarget, ...], t.List[str]) -> None
"""Filter out the cloud tests when the necessary config and resources are not available."""
aci = self._create_ansible_core_ci()
if aci.available:
@ -43,7 +47,7 @@ class AwsCloudProvider(CloudProvider):
super(AwsCloudProvider, self).filter(targets, exclude)
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(AwsCloudProvider, self).setup()
@ -55,7 +59,7 @@ class AwsCloudProvider(CloudProvider):
if not self._use_static_config():
self._setup_dynamic()
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Request AWS credentials through the Ansible Core CI service."""
display.info('Provisioning %s cloud environment.' % self.platform, verbosity=1)
@ -82,19 +86,15 @@ class AwsCloudProvider(CloudProvider):
self._write_config(config)
def _create_ansible_core_ci(self):
"""
:rtype: AnsibleCoreCI
"""
def _create_ansible_core_ci(self): # type: () -> AnsibleCoreCI
"""Return an AWS instance of AnsibleCoreCI."""
return AnsibleCoreCI(self.args, 'aws', 'aws', persist=False, stage=self.args.remote_stage, provider='aws', internal=True)
class AwsCloudEnvironment(CloudEnvironment):
"""AWS cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)
@ -119,11 +119,8 @@ class AwsCloudEnvironment(CloudEnvironment):
callback_plugins=['aws_resource_actions'],
)
def on_failure(self, target, tries):
"""
:type target: TestTarget
:type tries: int
"""
def on_failure(self, target, tries): # type: (IntegrationTarget, int) -> None
"""Callback to run when an integration target fails."""
if not tries and self.managed:
display.notice('If %s failed due to permissions, the IAM test policy may need to be updated. '
'https://docs.ansible.com/ansible/devel/dev_guide/platforms/aws_guidelines.html#aws-permissions-for-integration-tests.'

66
test/lib/ansible_test/_internal/cloud/azure.py
View file

@ -4,53 +4,57 @@ __metaclass__ = type
import os
from .. import types as t
from ..io import (
read_text_file,
)
from ..util import (
ApplicationError,
display,
ConfigParser,
display,
)
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..target import (
IntegrationTarget,
)
from ..http import (
HttpClient,
parse_qs,
urlparse,
urlunparse,
parse_qs,
)
from ..core_ci import (
AnsibleCoreCI,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class AzureCloudProvider(CloudProvider):
"""Azure cloud provider plugin. Sets up cloud resources before delegation."""
SHERLOCK_CONFIG_PATH = os.path.expanduser('~/.ansible-sherlock-ci.cfg')
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(AzureCloudProvider, self).__init__(args)
self.aci = None
self.uses_config = True
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]
"""
def filter(self, targets, exclude): # type: (t.Tuple[IntegrationTarget, ...], t.List[str]) -> None
"""Filter out the cloud tests when the necessary config and resources are not available."""
aci = self._create_ansible_core_ci()
if aci.available:
@ -61,7 +65,7 @@ class AzureCloudProvider(CloudProvider):
super(AzureCloudProvider, self).filter(targets, exclude)
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(AzureCloudProvider, self).setup()
@ -70,14 +74,14 @@ class AzureCloudProvider(CloudProvider):
get_config(self.config_path) # check required variables
def cleanup(self):
def cleanup(self): # type: () -> None
"""Clean up the cloud resource and any temporary configuration files after tests complete."""
if self.aci:
self.aci.stop()
super(AzureCloudProvider, self).cleanup()
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Request Azure credentials through Sherlock."""
display.info('Provisioning %s cloud environment.' % self.platform, verbosity=1)
@ -131,19 +135,15 @@ class AzureCloudProvider(CloudProvider):
self._write_config(config)
def _create_ansible_core_ci(self):
"""
:rtype: AnsibleCoreCI
"""
def _create_ansible_core_ci(self): # type: () -> AnsibleCoreCI
"""Return an Azure instance of AnsibleCoreCI."""
return AnsibleCoreCI(self.args, 'azure', 'azure', persist=False, stage=self.args.remote_stage, provider='azure', internal=True)
class AzureCloudEnvironment(CloudEnvironment):
"""Azure cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
env_vars = get_config(self.config_path)
display.sensitive.add(env_vars.get('AZURE_SECRET'))
@ -160,21 +160,15 @@ class AzureCloudEnvironment(CloudEnvironment):
ansible_vars=ansible_vars,
)
def on_failure(self, target, tries):
"""
:type target: TestTarget
:type tries: int
"""
def on_failure(self, target, tries): # type: (IntegrationTarget, int) -> None
"""Callback to run when an integration target fails."""
if not tries and self.managed:
display.notice('If %s failed due to permissions, the test policy may need to be updated. '
'For help, consult @mattclay or @gundalow on GitHub or #ansible-devel on IRC.' % target.name)
def get_config(config_path):
"""
:type config_path: str
:rtype: dict[str, str]
"""
def get_config(config_path): # type: (str) -> t.Dict[str, str]
"""Return a configuration dictionary parsed from the given configuration path."""
parser = ConfigParser()
parser.read(config_path)

47
test/lib/ansible_test/_internal/cloud/cloudscale.py
View file

@ -7,49 +7,40 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..util import (
ConfigParser,
display,
)
from ..util import ConfigParser, display
from ..config import (
IntegrationConfig,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class CloudscaleCloudProvider(CloudProvider):
"""Cloudscale cloud provider plugin. Sets up cloud resources before
delegation.
"""
def __init__(self, args):
"""
:type args: TestConfig
"""
"""Cloudscale cloud provider plugin. Sets up cloud resources before delegation."""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(CloudscaleCloudProvider, self).__init__(args)
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(CloudscaleCloudProvider, self).setup()
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
self.managed = False
self._use_static_config()
class CloudscaleCloudEnvironment(CloudEnvironment):
"""Cloudscale cloud environment plugin. Updates integration test environment
after delegation.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Cloudscale cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)

42
test/lib/ansible_test/_internal/cloud/cs.py
View file

@ -5,16 +5,16 @@ __metaclass__ = type
import json
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
)
from .. import types as t
from ..util import (
ApplicationError,
display,
ConfigParser,
display,
)
from ..config import (
IntegrationConfig,
)
from ..http import (
@ -30,15 +30,18 @@ from ..containers import (
wait_for_file,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
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
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(CsCloudProvider, self).__init__(args)
self.image = os.environ.get('ANSIBLE_CLOUDSTACK_CONTAINER', 'quay.io/ansible/cloudstack-test-container:1.4.0')
@ -48,7 +51,7 @@ class CsCloudProvider(CloudProvider):
self.uses_docker = True
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(CsCloudProvider, self).setup()
@ -57,7 +60,7 @@ class CsCloudProvider(CloudProvider):
else:
self._setup_dynamic()
def _setup_static(self):
def _setup_static(self): # type: () -> None
"""Configure CloudStack tests for use with static configuration."""
parser = ConfigParser()
parser.read(self.config_static_path)
@ -82,7 +85,7 @@ class CsCloudProvider(CloudProvider):
display.info('Read cs host "%s" and port %d from config: %s' % (self.host, self.port, self.config_static_path), verbosity=1)
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Create a CloudStack simulator using docker."""
config = self._read_config_template()
@ -129,11 +132,8 @@ class CsCloudProvider(CloudProvider):
self._write_config(config)
def _get_credentials(self, container_name):
"""Wait for the CloudStack simulator to return credentials.
:type container_name: str
:rtype: dict[str, str]
"""
def _get_credentials(self, container_name): # type: (str) -> t.Dict[str, t.Any]
"""Wait for the CloudStack simulator to return credentials."""
def check(value):
# noinspection PyBroadException
try:
@ -150,10 +150,8 @@ class CsCloudProvider(CloudProvider):
class CsCloudEnvironment(CloudEnvironment):
"""CloudStack cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)

60
test/lib/ansible_test/_internal/cloud/foreman.py
View file

@ -4,54 +4,49 @@ __metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..containers import (
run_support_container,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class ForemanProvider(CloudProvider):
"""Foreman plugin.
Sets up Foreman stub server for tests.
"""
"""Foreman plugin. Sets up Foreman stub server for tests."""
DOCKER_SIMULATOR_NAME = 'foreman-stub'
# Default image to run Foreman stub from.
#
# The simulator must be pinned to a specific version
# to guarantee CI passes with the version used.
#
# It's source source itself resides at:
# https://github.com/ansible/foreman-test-container
DOCKER_IMAGE = 'quay.io/ansible/foreman-test-container:1.4.0'
"""Default image to run Foreman stub from.
The simulator must be pinned to a specific version
to guarantee CI passes with the version used.
It's source source itself resides at:
https://github.com/ansible/foreman-test-container
"""
def __init__(self, args):
"""Set up container references for provider.
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(ForemanProvider, self).__init__(args)
self.__container_from_env = os.environ.get('ANSIBLE_FRMNSIM_CONTAINER')
"""Overrides target container, might be used for development.
"""
Overrides target container, might be used for development.
Use ANSIBLE_FRMNSIM_CONTAINER=whatever_you_want if you want
to use other image. Omit/empty otherwise.
"""
self.image = self.__container_from_env or self.DOCKER_IMAGE
self.uses_docker = True
def setup(self):
def setup(self): # type: () -> None
"""Setup cloud resource before delegation and reg cleanup callback."""
super(ForemanProvider, self).setup()
@ -60,7 +55,7 @@ class ForemanProvider(CloudProvider):
else:
self._setup_dynamic()
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Spawn a Foreman stub within docker container."""
foreman_port = 8080
@ -83,19 +78,14 @@ class ForemanProvider(CloudProvider):
self._set_cloud_config('FOREMAN_HOST', self.DOCKER_SIMULATOR_NAME)
self._set_cloud_config('FOREMAN_PORT', str(foreman_port))
def _setup_static(self):
def _setup_static(self): # type: () -> None
raise NotImplementedError()
class ForemanEnvironment(CloudEnvironment):
"""Foreman environment plugin.
Updates integration test environment after delegation.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Foreman environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
env_vars = dict(
FOREMAN_HOST=self._get_cloud_config('FOREMAN_HOST'),
FOREMAN_PORT=self._get_cloud_config('FOREMAN_PORT'),

37
test/lib/ansible_test/_internal/cloud/galaxy.py
View file

@ -5,10 +5,8 @@ __metaclass__ = type
import os
import tempfile
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..docker_util import (
@ -19,6 +17,12 @@ from ..containers import (
run_support_container,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
# We add BasicAuthentication, to make the tasks that deal with
# direct API access easier to deal with across galaxy_ng and pulp
@ -70,17 +74,11 @@ foreground {
class GalaxyProvider(CloudProvider):
"""Galaxy plugin.
Sets up pulp (ansible-galaxy) servers for tests.
"""
Galaxy plugin. Sets up pulp (ansible-galaxy) servers for tests.
The pulp source itself resides at: https://github.com/pulp/pulp-oci-images
"""
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(GalaxyProvider, self).__init__(args)
# Cannot use the latest container image as either galaxy_ng 4.2.0rc2 or pulp 0.5.0 has sporatic issues with
@ -94,7 +92,7 @@ class GalaxyProvider(CloudProvider):
self.uses_docker = True
def setup(self):
def setup(self): # type: () -> None
"""Setup cloud resource before delegation and reg cleanup callback."""
super(GalaxyProvider, self).setup()
@ -145,14 +143,9 @@ class GalaxyProvider(CloudProvider):
class GalaxyEnvironment(CloudEnvironment):
"""Galaxy environment plugin.
Updates integration test environment after delegation.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Galaxy environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
pulp_user = self._get_cloud_config('PULP_USER')
pulp_password = self._get_cloud_config('PULP_PASSWORD')
pulp_host = self._get_cloud_config('PULP_HOST')

19
test/lib/ansible_test/_internal/cloud/gcp.py
View file

@ -9,24 +9,25 @@ from ..util import (
ConfigParser,
)
from ..config import (
IntegrationConfig,
)
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class GcpCloudProvider(CloudProvider):
"""GCP cloud provider plugin. Sets up cloud resources before delegation."""
def __init__(self, args):
"""Set up container references for provider.
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(GcpCloudProvider, self).__init__(args)
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(GcpCloudProvider, self).setup()
@ -38,10 +39,8 @@ class GcpCloudProvider(CloudProvider):
class GcpCloudEnvironment(CloudEnvironment):
"""GCP cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)

58
test/lib/ansible_test/_internal/cloud/hcloud.py
View file

@ -2,40 +2,41 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from .. import types as t
from ..util import (
display,
ConfigParser,
display,
)
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..target import (
IntegrationTarget,
)
from ..core_ci import (
AnsibleCoreCI,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class HcloudCloudProvider(CloudProvider):
"""Hetzner Cloud provider plugin. Sets up cloud resources before
delegation.
"""
def __init__(self, args):
"""
:type args: TestConfig
"""
"""Hetzner Cloud provider plugin. Sets up cloud resources before delegation."""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(HcloudCloudProvider, self).__init__(args)
self.uses_config = True
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]
"""
def filter(self, targets, exclude): # type: (t.Tuple[IntegrationTarget, ...], t.List[str]) -> None
"""Filter out the cloud tests when the necessary config and resources are not available."""
aci = self._create_ansible_core_ci()
if aci.available:
@ -43,14 +44,14 @@ class HcloudCloudProvider(CloudProvider):
super(HcloudCloudProvider, self).filter(targets, exclude)
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(HcloudCloudProvider, self).setup()
if not self._use_static_config():
self._setup_dynamic()
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Request Hetzner credentials through the Ansible Core CI service."""
display.info('Provisioning %s cloud environment.' % self.platform, verbosity=1)
@ -76,22 +77,15 @@ class HcloudCloudProvider(CloudProvider):
self._write_config(config)
def _create_ansible_core_ci(self):
"""
:rtype: AnsibleCoreCI
"""
def _create_ansible_core_ci(self): # type: () -> AnsibleCoreCI
"""Return a Heztner instance of AnsibleCoreCI."""
return AnsibleCoreCI(self.args, 'hetzner', 'hetzner', persist=False, stage=self.args.remote_stage, provider='hetzner', internal=True)
class HcloudCloudEnvironment(CloudEnvironment):
"""Hetzner Cloud cloud environment plugin. Updates integration test environment
after delegation.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Hetzner Cloud cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)

14
test/lib/ansible_test/_internal/cloud/httptester.py
View file

@ -4,12 +4,6 @@ __metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
)
from ..util import (
display,
generate_password,
@ -23,6 +17,12 @@ from ..containers import (
run_support_container,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
KRB5_PASSWORD_ENV = 'KRB5_PASSWORD'
@ -83,7 +83,7 @@ class HttptesterProvider(CloudProvider):
class HttptesterEnvironment(CloudEnvironment):
"""HTTP Tester environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Returns the cloud environment config."""
"""Return environment configuration for use in the test environment after delegation."""
return CloudEnvironmentConfig(
env_vars=dict(
HTTPTESTER='1', # backwards compatibility for tests intended to work with or without HTTP Tester

58
test/lib/ansible_test/_internal/cloud/nios.py
View file

@ -4,43 +4,40 @@ __metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..config import (
IntegrationConfig,
)
from ..containers import (
run_support_container,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class NiosProvider(CloudProvider):
"""Nios plugin.
Sets up NIOS mock server for tests.
"""
"""Nios plugin. Sets up NIOS mock server for tests."""
DOCKER_SIMULATOR_NAME = 'nios-simulator'
# Default image to run the nios simulator.
#
# The simulator must be pinned to a specific version
# to guarantee CI passes with the version used.
#
# It's source source itself resides at:
# https://github.com/ansible/nios-test-container
DOCKER_IMAGE = 'quay.io/ansible/nios-test-container:1.3.0'
"""Default image to run the nios simulator.
The simulator must be pinned to a specific version
to guarantee CI passes with the version used.
It's source source itself resides at:
https://github.com/ansible/nios-test-container
"""
def __init__(self, args):
"""Set up container references for provider.
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(NiosProvider, self).__init__(args)
self.__container_from_env = os.environ.get('ANSIBLE_NIOSSIM_CONTAINER')
"""Overrides target container, might be used for development.
"""
Overrides target container, might be used for development.
Use ANSIBLE_NIOSSIM_CONTAINER=whatever_you_want if you want
to use other image. Omit/empty otherwise.
@ -50,7 +47,7 @@ class NiosProvider(CloudProvider):
self.uses_docker = True
def setup(self):
def setup(self): # type: () -> None
"""Setup cloud resource before delegation and reg cleanup callback."""
super(NiosProvider, self).setup()
@ -59,7 +56,7 @@ class NiosProvider(CloudProvider):
else:
self._setup_dynamic()
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Spawn a NIOS simulator within docker container."""
nios_port = 443
@ -81,19 +78,14 @@ class NiosProvider(CloudProvider):
self._set_cloud_config('NIOS_HOST', self.DOCKER_SIMULATOR_NAME)
def _setup_static(self):
def _setup_static(self): # type: () -> None
raise NotImplementedError()
class NiosEnvironment(CloudEnvironment):
"""NIOS environment plugin.
Updates integration test environment after delegation.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""NIOS environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
ansible_vars = dict(
nios_provider=dict(
host=self._get_cloud_config('NIOS_HOST'),

28
test/lib/ansible_test/_internal/cloud/opennebula.py
View file

@ -2,21 +2,21 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..util import (
ConfigParser,
display,
)
from ..util import (
display,
ConfigParser,
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class OpenNebulaCloudProvider(CloudProvider):
"""Checks if a configuration file has been passed or fixtures are going to be used for testing"""
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(OpenNebulaCloudProvider, self).setup()
@ -25,7 +25,7 @@ class OpenNebulaCloudProvider(CloudProvider):
self.uses_config = True
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
display.info('No config file provided, will run test from fixtures')
config = self._read_config_template()
@ -41,13 +41,9 @@ class OpenNebulaCloudProvider(CloudProvider):
class OpenNebulaCloudEnvironment(CloudEnvironment):
"""
Updates integration test environment after delegation. Will setup the config file as parameter.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Updates integration test environment after delegation. Will setup the config file as parameter."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)

41
test/lib/ansible_test/_internal/cloud/openshift.py
View file

@ -4,12 +4,6 @@ __metaclass__ = type
import re
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
)
from ..io import (
read_text_file,
)
@ -18,20 +12,27 @@ from ..util import (
display,
)
from ..config import (
IntegrationConfig,
)
from ..containers import (
run_support_container,
wait_for_file,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class OpenShiftCloudProvider(CloudProvider):
"""OpenShift cloud provider plugin. Sets up cloud resources before delegation."""
DOCKER_CONTAINER_NAME = 'openshift-origin'
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(OpenShiftCloudProvider, self).__init__(args, config_extension='.kubeconfig')
# The image must be pinned to a specific version to guarantee CI passes with the version used.
@ -40,7 +41,7 @@ class OpenShiftCloudProvider(CloudProvider):
self.uses_docker = True
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(OpenShiftCloudProvider, self).setup()
@ -49,7 +50,7 @@ class OpenShiftCloudProvider(CloudProvider):
else:
self._setup_dynamic()
def _setup_static(self):
def _setup_static(self): # type: () -> None
"""Configure OpenShift tests for use with static configuration."""
config = read_text_file(self.config_static_path)
@ -58,7 +59,7 @@ class OpenShiftCloudProvider(CloudProvider):
if not match:
display.warning('Could not find OpenShift endpoint in kubeconfig.')
def _setup_dynamic(self):
def _setup_dynamic(self): # type: () -> None
"""Create a OpenShift container using docker."""
port = 8443
@ -88,12 +89,8 @@ class OpenShiftCloudProvider(CloudProvider):
self._write_config(config)
def _get_config(self, container_name, server):
"""Get OpenShift config from container.
:type container_name: str
:type server: str
:rtype: dict[str, str]
"""
def _get_config(self, container_name, server): # type: (str, str) -> str
"""Get OpenShift config from container."""
stdout = wait_for_file(self.args, container_name, '/var/lib/origin/openshift.local.config/master/admin.kubeconfig', sleep=10, tries=30)
config = stdout
@ -105,10 +102,8 @@ class OpenShiftCloudProvider(CloudProvider):
class OpenShiftCloudEnvironment(CloudEnvironment):
"""OpenShift cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
env_vars = dict(
K8S_AUTH_KUBECONFIG=self.config_path,
)

40
test/lib/ansible_test/_internal/cloud/scaleway.py
View file

@ -2,48 +2,40 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
)
from ..util import (
ConfigParser,
display,
)
from ..config import (
IntegrationConfig,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class ScalewayCloudProvider(CloudProvider):
"""Checks if a configuration file has been passed or fixtures are going to be used for testing"""
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(ScalewayCloudProvider, self).__init__(args)
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(ScalewayCloudProvider, self).setup()
if os.path.isfile(self.config_static_path):
self.config_path = self.config_static_path
self.managed = False
self._use_static_config()
class ScalewayCloudEnvironment(CloudEnvironment):
"""
Updates integration test environment after delegation. Will setup the config file as parameter.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Updates integration test environment after delegation. Will setup the config file as parameter."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)

37
test/lib/ansible_test/_internal/cloud/vcenter.py
View file

@ -4,31 +4,32 @@ __metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
from ..util import (
ApplicationError,
ConfigParser,
display,
)
from ..util import (
display,
ConfigParser,
ApplicationError,
from ..config import (
IntegrationConfig,
)
from ..containers import (
run_support_container,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class VcenterProvider(CloudProvider):
"""VMware vcenter/esx plugin. Sets up cloud resources for tests."""
DOCKER_SIMULATOR_NAME = 'vcenter-simulator'
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(VcenterProvider, self).__init__(args)
# The simulator must be pinned to a specific version to guarantee CI passes with the version used.
@ -48,7 +49,7 @@ class VcenterProvider(CloudProvider):
self.uses_docker = False
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(VcenterProvider, self).setup()
@ -63,7 +64,7 @@ class VcenterProvider(CloudProvider):
else:
raise ApplicationError('Unknown vmware_test_platform: %s' % self.vmware_test_platform)
def _setup_dynamic_simulator(self):
def _setup_dynamic_simulator(self): # type: () -> None
"""Create a vcenter simulator using docker."""
ports = [
443,
@ -86,17 +87,15 @@ class VcenterProvider(CloudProvider):
self._set_cloud_config('vcenter_hostname', self.DOCKER_SIMULATOR_NAME)
def _setup_static(self):
def _setup_static(self): # type: () -> None
if not os.path.exists(self.config_static_path):
raise ApplicationError('Configuration file does not exist: %s' % self.config_static_path)
class VcenterEnvironment(CloudEnvironment):
"""VMware vcenter/esx environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
try:
# We may be in a container, so we cannot just reach VMWARE_TEST_PLATFORM,
# We do a try/except instead

39
test/lib/ansible_test/_internal/cloud/vultr.py
View file

@ -2,47 +2,40 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
)
from ..util import (
ConfigParser,
display,
)
from ..config import (
IntegrationConfig,
)
from . import (
CloudEnvironment,
CloudEnvironmentConfig,
CloudProvider,
)
class VultrCloudProvider(CloudProvider):
"""Checks if a configuration file has been passed or fixtures are going to be used for testing"""
def __init__(self, args):
"""
:type args: TestConfig
"""
def __init__(self, args): # type: (IntegrationConfig) -> None
super(VultrCloudProvider, self).__init__(args)
self.uses_config = True
def setup(self):
def setup(self): # type: () -> None
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(VultrCloudProvider, self).setup()
if os.path.isfile(self.config_static_path):
self.config_path = self.config_static_path
self.managed = False
self._use_static_config()
class VultrCloudEnvironment(CloudEnvironment):
"""
Updates integration test environment after delegation. Will setup the config file as parameter.
"""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
"""Updates integration test environment after delegation. Will setup the config file as parameter."""
def get_environment_config(self): # type: () -> CloudEnvironmentConfig
"""Return environment configuration for use in the test environment after delegation."""
parser = ConfigParser()
parser.read(self.config_path)