saltstack/salt
1
0
Fork 0
mirror of https://github.com/saltstack/salt.git synced 2025-04-17 10:10:20 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Merge pull request #54121 from Akm0d/gce_cloud_test

Browse source 
Refactored cloud tests
This commit is contained in:
Daniel Wozniak 2019-08-16 10:24:13 -07:00 • committed by GitHub
commit 27d89bda2f
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 464 additions and 1292 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

99
tests/integration/cloud/clouds/test_digitalocean.py
View file

@ -7,74 +7,28 @@ Integration tests for DigitalOcean APIv2
from __future__ import absolute_import, print_function, unicode_literals
import base64
import hashlib
import os
from Crypto.PublicKey import RSA
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
from tests.integration.cloud.helpers.cloud_test_base import CloudTest, TIMEOUT
# Import Salt Libs
from salt.config import cloud_providers_config
from salt.ext.six.moves import range
import salt.utils.stringutils
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'digitalocean'
class DigitalOceanTest(ShellCase):
class DigitalOceanTest(CloudTest):
'''
Integration tests for the DigitalOcean cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(DigitalOceanTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'digitalocean-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if personal access token, ssh_key_file, and ssh_key_names are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
personal_token = config[profile_str][PROVIDER_NAME]['personal_access_token']
ssh_file = config[profile_str][PROVIDER_NAME]['ssh_key_file']
ssh_name = config[profile_str][PROVIDER_NAME]['ssh_key_name']
if personal_token == '' or ssh_file == '' or ssh_name == '':
self.skipTest(
'A personal access token, an ssh key file, and an ssh key name '
'must be provided to run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'digitalocean'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('personal_access_token', 'ssh_key_file', 'ssh_key_name')
def test_list_images(self):
'''
Tests the return of running the --list-images command for digitalocean
'''
image_list = self.run_cloud('--list-images {0}'.format(PROVIDER_NAME))
image_list = self.run_cloud('--list-images {0}'.format(self.PROVIDER))
self.assertIn(
'14.04.5 x64',
[i.strip() for i in image_list]
@ -84,7 +38,7 @@ class DigitalOceanTest(ShellCase):
'''
Tests the return of running the --list-locations command for digitalocean
'''
_list_locations = self.run_cloud('--list-locations {0}'.format(PROVIDER_NAME))
_list_locations = self.run_cloud('--list-locations {0}'.format(self.PROVIDER))
self.assertIn(
'San Francisco 2',
[i.strip() for i in _list_locations]
@ -94,7 +48,7 @@ class DigitalOceanTest(ShellCase):
'''
Tests the return of running the --list-sizes command for digitalocean
'''
_list_sizes = self.run_cloud('--list-sizes {0}'.format(PROVIDER_NAME))
_list_sizes = self.run_cloud('--list-sizes {0}'.format(self.PROVIDER))
self.assertIn(
'16gb',
[i.strip() for i in _list_sizes]
@ -104,7 +58,7 @@ class DigitalOceanTest(ShellCase):
'''
Test key management
'''
do_key_name = INSTANCE_NAME + '-key'
do_key_name = self.instance_name + '-key'
# generate key and fingerprint
ssh_key = RSA.generate(4096)
@ -113,7 +67,8 @@ class DigitalOceanTest(ShellCase):
finger_print = ':'.join([key_hex[x:x+2] for x in range(0, len(key_hex), 2)])
try:
_key = self.run_cloud('-f create_key {0} name="{1}" public_key="{2}"'.format(PROVIDER_NAME, do_key_name, pub))
_key = self.run_cloud('-f create_key {0} name="{1}" public_key="{2}"'.format(self.PROVIDER,
do_key_name, pub))
# Upload public key
self.assertIn(
@ -122,7 +77,7 @@ class DigitalOceanTest(ShellCase):
)
# List all keys
list_keypairs = self.run_cloud('-f list_keypairs {0}'.format(PROVIDER_NAME))
list_keypairs = self.run_cloud('-f list_keypairs {0}'.format(self.PROVIDER))
self.assertIn(
finger_print,
@ -130,45 +85,25 @@ class DigitalOceanTest(ShellCase):
)
# List key
show_keypair = self.run_cloud('-f show_keypair {0} keyname={1}'.format(PROVIDER_NAME,
do_key_name))
show_keypair = self.run_cloud('-f show_keypair {0} keyname={1}'.format(self.PROVIDER, do_key_name))
self.assertIn(
finger_print,
[i.strip() for i in show_keypair]
)
except AssertionError:
# Delete the public key if the above assertions fail
self.run_cloud('-f remove_key {0} id={1}'.format(PROVIDER_NAME, finger_print))
self.run_cloud('-f remove_key {0} id={1}'.format(self.PROVIDER, finger_print))
raise
finally:
# Delete public key
self.assertTrue(self.run_cloud('-f remove_key {0} id={1}'.format(PROVIDER_NAME, finger_print)))
self.assertTrue(self.run_cloud('-f remove_key {0} id={1}'.format(self.PROVIDER, finger_print)))
def test_instance(self):
'''
Test creating an instance on DigitalOcean
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud('-p digitalocean-test {0}'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
ret_str = self.run_cloud('-p digitalocean-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_str)
# delete the instance
try:
self.assertIn(
'True',
[i.strip() for i in self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
raise
# Final clean-up of created instance, in case something went wrong.
# This was originally in a tearDown function, but that didn't make sense
# To run this for each test when not all tests create instances.
if INSTANCE_NAME in [i.strip() for i in self.run_cloud('--query')]:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
self.assertDestroyInstance()

100
tests/integration/cloud/clouds/test_dimensiondata.py
View file

@ -5,84 +5,23 @@ Integration tests for the Dimension Data cloud provider
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
import random
import string
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest
# Import Salt Libs
from salt.config import cloud_providers_config
from salt.ext.six.moves import range # pylint: disable=import-error,redefined-builtin
from tests.integration.cloud.helpers.cloud_test_base import CloudTest, TIMEOUT
def _random_name(size=6):
'''
Generates a random cloud instance name
'''
return 'cloud-test-' + ''.join(
random.choice(string.ascii_lowercase + string.digits)
for x in range(size)
)
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = _random_name()
PROVIDER_NAME = 'dimensiondata'
class DimensionDataTest(ShellCase):
class DimensionDataTest(CloudTest):
'''
Integration tests for the Dimension Data cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(DimensionDataTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'dimensiondata-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if user_id, key, and region are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
user_id = config[profile_str][PROVIDER_NAME]['user_id']
key = config[profile_str][PROVIDER_NAME]['key']
region = config[profile_str][PROVIDER_NAME]['region']
if user_id == '' or key == '' or region == '':
self.skipTest(
'A user Id, password, and a region '
'must be provided to run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'dimensiondata'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('key', 'region', 'user_id')
def test_list_images(self):
'''
Tests the return of running the --list-images command for the dimensiondata cloud provider
'''
image_list = self.run_cloud('--list-images {0}'.format(PROVIDER_NAME))
image_list = self.run_cloud('--list-images {0}'.format(self.PROVIDER))
self.assertIn(
'Ubuntu 14.04 2 CPU',
[i.strip() for i in image_list]
@ -92,7 +31,7 @@ class DimensionDataTest(ShellCase):
'''
Tests the return of running the --list-locations command for the dimensiondata cloud provider
'''
_list_locations = self.run_cloud('--list-locations {0}'.format(PROVIDER_NAME))
_list_locations = self.run_cloud('--list-locations {0}'.format(self.PROVIDER))
self.assertIn(
'Australia - Melbourne MCP2',
[i.strip() for i in _list_locations]
@ -102,7 +41,7 @@ class DimensionDataTest(ShellCase):
'''
Tests the return of running the --list-sizes command for the dimensiondata cloud provider
'''
_list_sizes = self.run_cloud('--list-sizes {0}'.format(PROVIDER_NAME))
_list_sizes = self.run_cloud('--list-sizes {0}'.format(self.PROVIDER))
self.assertIn(
'default',
[i.strip() for i in _list_sizes]
@ -113,26 +52,7 @@ class DimensionDataTest(ShellCase):
Test creating an instance on Dimension Data's cloud
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud('-p dimensiondata-test {0}'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
ret_val = self.run_cloud('-p dimensiondata-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_val)
# delete the instance
try:
self.assertIn(
'True',
[i.strip() for i in self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
raise
# Final clean-up of created instance, in case something went wrong.
# This was originally in a tearDown function, but that didn't make sense
# To run this for each test when not all tests create instances.
if INSTANCE_NAME in [i.strip() for i in self.run_cloud('--query')]:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
self.assertDestroyInstance()

186
tests/integration/cloud/clouds/test_ec2.py
View file

@ -9,115 +9,69 @@ import os
import yaml
# Import Salt Libs
from salt.config import cloud_providers_config
import salt.utils.cloud
import salt.utils.files
import salt.utils.yaml
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
from tests.support.helpers import expensiveTest
from tests.support.unit import skipIf
from tests.support import win_installer
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'ec2'
from tests.integration.cloud.helpers.cloud_test_base import CloudTest
HAS_WINRM = salt.utils.cloud.HAS_WINRM and salt.utils.cloud.HAS_SMB
# THis test needs a longer timeout than other cloud tests
TIMEOUT = 1200
class EC2Test(ShellCase):
class EC2Test(CloudTest):
'''
Integration tests for the EC2 cloud provider in Salt-Cloud
'''
PROVIDER = 'ec2'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('id', 'key', 'keyname', 'private_key', 'location')
def _installer_name(self):
'''
Determine the downloaded installer name by searching the files
directory for the firt file that loosk like an installer.
'''
@staticmethod
def __fetch_installer():
# Determine the downloaded installer name by searching the files
# directory for the first file that looks like an installer.
for path, dirs, files in os.walk(FILES):
for file in files:
if file.startswith(win_installer.PREFIX):
return file
break
return
def _fetch_latest_installer(self):
'''
Download the latest Windows installer executable
'''
# If the installer wasn't found in the previous steps, download the latest Windows installer executable
name = win_installer.latest_installer_name()
path = os.path.join(FILES, name)
with salt.utils.files.fopen(path, 'wb') as fp:
win_installer.download_and_verify(fp, name)
return name
def _ensure_installer(self):
@property
def installer(self):
'''
Make sure the testing environment has a Windows installer executbale.
Make sure the testing environment has a Windows installer executable.
'''
name = self._installer_name()
if name:
return name
return self._fetch_latest_installer()
if not hasattr(self, '_installer'):
self._installer = self.__fetch_installer()
return self._installer
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(EC2Test, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'ec2-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if id, key, keyname, securitygroup, private_key, location,
# and provider are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
id_ = config[profile_str][PROVIDER_NAME]['id']
key = config[profile_str][PROVIDER_NAME]['key']
key_name = config[profile_str][PROVIDER_NAME]['keyname']
private_key = config[profile_str][PROVIDER_NAME]['private_key']
location = config[profile_str][PROVIDER_NAME]['location']
group_or_subnet = config[profile_str][PROVIDER_NAME].get('securitygroup', '')
group_or_subnet = self.provider_config.get('securitygroup')
if not group_or_subnet:
group_or_subnet = config[profile_str][PROVIDER_NAME].get('subnetid', '')
group_or_subnet = self.provider_config.get('subnetid')
conf_items = [id_, key, key_name, private_key, location, group_or_subnet]
missing_conf_item = []
if not group_or_subnet:
self.skipTest('securitygroup or subnetid missing for {} config'.format(self.PROVIDER))
for item in conf_items:
if item == '':
missing_conf_item.append(item)
if missing_conf_item:
self.skipTest(
'An id, key, keyname, security group, private key, and location must '
'be provided to run these tests. One or more of these elements is '
'missing. Check tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
self.INSTALLER = self._ensure_installer()
super(EC2Test, self).setUp()
def override_profile_config(self, name, data):
conf_path = os.path.join(self.config_dir, 'cloud.profiles.d', 'ec2.conf')
@ -140,76 +94,52 @@ class EC2Test(ShellCase):
dfp.write(sfp.read())
return dst
def _test_instance(self, profile='ec2-test', debug=False, timeout=TIMEOUT):
def _test_instance(self, profile, debug):
'''
Tests creating and deleting an instance on EC2 (classic)
'''
# create the instance
cmd = '-p {0}'.format(profile)
cmd = ['-p', profile]
if debug:
cmd += ' -l debug'
cmd += ' {0}'.format(INSTANCE_NAME)
instance = self.run_cloud(cmd, timeout=timeout)
ret_str = '{0}:'.format(INSTANCE_NAME)
cmd.extend(['-l', 'debug'])
cmd.append(self.instance_name)
ret_val = self.run_cloud(' '.join(cmd), timeout=TIMEOUT)
# check if instance returned with salt installed
try:
self.assertIn(ret_str, instance)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=timeout)
raise
self.assertInstanceExists(ret_val)
# Let the instance exist for a bit before destroying it, otherwise the test will fail
# delete the instance
delete = self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=timeout)
ret_str = ' shutting-down'
# check if deletion was performed appropriately
try:
self.assertIn(ret_str, delete)
except AssertionError:
raise
self.assertDestroyInstance()
def test_instance_rename(self):
'''
Tests creating and renaming an instance on EC2 (classic)
'''
# Start with a name that is different from usual so that it will get deleted normally after the test
changed_name = self.instance_name + '-changed'
# create the instance
rename = INSTANCE_NAME + '-rename'
instance = self.run_cloud('-p ec2-test {0} --no-deploy'.format(INSTANCE_NAME), timeout=TIMEOUT)
ret_str = '{0}:'.format(INSTANCE_NAME)
ret_val = self.run_cloud('-p ec2-test {0} --no-deploy'.format(changed_name), timeout=TIMEOUT)
# check if instance returned
try:
self.assertIn(ret_str, instance)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
raise
self.assertInstanceExists(ret_val, changed_name)
change_name = self.run_cloud('-a rename {0} newname={1} --assume-yes'.format(INSTANCE_NAME, rename), timeout=TIMEOUT)
change_name = self.run_cloud('-a rename {0} newname={1} --assume-yes'.format(changed_name, self.instance_name),
timeout=TIMEOUT)
check_rename = self.run_cloud('-a show_instance {0} --assume-yes'.format(rename), [rename])
exp_results = [' {0}:'.format(rename), ' size:',
check_rename = self.run_cloud('-a show_instance {0} --assume-yes'.format(self.instance_name), [self.instance_name])
exp_results = [' {0}:'.format(self.instance_name), ' size:',
' architecture:']
try:
for result in exp_results:
self.assertIn(result, check_rename[0])
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
raise
for result in exp_results:
self.assertIn(result, check_rename[0])
# delete the instance
delete = self.run_cloud('-d {0} --assume-yes'.format(rename), timeout=TIMEOUT)
ret_str = ' shutting-down'
# check if deletion was performed appropriately
self.assertIn(ret_str, delete)
self.assertDestroyInstance()
def test_instance(self):
'''
Tests creating and deleting an instance on EC2 (classic)
'''
self._test_instance('ec2-test')
self._test_instance('ec2-test', debug=False)
def test_win2012r2_psexec(self):
'''
@ -218,15 +148,16 @@ class EC2Test(ShellCase):
'''
# TODO: psexec calls hang and the test fails by timing out. The same
# same calls succeed when run outside of the test environment.
# FIXME? Does this override need to be undone at the end of the test?
self.override_profile_config(
'ec2-win2012r2-test',
{
'use_winrm': False,
'userdata_file': self.copy_file('windows-firewall-winexe.ps1'),
'win_installer': self.copy_file(self.INSTALLER),
'win_installer': self.copy_file(self.installer),
},
)
self._test_instance('ec2-win2012r2-test', debug=True, timeout=TIMEOUT)
self._test_instance('ec2-win2012r2-test', debug=True)
@skipIf(not HAS_WINRM, 'Skip when winrm dependencies are missing')
def test_win2012r2_winrm(self):
@ -238,30 +169,30 @@ class EC2Test(ShellCase):
'ec2-win2012r2-test',
{
'userdata_file': self.copy_file('windows-firewall.ps1'),
'win_installer': self.copy_file(self.INSTALLER),
'win_installer': self.copy_file(self.installer),
'winrm_ssl_verify': False,
'use_winrm': True,
}
)
self._test_instance('ec2-win2012r2-test', debug=True, timeout=TIMEOUT)
self._test_instance('ec2-win2012r2-test', debug=True)
def test_win2016_psexec(self):
'''
Tests creating and deleting a Windows 2016 instance on EC2 using winrm
(classic)
'''
# TODO: winexe calls hang and the test fails by timing out. The same
# TODO: winexe calls hang and the test fails by timing out. The
# same calls succeed when run outside of the test environment.
self.override_profile_config(
'ec2-win2016-test',
{
'use_winrm': False,
'userdata_file': self.copy_file('windows-firewall-winexe.ps1'),
'win_installer': self.copy_file(self.INSTALLER),
'win_installer': self.copy_file(self.installer),
},
)
self._test_instance('ec2-win2016-test', debug=True, timeout=TIMEOUT)
self._test_instance('ec2-win2016-test', debug=True)
@skipIf(not HAS_WINRM, 'Skip when winrm dependencies are missing')
def test_win2016_winrm(self):
@ -273,21 +204,10 @@ class EC2Test(ShellCase):
'ec2-win2016-test',
{
'userdata_file': self.copy_file('windows-firewall.ps1'),
'win_installer': self.copy_file(self.INSTALLER),
'win_installer': self.copy_file(self.installer),
'winrm_ssl_verify': False,
'use_winrm': True,
}
)
self._test_instance('ec2-win2016-test', debug=True, timeout=TIMEOUT)
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
self._test_instance('ec2-win2016-test', debug=True)

122
tests/integration/cloud/clouds/test_gce.py
View file

@ -6,140 +6,38 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Libs
from salt.config import cloud_providers_config
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name, flaky
TIMEOUT = 500
from tests.integration.cloud.helpers.cloud_test_base import TIMEOUT, CloudTest
class GCETest(ShellCase):
class GCETest(CloudTest):
'''
Integration tests for the GCE cloud provider in Salt-Cloud
'''
PROVIDER = 'gce'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('project', 'service_account_email_address', 'service_account_private_key')
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(GCETest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'gce-config:'
provider = 'gce'
providers = self.run_cloud('--list-providers')
# Create the cloud instance name to be used throughout the tests
self.INSTANCE_NAME = generate_random_name('cloud-test-').lower()
if profile_str not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(provider)
)
# check if project, service_account_email_address, service_account_private_key
# and provider are present
path = os.path.join(FILES,
'conf',
'cloud.providers.d',
provider + '.conf')
config = cloud_providers_config(path)
project = config['gce-config']['gce']['project']
service_account_email_address = config['gce-config']['gce']['service_account_email_address']
service_account_private_key = config['gce-config']['gce']['service_account_private_key']
conf_items = [project, service_account_email_address, service_account_private_key]
missing_conf_item = []
for item in conf_items:
if item == '':
missing_conf_item.append(item)
if missing_conf_item:
self.skipTest(
'An project, service_account_email_address, service_account_private_key must '
'be provided to run these tests. One or more of these elements is '
'missing. Check tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(provider)
)
@flaky
def test_instance(self):
'''
Tests creating and deleting an instance on GCE
'''
# create the instance
instance = self.run_cloud('-p gce-test {0}'.format(self.INSTANCE_NAME), timeout=TIMEOUT)
ret_str = '{0}:'.format(self.INSTANCE_NAME)
ret_str = self.run_cloud('-p gce-test {0}'.format(self.instance_name), timeout=TIMEOUT)
# check if instance returned with salt installed
try:
self.assertIn(ret_str, instance)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(self.INSTANCE_NAME), timeout=TIMEOUT)
raise
self.assertInstanceExists(ret_str)
self.assertDestroyInstance()
# delete the instance
delete = self.run_cloud('-d {0} --assume-yes'.format(self.INSTANCE_NAME), timeout=TIMEOUT)
# example response: ['gce-config:', '----------', ' gce:', '----------', 'cloud-test-dq4e6c:', 'True', '']
delete_str = ''.join(delete)
# check if deletion was performed appropriately
try:
self.assertIn(self.INSTANCE_NAME, delete_str)
self.assertIn('True', delete_str)
except AssertionError:
raise
@flaky
def test_instance_extra(self):
'''
Tests creating and deleting an instance on GCE
'''
# create the instance
instance = self.run_cloud('-p gce-test-extra \
{0}'.format(self.INSTANCE_NAME),
timeout=TIMEOUT)
ret_str = '{0}:'.format(self.INSTANCE_NAME)
ret_str = self.run_cloud('-p gce-test-extra {0}'.format(self.instance_name), timeout=TIMEOUT)
# check if instance returned with salt installed
try:
self.assertIn(ret_str, instance)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(self.INSTANCE_NAME), timeout=TIMEOUT)
raise
# delete the instance
delete = self.run_cloud('-d {0} --assume-yes'.format(self.INSTANCE_NAME), timeout=TIMEOUT)
# example response: ['gce-config:', '----------', ' gce:', '----------', 'cloud-test-dq4e6c:', 'True', '']
delete_str = ''.join(delete)
# check if deletion was performed appropriately
try:
self.assertIn(self.INSTANCE_NAME, delete_str)
self.assertIn('True', delete_str)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
# salt-cloud -a show_instance myinstance
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(self.INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(self.INSTANCE_NAME), timeout=TIMEOUT)
self.assertInstanceExists(ret_str)
self.assertDestroyInstance()

82
tests/integration/cloud/clouds/test_gogrid.py
View file

@ -5,96 +5,28 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
from tests.support.unit import skipIf
# Import Salt Libs
from salt.config import cloud_providers_config
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'gogrid'
from tests.integration.cloud.helpers.cloud_test_base import CloudTest, TIMEOUT
@skipIf(True, 'waiting on bug report fixes from #13365')
class GoGridTest(ShellCase):
class GoGridTest(CloudTest):
'''
Integration tests for the GoGrid cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(GoGridTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'gogrid-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if client_key and api_key are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
api = config[profile_str][PROVIDER_NAME]['apikey']
shared_secret = config[profile_str][PROVIDER_NAME]['sharedsecret']
if api == '' or shared_secret == '':
self.skipTest(
'An api key and shared secret must be provided to run these tests. '
'Check tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'gogrid'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('apikey', 'sharedsecret')
def test_instance(self):
'''
Test creating an instance on GoGrid
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud('-p gogrid-test {0}'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
ret_str = self.run_cloud('-p gogrid-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_str)
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
self.assertDestroyInstance()

99
tests/integration/cloud/clouds/test_joyent.py
View file

@ -1,99 +0,0 @@
# -*- coding: utf-8 -*-
'''
:codeauthor: Nicole Thomas <nicole@saltstack.com>
'''
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
# Import Salt Libs
from salt.config import cloud_providers_config
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'joyent'
class JoyentTest(ShellCase):
'''
Integration tests for the Joyent cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(JoyentTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'joyent-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if user, password, private_key, and keyname are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
user = config[profile_str][PROVIDER_NAME]['user']
password = config[profile_str][PROVIDER_NAME]['password']
private_key = config[profile_str][PROVIDER_NAME]['private_key']
keyname = config[profile_str][PROVIDER_NAME]['keyname']
if user == '' or password == '' or private_key == '' or keyname == '':
self.skipTest(
'A user name, password, private_key file path, and a key name '
'must be provided to run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
def test_instance(self):
'''
Test creating and deleting instance on Joyent
'''
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud('-p joyent-test {0}'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)

83
tests/integration/cloud/clouds/test_linode.py
View file

@ -5,94 +5,25 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
# Import Salt Libs
from salt.config import cloud_providers_config
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'linode'
from tests.integration.cloud.helpers.cloud_test_base import CloudTest, TIMEOUT
class LinodeTest(ShellCase):
class LinodeTest(CloudTest):
'''
Integration tests for the Linode cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(LinodeTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'linode-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if personal access token, ssh_key_file, and ssh_key_names are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
api = config[profile_str][PROVIDER_NAME]['apikey']
password = config[profile_str][PROVIDER_NAME]['password']
if api == '' or password == '':
self.skipTest(
'An api key and password must be provided to run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'.format(
PROVIDER_NAME
)
)
PROVIDER = 'linode'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('apikey', 'password')
def test_instance(self):
'''
Test creating an instance on Linode
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud('-p linode-test {0}'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
ret_str = self.run_cloud('-p linode-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_str)
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
self.assertDestroyInstance()

115
tests/integration/cloud/clouds/test_msazure.py
View file

@ -5,22 +5,14 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
import logging
import time
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.integration.cloud.helpers.cloud_test_base import TIMEOUT as CLOUD_TIMEOUT, CloudTest
from tests.support.unit import skipIf
from tests.support.helpers import expensiveTest, generate_random_name
# Import Salt Libs
from salt.config import cloud_providers_config
from salt.utils.versions import LooseVersion
from salt.ext.six.moves import range
TIMEOUT = 1000
try:
import azure # pylint: disable=unused-import
@ -31,14 +23,11 @@ except ImportError:
if HAS_AZURE and not hasattr(azure, '__version__'):
import azure.common
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'azurearm'
PROFILE_NAME = 'azure-test'
REQUIRED_AZURE = '0.11.1'
log = logging.getLogger(__name__)
TIMEOUT = CLOUD_TIMEOUT * 2
REQUIRED_AZURE = '0.11.1'
def __has_required_azure():
'''
@ -55,101 +44,21 @@ def __has_required_azure():
return False
@skipIf(HAS_AZURE is False, 'These tests require the Azure Python SDK to be installed.')
@skipIf(__has_required_azure() is False, 'The Azure Python SDK must be >= 0.11.1.')
class AzureTest(ShellCase):
@skipIf(not HAS_AZURE, 'These tests require the Azure Python SDK to be installed.')
@skipIf(not __has_required_azure(), 'The Azure Python SDK must be >= 0.11.1.')
class AzureTest(CloudTest):
'''
Integration tests for the Azure cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(AzureTest, self).setUp()
# check if appropriate cloud provider and profile files are present
provider_str = 'azure-config'
providers = self.run_cloud('--list-providers')
if provider_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if subscription_id and certificate_path are present in provider file
provider_config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
sub_id = provider_config[provider_str][PROVIDER_NAME]['subscription_id']
if sub_id == '':
self.skipTest(
'A subscription_id and certificate_path must be provided to run '
'these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'.format(
PROVIDER_NAME
)
)
PROVIDER = 'azurearm'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('subscription_id',)
def test_instance(self):
'''
Test creating an instance on Azure
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud(
'-p {0} {1}'.format(
PROFILE_NAME,
INSTANCE_NAME
), timeout=TIMEOUT
)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME),
timeout=TIMEOUT)
raise
ret_val = self.run_cloud('-p {0} {1}'.format(self.profile_str, self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_val)
# Try up to 10 times to delete the instance since it might not be
# available for deletion right away.
for num_try in range(10):
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud(
'-d {0} --assume-yes'.format(
INSTANCE_NAME
), timeout=TIMEOUT
)]
)
except AssertionError:
# The deletion did not succeed, wait 10s and try again
if num_try < 9:
log.warning('Unable to delete azure instance on try %d', num_try)
time.sleep(10)
else:
raise
else:
# The deletion succeeded
break
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME),
timeout=TIMEOUT)
self.assertDestroyInstance()

90
tests/integration/cloud/clouds/test_oneandone.py
View file

@ -5,16 +5,10 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.integration.cloud.helpers.cloud_test_base import CloudTest, TIMEOUT
from tests.support.unit import skipIf
from tests.support.helpers import expensiveTest, generate_random_name
# Import Salt Libs
from salt.config import cloud_providers_config
# Import Third-Party Libs
try:
@ -24,59 +18,19 @@ except ImportError:
HAS_ONEANDONE = False
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'oneandone'
DRIVER_NAME = 'oneandone'
@skipIf(HAS_ONEANDONE is False, 'salt-cloud requires >= 1and1 1.2.0')
class OneAndOneTest(ShellCase):
class OneAndOneTest(CloudTest):
'''
Integration tests for the 1and1 cloud provider
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(OneAndOneTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'oneandone-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf '
'files in tests/integration/files/conf/cloud.*.d/ to run '
'these tests.'.format(PROVIDER_NAME)
)
# check if api_token present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
api_token = config[profile_str][DRIVER_NAME]['api_token']
if api_token == '':
self.skipTest(
'api_token must be provided to '
'run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'oneandone'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('api_token',)
def test_list_images(self):
'''
Tests the return of running the --list-images command for 1and1
'''
image_list = self.run_cloud('--list-images {0}'.format(PROVIDER_NAME))
image_list = self.run_cloud('--list-images {0}'.format(self.PROVIDER_NAME))
self.assertIn(
'coreOSimage',
[i.strip() for i in image_list]
@ -87,35 +41,7 @@ class OneAndOneTest(ShellCase):
Test creating an instance on 1and1
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud(
'-p oneandone-test {0}'.format(INSTANCE_NAME), timeout=500
)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
ret_str = self.run_cloud('-p oneandone-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_str)
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud(
'-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500
)]
)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
self.assertDestroyInstance()

85
tests/integration/cloud/clouds/test_openstack.py
View file

@ -6,17 +6,15 @@ Tests for the Openstack Cloud Provider
# Import python libs
from __future__ import absolute_import, print_function, unicode_literals
import logging
import os
# Import Salt Testing libs
from tests.support.case import ModuleCase, ShellCase
from tests.support.paths import FILES
from tests.support.case import ModuleCase
from tests.support.unit import skipIf
from tests.support.helpers import destructiveTest, expensiveTest, generate_random_name
from tests.support.helpers import destructiveTest
from tests.support.mixins import SaltReturnAssertsMixin
# Import Salt Libs
from salt.config import cloud_providers_config
from tests.integration.cloud.helpers.cloud_test_base import TIMEOUT, CloudTest
log = logging.getLogger(__name__)
@ -31,15 +29,11 @@ except ImportError:
# Import Third-Party Libs
try:
import shade # pylint: disable=unused-import
HAS_SHADE = True
except ImportError:
HAS_SHADE = False
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'openstack'
DRIVER_NAME = 'openstack'
@skipIf(
not HAS_KEYSTONE,
@ -176,78 +170,19 @@ class OpenstackTest(ModuleCase, SaltReturnAssertsMixin):
@skipIf(not HAS_SHADE, 'openstack driver requires `shade`')
class RackspaceTest(ShellCase):
class RackspaceTest(CloudTest):
'''
Integration tests for the Rackspace cloud provider using the Openstack driver
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(RackspaceTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'openstack-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if personal access token, ssh_key_file, and ssh_key_names are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
region_name = config[profile_str][DRIVER_NAME].get('region_name')
auth = config[profile_str][DRIVER_NAME].get('auth')
cloud = config[profile_str][DRIVER_NAME].get('cloud')
if not region_name or not (auth or cloud):
self.skipTest(
'A region_name and (auth or cloud) must be provided to run these '
'tests. Check tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'openstack'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('auth', 'cloud', 'region_name')
def test_instance(self):
'''
Test creating an instance on rackspace with the openstack driver
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud('-p rackspace-test {0}'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
raise
ret_val = self.run_cloud('-p rackspace-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_val)
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)]
)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500)
self.assertDestroyInstance()

115
tests/integration/cloud/clouds/test_profitbricks.py
View file

@ -5,18 +5,13 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.unit import skipIf
from tests.support.helpers import expensiveTest, generate_random_name
# Import Salt Libs
from salt.config import cloud_providers_config
# Import Third-Party Libs
from tests.integration.cloud.helpers.cloud_test_base import TIMEOUT, CloudTest
try:
# pylint: disable=unused-import
from profitbricks.client import ProfitBricksService
@ -24,62 +19,31 @@ try:
except ImportError:
HAS_PROFITBRICKS = False
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'profitbricks'
DRIVER_NAME = 'profitbricks'
@skipIf(HAS_PROFITBRICKS is False, 'salt-cloud requires >= profitbricks 4.1.0')
class ProfitBricksTest(ShellCase):
class ProfitBricksTest(CloudTest):
'''
Integration tests for the ProfitBricks cloud provider
'''
PROVIDER = 'profitbricks'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('username', 'password', 'datacenter_id')
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(ProfitBricksTest, self).setUp()
username = self.provider_config.get('username')
password = self.provider_config.get('password')
# check if appropriate cloud provider and profile files are present
profile_str = 'profitbricks-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf '
'files in tests/integration/files/conf/cloud.*.d/ to run '
'these tests.'.format(PROVIDER_NAME)
)
# A default username and password must be hard-coded as defaults as per issue #46265
# If they are 'foo' and 'bar' it is the same as not being set
# check if credentials and datacenter_id present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
username = config[profile_str][DRIVER_NAME]['username']
password = config[profile_str][DRIVER_NAME]['password']
datacenter_id = config[profile_str][DRIVER_NAME]['datacenter_id']
self.datacenter_id = datacenter_id
if username in ('' or 'foo') or password in ('' or 'bar') or datacenter_id == '':
self.skipTest(
'A username, password, and an datacenter must be provided to '
'run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
self.skipTest('Conf items are missing that must be provided to run these tests: username, password'
'\nCheck tests/integration/files/conf/cloud.providers.d/{0}.conf'.format(self.PROVIDER))
def test_list_images(self):
'''
Tests the return of running the --list-images command for ProfitBricks
'''
list_images = self.run_cloud('--list-images {0}'.format(PROVIDER_NAME))
list_images = self.run_cloud('--list-images {0}'.format(self.PROVIDER))
self.assertIn(
'Ubuntu-16.04-LTS-server-2017-10-01',
[i.strip() for i in list_images]
@ -90,7 +54,7 @@ class ProfitBricksTest(ShellCase):
Tests the return of running the -f list_images
command for ProfitBricks
'''
cmd = '-f list_images {0}'.format(PROVIDER_NAME)
cmd = '-f list_images {0}'.format(self.PROVIDER)
list_images = self.run_cloud(cmd)
self.assertIn(
'- ubuntu:latest',
@ -101,7 +65,7 @@ class ProfitBricksTest(ShellCase):
'''
Tests the return of running the --list_sizes command for ProfitBricks
'''
list_sizes = self.run_cloud('--list-sizes {0}'.format(PROVIDER_NAME))
list_sizes = self.run_cloud('--list-sizes {0}'.format(self.PROVIDER))
self.assertIn(
'Micro Instance:',
[i.strip() for i in list_sizes]
@ -112,10 +76,10 @@ class ProfitBricksTest(ShellCase):
Tests the return of running the -f list_datacenters
command for ProfitBricks
'''
cmd = '-f list_datacenters {0}'.format(PROVIDER_NAME)
cmd = '-f list_datacenters {0}'.format(self.PROVIDER)
list_datacenters = self.run_cloud(cmd)
self.assertIn(
self.datacenter_id,
self.provider_config['datacenter_id'],
[i.strip() for i in list_datacenters]
)
@ -123,7 +87,7 @@ class ProfitBricksTest(ShellCase):
'''
Tests the return of running the -f list_nodes command for ProfitBricks
'''
list_nodes = self.run_cloud('-f list_nodes {0}'.format(PROVIDER_NAME))
list_nodes = self.run_cloud('-f list_nodes {0}'.format(self.PROVIDER))
self.assertIn(
'state:',
[i.strip() for i in list_nodes]
@ -139,7 +103,7 @@ class ProfitBricksTest(ShellCase):
Tests the return of running the -f list_nodes_full
command for ProfitBricks
'''
cmd = '-f list_nodes_full {0}'.format(PROVIDER_NAME)
cmd = '-f list_nodes_full {0}'.format(self.PROVIDER)
list_nodes = self.run_cloud(cmd)
self.assertIn(
'state:',
@ -156,7 +120,7 @@ class ProfitBricksTest(ShellCase):
Tests the return of running the --list-locations
command for ProfitBricks
'''
cmd = '--list-locations {0}'.format(PROVIDER_NAME)
cmd = '--list-locations {0}'.format(self.PROVIDER)
list_locations = self.run_cloud(cmd)
self.assertIn(
@ -184,42 +148,7 @@ class ProfitBricksTest(ShellCase):
Test creating an instance on ProfitBricks
'''
# check if instance with salt installed returned
try:
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in self.run_cloud(
'-p profitbricks-test {0}'.format(INSTANCE_NAME),
timeout=500
)]
)
except AssertionError:
self.run_cloud(
'-d {0} --assume-yes'.format(INSTANCE_NAME),
timeout=500
)
raise
ret_str = self.run_cloud('-p profitbricks-test {0}'.format(self.instance_name), timeout=TIMEOUT)
self.assertInstanceExists(ret_str)
# delete the instance
try:
self.assertIn(
INSTANCE_NAME + ':',
[i.strip() for i in self.run_cloud(
'-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=500
)]
)
except AssertionError:
raise
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret in query:
self.run_cloud(
'-d {0} --assume-yes'.format(INSTANCE_NAME),
timeout=500
)
self.assertDestroyInstance()

126
tests/integration/cloud/clouds/test_vmware.py
View file

@ -5,147 +5,55 @@
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
# Import Salt Libs
from salt.config import cloud_providers_config, cloud_config
from salt.ext import six
# Import Salt Testing LIbs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'vmware'
TIMEOUT = 500
from tests.integration.cloud.helpers.cloud_test_base import TIMEOUT, CloudTest
class VMWareTest(ShellCase):
class VMWareTest(CloudTest):
'''
Integration tests for the vmware cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
# check if appropriate cloud provider and profile files are present
profile_str = 'vmware-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if user, password, url and provider are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
user = config[profile_str][PROVIDER_NAME]['user']
password = config[profile_str][PROVIDER_NAME]['password']
url = config[profile_str][PROVIDER_NAME]['url']
conf_items = [user, password, url]
missing_conf_item = []
for item in conf_items:
if item == '':
missing_conf_item.append(item)
if missing_conf_item:
self.skipTest(
'A user, password, and url must be provided to run these tests.'
'One or more of these elements is missing. Check'
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'vmware'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('password', 'user', 'url')
def test_instance(self):
'''
Tests creating and deleting an instance on vmware and installing salt
'''
# create the instance
profile = os.path.join(
FILES,
'conf',
'cloud.profiles.d',
PROVIDER_NAME + '.conf'
)
disk_datastore = self.config['vmware-test']['devices']['disk']['Hard disk 2']['datastore']
profile_config = cloud_config(profile)
disk_datastore = profile_config['vmware-test']['devices']['disk']['Hard disk 2']['datastore']
instance = self.run_cloud('-p vmware-test {0}'.format(INSTANCE_NAME), timeout=TIMEOUT)
ret_str = '{0}:'.format(INSTANCE_NAME)
disk_datastore_str = ' [{0}] {1}/Hard disk 2-flat.vmdk'.format(disk_datastore, INSTANCE_NAME)
ret_val = self.run_cloud('-p vmware-test {0}'.format(self.instance_name), timeout=TIMEOUT)
disk_datastore_str = ' [{0}] {1}/Hard disk 2-flat.vmdk'.format(disk_datastore,
self.instance_name)
# check if instance returned with salt installed
try:
self.assertIn(ret_str, instance)
self.assertIn(disk_datastore_str, instance,
msg='Hard Disk 2 did not use the Datastore {0} '.format(disk_datastore))
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
raise
self.assertInstanceExists(ret_val)
self.assertIn(disk_datastore_str, ret_val,
msg='Hard Disk 2 did not use the Datastore {0} '.format(disk_datastore))
# delete the instance
delete = self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
ret_str = '{0}:\', \' True'.format(INSTANCE_NAME)
# check if deletion was performed appropriately
self.assertIn(ret_str, six.text_type(delete))
self.assertDestroyInstance()
def test_snapshot(self):
'''
Tests creating snapshot and creating vm with --no-deploy
'''
# create the instance
instance = self.run_cloud('-p vmware-test {0} --no-deploy'.format(INSTANCE_NAME),
timeout=TIMEOUT)
ret_str = '{0}:'.format(INSTANCE_NAME)
ret_val = self.run_cloud('-p vmware-test {0} --no-deploy'.format(self.instance_name),
timeout=TIMEOUT)
# check if instance returned with salt installed
try:
self.assertIn(ret_str, instance)
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
raise
self.assertInstanceExists(ret_val)
create_snapshot = self.run_cloud('-a create_snapshot {0} \
snapshot_name=\'Test Cloud\' \
memdump=True -y'.format(INSTANCE_NAME),
timeout=TIMEOUT)
memdump=True -y'.format(self.instance_name), timeout=TIMEOUT)
s_ret_str = 'Snapshot created successfully'
self.assertIn(s_ret_str, six.text_type(create_snapshot))
# delete the instance
delete = self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
ret_str = '{0}:\', \' True'.format(INSTANCE_NAME)
self.assertIn(ret_str, six.text_type(delete))
def tearDown(self):
'''
Clean up after tests
'''
query = self.run_cloud('--query')
ret_str = ' {0}:'.format(INSTANCE_NAME)
# if test instance is still present, delete it
if ret_str in query:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
self.assertDestroyInstance()

175
tests/integration/cloud/clouds/test_vultrpy.py
View file

@ -5,75 +5,25 @@ Integration tests for Vultr
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
import time
# Import Salt Testing Libs
from tests.support.case import ShellCase
from tests.support.paths import FILES
from tests.support.helpers import expensiveTest, generate_random_name
from tests.integration.cloud.helpers.cloud_test_base import CloudTest, TIMEOUT
from tests.support.unit import skipIf
# Import Salt Libs
from salt.config import cloud_providers_config
from salt.ext import six
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = generate_random_name('CLOUD-TEST-')
PROVIDER_NAME = 'vultr'
TIMEOUT = 500
class VultrTest(ShellCase):
class VultrTest(CloudTest):
'''
Integration tests for the Vultr cloud provider in Salt-Cloud
'''
@expensiveTest
def setUp(self):
'''
Sets up the test requirements
'''
super(VultrTest, self).setUp()
# check if appropriate cloud provider and profile files are present
profile_str = 'vultr-config'
providers = self.run_cloud('--list-providers')
if profile_str + ':' not in providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(PROVIDER_NAME)
)
# check if api_key, ssh_key_file, and ssh_key_names are present
config = cloud_providers_config(
os.path.join(
FILES,
'conf',
'cloud.providers.d',
PROVIDER_NAME + '.conf'
)
)
api_key = config[profile_str][PROVIDER_NAME]['api_key']
ssh_file = config[profile_str][PROVIDER_NAME]['ssh_key_file']
ssh_name = config[profile_str][PROVIDER_NAME]['ssh_key_name']
if api_key == '' or ssh_file == '' or ssh_name == '':
self.skipTest(
'An API key, an ssh key file, and an ssh key name '
'must be provided to run these tests. Check '
'tests/integration/files/conf/cloud.providers.d/{0}.conf'
.format(PROVIDER_NAME)
)
PROVIDER = 'vultr'
REQUIRED_PROVIDER_CONFIG_ITEMS = ('api_key', 'ssh_key_file', 'ssh_key_name')
def test_list_images(self):
'''
Tests the return of running the --list-images command for Vultr
'''
image_list = self.run_cloud('--list-images {0}'.format(PROVIDER_NAME))
image_list = self.run_cloud('--list-images {0}'.format(self.PROVIDER))
self.assertIn(
'Debian 8 x64 (jessie)',
@ -84,7 +34,7 @@ class VultrTest(ShellCase):
'''
Tests the return of running the --list-locations command for Vultr
'''
location_list = self.run_cloud('--list-locations {0}'.format(PROVIDER_NAME))
location_list = self.run_cloud('--list-locations {0}'.format(self.PROVIDER))
self.assertIn(
'New Jersey',
[i.strip() for i in location_list]
@ -94,51 +44,51 @@ class VultrTest(ShellCase):
'''
Tests the return of running the --list-sizes command for Vultr
'''
size_list = self.run_cloud('--list-sizes {0}'.format(PROVIDER_NAME))
size_list = self.run_cloud('--list-sizes {0}'.format(self.PROVIDER))
self.assertIn(
'32768 MB RAM,4x110 GB SSD,40.00 TB BW',
[i.strip() for i in size_list]
)
# Commented for now, Vultr driver does not yet support key management
# def test_key_management(self):
# '''
# Test key management
# '''
# pub = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAQQDDHr/jh2Jy4yALcK4JyWbVkPRaWmhck3IgCoeOO3z1e2dBowLh64QAM+Qb72pxekALga2oi4GvT+TlWNhzPH4V example'
# finger_print = '3b:16:bf:e4:8b:00:8b:b8:59:8c:a9:d3:f0:19:45:fa'
#
# _key = self.run_cloud('-f create_key {0} name="MyPubKey" public_key="{1}"'.format(PROVIDER_NAME, pub))
#
# # Upload public key
# self.assertIn(
# finger_print,
# [i.strip() for i in _key]
# )
#
# try:
# # List all keys
# list_keypairs = self.run_cloud('-f list_keypairs {0}'.format(PROVIDER_NAME))
#
# self.assertIn(
# finger_print,
# [i.strip() for i in list_keypairs]
# )
#
# # List key
# show_keypair = self.run_cloud('-f show_keypair {0} keyname={1}'.format(PROVIDER_NAME, 'MyPubKey'))
#
# self.assertIn(
# finger_print,
# [i.strip() for i in show_keypair]
# )
# except AssertionError:
# # Delete the public key if the above assertions fail
# self.run_cloud('-f remove_key {0} id={1}'.format(PROVIDER_NAME, finger_print))
# raise
#
# # Delete public key
# self.assertTrue(self.run_cloud('-f remove_key {0} id={1}'.format(PROVIDER_NAME, finger_print)))
# def test_key_management(self):
# '''
# Test key management
# '''
# pub = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAAAQQDDHr/jh2Jy4yALcK4JyWbVkPRaWmhck3IgCoeOO3z1e2dBowLh64QAM+Qb72pxekALga2oi4GvT+TlWNhzPH4V example'
# finger_print = '3b:16:bf:e4:8b:00:8b:b8:59:8c:a9:d3:f0:19:45:fa'
#
# _key = self.run_cloud('-f create_key {0} name="MyPubKey" public_key="{1}"'.format(self.PROVIDER, pub))
#
# # Upload public key
# self.assertIn(
# finger_print,
# [i.strip() for i in _key]
# )
#
# try:
# # List all keys
# list_keypairs = self.run_cloud('-f list_keypairs {0}'.format(self.PROVIDER))
#
# self.assertIn(
# finger_print,
# [i.strip() for i in list_keypairs]
# )
#
# # List key
# show_keypair = self.run_cloud('-f show_keypair {0} keyname={1}'.format(self.PROVIDER, 'MyPubKey'))
#
# self.assertIn(
# finger_print,
# [i.strip() for i in show_keypair]
# )
# except AssertionError:
# # Delete the public key if the above assertions fail
# self.run_cloud('-f remove_key {0} id={1}'.format(self.PROVIDER, finger_print))
# raise
#
# # Delete public key
# self.assertTrue(self.run_cloud('-f remove_key {0} id={1}'.format(self.PROVIDER, finger_print)))
@skipIf(True, 'Skipped temporarily')
def test_instance(self):
@ -146,36 +96,9 @@ class VultrTest(ShellCase):
Test creating an instance on Vultr
'''
# check if instance with salt installed returned
try:
create_vm = self.run_cloud('-p vultr-test {0}'.format(INSTANCE_NAME), timeout=800)
self.assertIn(
INSTANCE_NAME,
[i.strip() for i in create_vm]
)
self.assertNotIn('Failed to start', six.text_type(create_vm))
except AssertionError:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
raise
ret_val = self.run_cloud('-p vultr-test {0}'.format(self.instance_name), timeout=TIMEOUT + 300)
self.assertInstanceExists(ret_val)
# Vultr won't let us delete an instance less than 5 minutes old.
time.sleep(420)
# delete the instance
results = self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
try:
self.assertIn(
'True',
[i.strip() for i in results]
)
except AssertionError:
raise
# Final clean-up of created instance, in case something went wrong.
# This was originally in a tearDown function, but that didn't make sense
# To run this for each test when not all tests create instances.
# Also, Vultr won't let instances be deleted unless they have been alive for 5 minutes.
# If we exceed 6 minutes and the instance is still there, quit
ct = 0
while ct < 12 and INSTANCE_NAME in [i.strip() for i in self.run_cloud('--query')]:
self.run_cloud('-d {0} --assume-yes'.format(INSTANCE_NAME), timeout=TIMEOUT)
time.sleep(30)
ct = ct + 1
time.sleep(300)
self.assertDestroyInstance()

227
tests/integration/cloud/helpers/cloud_test_base.py Normal file
View file

@ -0,0 +1,227 @@
# -*- coding: utf-8 -*-
'''
Tests for the Openstack Cloud Provider
'''
# Import python libs
from __future__ import absolute_import, print_function, unicode_literals
import logging
from os import path
from time import sleep
# Import Salt Testing libs
from tests.support.case import ShellCase
from tests.support.helpers import generate_random_name, expensiveTest
from tests.support.paths import FILES
# Import Salt Libs
from salt.config import cloud_config, cloud_providers_config
from salt.ext.six.moves import range
TIMEOUT = 500
log = logging.getLogger(__name__)
class CloudTest(ShellCase):
PROVIDER = ''
REQUIRED_PROVIDER_CONFIG_ITEMS = tuple()
__RE_RUN_DELAY = 15
__RE_TRIES = 12
def query_instances(self):
'''
Standardize the data returned from a salt-cloud --query
'''
return set(x.strip(': ') for x in self.run_cloud('--query') if x.lstrip().lower().startswith('cloud-test-'))
def _instance_exists(self, instance_name=None, query=None):
'''
:param instance_name: The name of the instance to check for in salt-cloud.
For example this is may used when a test temporarily renames an instance
:param query: The result of a salt-cloud --query run outside of this function
'''
# salt-cloud -a show_instance myinstance
if not instance_name:
instance_name = self.instance_name
if not query:
query = self.query_instances()
log.debug('Checking for "{}" in {}'.format(instance_name, query))
if isinstance(query, set):
return instance_name in query
return any(instance_name == q.strip(': ') for q in query)
def assertInstanceExists(self, creation_ret=None, instance_name=None):
'''
:param instance_name: Override the checked instance name, otherwise the class default will be used.
:param creation_ret: The return value from the run_cloud() function that created the instance
'''
if not instance_name:
instance_name = self.instance_name
# Verify that the instance exists via query
query = self.query_instances()
for tries in range(self.__RE_TRIES):
if self._instance_exists(instance_name, query):
log.debug('Instance "{}" reported after {} seconds'.format(instance_name, tries * self.__RE_RUN_DELAY))
break
else:
sleep(self.__RE_RUN_DELAY)
query = self.query_instances()
# Assert that the last query was successful
self.assertTrue(self._instance_exists(instance_name, query),
'Instance "{}" was not created successfully: |\n\t{}\n\t|`'.format(
instance_name, '\n\t'.join(creation_ret if creation_ret else query)))
log.debug('Instance exists and was created: "{}"'.format(instance_name))
# If it exists but doesn't show up in the creation_ret, there was probably an error during creation
if creation_ret:
self.assertIn(instance_name, [i.strip(': ') for i in creation_ret],
'An error occured during instance creation: |\n\t{}\n\t|'.format(
'\n\t'.join(creation_ret if creation_ret else query)
))
def _destroy_instance(self):
log.debug('Deleting instance "{}"'.format(self.instance_name))
delete = self.run_cloud('-d {0} --assume-yes'.format(self.instance_name), timeout=TIMEOUT)
# example response: ['gce-config:', '----------', ' gce:', '----------', 'cloud-test-dq4e6c:', 'True', '']
delete_str = ''.join(delete)
log.debug('Deletion status: {}'.format(delete_str))
if any([x in delete_str for x in (
'True',
'was successfully deleted'
)]):
log.debug('Instance "{}" was successfully deleted'.format(self.instance_name))
elif any([x in delete_str for x in (
'shutting-down',
'.delete',
)]):
log.debug('Instance "{}" is cleaning up'.format(self.instance_name))
sleep(30)
else:
log.warning('Instance "{}" may not have been deleted properly'.format(self.instance_name))
return delete_str
def assertDestroyInstance(self):
delete_str = self._destroy_instance()
# It might take a while to register that deletion has happened with `salt-cloud --query`
query = self.query_instances()
for tries in range(self.__RE_TRIES):
# If the instance doesn't exist, then deletion was a success. Move on
if not self._instance_exists(self.instance_name, query):
log.debug('Instance "{}" reported as deleted after {} seconds'.format(self.instance_name,
tries * self.__RE_RUN_DELAY))
break
else:
# Wait a bit and check again
sleep(self.__RE_RUN_DELAY)
query = self.query_instances()
# The instance should be reported as destroyed by the final query, otherwise fail
self.assertFalse(self._instance_exists(query), 'Could not destroy "{}". Delete_str: `{}`'
.format(self.instance_name, delete_str))
log.debug('Instance "{}" no longer exists'.format(self.instance_name))
@property
def instance_name(self):
if not hasattr(self, '_instance_name'):
# Create the cloud instance name to be used throughout the tests
subclass = self.__class__.__name__.strip('Test')
# Use the first three letters of the subclass, fill with '-' if too short
self._instance_name = generate_random_name('cloud-test-{:-<3}-'.format(subclass[:3])).lower()
return self._instance_name
@property
def providers(self):
if not hasattr(self, '_providers'):
self._providers = self.run_cloud('--list-providers')
return self._providers
@property
def provider_config(self):
if not hasattr(self, '_provider_config'):
self._provider_config = cloud_providers_config(
path.join(
FILES,
'conf',
'cloud.providers.d',
self.PROVIDER + '.conf'
)
)
return self._provider_config[self.profile_str][self.PROVIDER]
@property
def config(self):
if not hasattr(self, '_config'):
self._config = cloud_config(
path.join(
FILES,
'conf',
'cloud.profiles.d',
self.PROVIDER + '.conf'
)
)
return self._config
@property
def profile_str(self):
return self.PROVIDER + '-config'
@expensiveTest
def setUp(self):
'''
Sets up the test requirements. In child classes, define PROVIDER and REQUIRED_CONFIG_ITEMS or this will fail
'''
super(CloudTest, self).setUp()
if not self.PROVIDER:
self.fail('A PROVIDER must be defined for this test')
# check if appropriate cloud provider and profile files are present
if self.profile_str + ':' not in self.providers:
self.skipTest(
'Configuration file for {0} was not found. Check {0}.conf files '
'in tests/integration/files/conf/cloud.*.d/ to run these tests.'
.format(self.PROVIDER)
)
missing_conf_item = []
for att in self.REQUIRED_PROVIDER_CONFIG_ITEMS:
if not self.provider_config.get(att):
missing_conf_item.append(att)
if missing_conf_item:
self.skipTest('Conf items are missing that must be provided to run these tests: {}'
.format(', '.join(missing_conf_item)) +
'\nCheck tests/integration/files/conf/cloud.providers.d/{0}.conf'.format(self.PROVIDER))
self.assertFalse(self._instance_exists(),
'The instance "{}" exists before it was created by the test'.format(self.instance_name))
def tearDown(self):
'''
Clean up after tests, If the instance still exists for any reason, delete it.
Instances should be destroyed before the tearDown, _destroy_instance() should be called exactly
one time in a test for each instance created. This is a failSafe and something went wrong
if the tearDown is where an instance is destroyed.
'''
# Make sure that the instance for sure gets deleted, but fail the test if it happens in the tearDown
instance_deleted_before_teardown = True
for _ in range(12):
if self._instance_exists():
sleep(30)
instance_deleted_before_teardown = False
self._destroy_instance()
self.assertFalse(self._instance_exists(), 'Instance exists after multiple attempts to delete: {}'
.format(self.instance_name))
# Destroying instances in the tearDown is a contingency, not the way things should work by default.
self.assertTrue(instance_deleted_before_teardown,
'The Instance "{}" was deleted during the tearDown, not the test.'.format(self.instance_name))

52
tests/integration/cloud/test_cloud.py
View file

@ -5,55 +5,35 @@ Integration tests for functions located in the salt.cloud.__init__.py file.
# Import Python Libs
from __future__ import absolute_import, print_function, unicode_literals
import os
import random
import string
# Import Salt Testing libs
from tests.support.case import ShellCase
from tests.integration.cloud.helpers.cloud_test_base import CloudTest
from tests.support.helpers import expensiveTest
from tests.support.runtests import RUNTIME_VARS
# Import Salt libs
import salt.cloud
from salt.ext.six.moves import range
def __random_name(size=6):
'''
Generates a random cloud instance name
'''
return 'CLOUD-TEST-' + ''.join(
random.choice(string.ascii_uppercase + string.digits)
for x in range(size)
)
# Create the cloud instance name to be used throughout the tests
INSTANCE_NAME = __random_name()
class CloudClientTestCase(ShellCase):
class CloudClientTestCase(CloudTest):
'''
Integration tests for the CloudClient class. Uses DigitalOcean as a salt-cloud provider.
'''
PROVIDER = 'digitalocean'
REQUIRED_PROVIDER_CONFIG_ITEMS = tuple()
IMAGE_NAME = '14.04.5 x64'
@expensiveTest
def setUp(self):
self.config_file = os.path.join(RUNTIME_VARS.TMP_CONF_CLOUD_PROVIDER_INCLUDES,
'digitalocean.conf')
self.provider_name = 'digitalocean-config'
self.image_name = '14.04.5 x64'
# Use a --list-images salt-cloud call to see if the DigitalOcean provider is
# configured correctly before running any tests.
images = self.run_cloud('--list-images {0}'.format(self.provider_name))
images = self.run_cloud('--list-images {0}'.format(self.PROVIDER))
if self.image_name not in [i.strip() for i in images]:
self.skipTest(
'Image \'{0}\' was not found in image search. Is the {1} provider '
'configured correctly for this test?'.format(
self.provider_name,
self.PROVIDER,
self.image_name
)
)
@ -71,20 +51,18 @@ class CloudClientTestCase(ShellCase):
cloud_client = salt.cloud.CloudClient(self.config_file)
# Create the VM using salt.cloud.CloudClient.create() instead of calling salt-cloud
created = cloud_client.create(
provider=self.provider_name,
names=[INSTANCE_NAME],
image=self.image_name,
location='sfo1',
size='512mb',
vm_size='512mb'
ret_val = cloud_client.create(
provider=self.PROVIDER,
names=[self.instance_name],
image=self.IMAGE_NAME,
location='sfo1', size='512mb', vm_size='512mb'
)
# Check that the VM was created correctly
self.assertIn(INSTANCE_NAME, created)
self.assertInstanceExists(ret_val)
# Clean up after ourselves and delete the VM
deleted = cloud_client.destroy(names=[INSTANCE_NAME])
deleted = cloud_client.destroy(names=[self.instance_name])
# Check that the VM was deleted correctly
self.assertIn(INSTANCE_NAME, deleted)
self.assertIn(self.instance_name, deleted)