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

214 lines
6.3 KiB
Python

"""Azure plugin for integration tests."""
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
from ..util import (
ApplicationError,
display,
is_shippable,
ConfigParser,
)
from . import (
CloudProvider,
CloudEnvironment,
CloudEnvironmentConfig,
)
from ..http import (
HttpClient,
urlparse,
urlunparse,
parse_qs,
)
from ..core_ci import (
AnsibleCoreCI,
)
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
"""
super(AzureCloudProvider, self).__init__(args)
self.aci = None
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 os.path.isfile(self.SHERLOCK_CONFIG_PATH):
return
if is_shippable():
return
super(AzureCloudProvider, self).filter(targets, exclude)
def setup(self):
"""Setup the cloud resource before delegation and register a cleanup callback."""
super(AzureCloudProvider, self).setup()
if not self._use_static_config():
self._setup_dynamic()
get_config(self.config_path) # check required variables
def cleanup(self):
"""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):
"""Request Azure credentials through Sherlock."""
display.info('Provisioning %s cloud environment.' % self.platform, verbosity=1)
config = self._read_config_template()
response = {}
if os.path.isfile(self.SHERLOCK_CONFIG_PATH):
with open(self.SHERLOCK_CONFIG_PATH, 'r') as sherlock_fd:
sherlock_uri = sherlock_fd.readline().strip() + '&rgcount=2'
parts = urlparse(sherlock_uri)
query_string = parse_qs(parts.query)
base_uri = urlunparse(parts[:4] + ('', ''))
if 'code' not in query_string:
example_uri = 'https://example.azurewebsites.net/api/sandbox-provisioning'
raise ApplicationError('The Sherlock URI must include the API key in the query string. Example: %s?code=xxx' % example_uri)
display.info('Initializing azure/sherlock from: %s' % base_uri, verbosity=1)
http = HttpClient(self.args)
result = http.get(sherlock_uri)
display.info('Started azure/sherlock from: %s' % base_uri, verbosity=1)
if not self.args.explain:
response = result.json()
else:
aci = self._create_ansible_core_ci()
aci_result = aci.start()
if not self.args.explain:
response = aci_result['azure']
self.aci = aci
if not self.args.explain:
values = dict(
AZURE_CLIENT_ID=response['clientId'],
AZURE_SECRET=response['clientSecret'],
AZURE_SUBSCRIPTION_ID=response['subscriptionId'],
AZURE_TENANT=response['tenantId'],
RESOURCE_GROUP=response['resourceGroupNames'][0],
RESOURCE_GROUP_SECONDARY=response['resourceGroupNames'][1],
)
display.sensitive.add(values['AZURE_SECRET'])
config = '\n'.join('%s: %s' % (key, values[key]) for key in sorted(values))
config = '[default]\n' + config
self._write_config(config)
def _create_ansible_core_ci(self):
"""
:rtype: AnsibleCoreCI
"""
return AnsibleCoreCI(self.args, 'azure', 'azure', persist=False, stage=self.args.remote_stage, provider=self.args.remote_provider)
class AzureCloudEnvironment(CloudEnvironment):
"""Azure cloud environment plugin. Updates integration test environment after delegation."""
def get_environment_config(self):
"""
:rtype: CloudEnvironmentConfig
"""
env_vars = get_config(self.config_path)
display.sensitive.add(env_vars.get('AZURE_SECRET'))
display.sensitive.add(env_vars.get('AZURE_PASSWORD'))
ansible_vars = dict(
resource_prefix=self.resource_prefix,
)
ansible_vars.update(dict((key.lower(), value) for key, value in env_vars.items()))
return CloudEnvironmentConfig(
env_vars=env_vars,
ansible_vars=ansible_vars,
)
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 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]
"""
parser = ConfigParser()
parser.read(config_path)
config = dict((key.upper(), value) for key, value in parser.items('default'))
rg_vars = (
'RESOURCE_GROUP',
'RESOURCE_GROUP_SECONDARY',
)
sp_vars = (
'AZURE_CLIENT_ID',
'AZURE_SECRET',
'AZURE_SUBSCRIPTION_ID',
'AZURE_TENANT',
)
ad_vars = (
'AZURE_AD_USER',
'AZURE_PASSWORD',
'AZURE_SUBSCRIPTION_ID',
)
rg_ok = all(var in config for var in rg_vars)
sp_ok = all(var in config for var in sp_vars)
ad_ok = all(var in config for var in ad_vars)
if not rg_ok:
raise ApplicationError('Resource groups must be defined with: %s' % ', '.join(sorted(rg_vars)))
if not sp_ok and not ad_ok:
raise ApplicationError('Credentials must be defined using either:\nService Principal: %s\nActive Directory: %s' % (
', '.join(sorted(sp_vars)), ', '.join(sorted(ad_vars))))
return config