valitydev/salt
14
0
Fork 0
mirror of https://github.com/valitydev/salt.git synced 2024-11-09 01:36:48 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
78fc0f8799
salt/tests/unit/config_test.py
Colton Myers e386db2c34 Update syndic_config test for new cachedir
2015-03-19 11:33:09 -06:00

933 lines
38 KiB
Python
Raw Blame History

# -*- coding: utf-8 -*-
'''
:codeauthor: :email:`Pedro Algarvio (pedro@algarvio.me)`
tests.unit.config_test
~~~~~~~~~~~~~~~~~~~~~~
'''
# Import python libs
import logging
import os
import shutil
import tempfile
from contextlib import contextmanager
# Import Salt Testing libs
from salttesting import TestCase
from salttesting.mock import MagicMock, patch
from salttesting.helpers import ensure_in_syspath, TestsLoggingHandler
from salt.exceptions import CommandExecutionError
ensure_in_syspath('../')
# Import salt libs
import salt.minion
import salt.utils
import salt.utils.network
import integration
from salt import config as sconfig
from salt.exceptions import SaltCloudConfigError
# Import Third-Party Libs
import yaml
log = logging.getLogger(__name__)
# mock hostname should be more complex than the systems FQDN
MOCK_HOSTNAME = 'very.long.complex.fqdn.that.is.crazy.extra.long.example.com'
MOCK_ETC_HOSTS = (
'##\n'
'# Host Database\n'
'#\n'
'# localhost is used to configure the loopback interface\n'
'# when the system is booting. Do not change this entry.\n'
'##\n'
'\n' # This empty line MUST STAY HERE, it factors into the tests
'127.0.0.1 localhost ' + MOCK_HOSTNAME + '\n'
'10.0.0.100 ' + MOCK_HOSTNAME + '\n'
'200.200.200.2 other.host.alias.com\n'
'::1 ip6-localhost ip6-loopback\n'
'fe00::0 ip6-localnet\n'
'ff00::0 ip6-mcastprefix\n'
)
MOCK_ETC_HOSTNAME = '{0}\n'.format(MOCK_HOSTNAME)
PATH = 'path/to/some/cloud/conf/file'
DEFAULT = {'default_include': PATH}
def _unhandled_mock_read(filename):
'''
Raise an error because we should not be calling salt.utils.fopen()
'''
raise CommandExecutionError('Unhandled mock read for {0}'.format(filename))
@contextmanager
def _fopen_side_effect_etc_hostname(filename):
'''
Mock reading from /etc/hostname
'''
log.debug('Mock-reading {0}'.format(filename))
if filename == '/etc/hostname':
mock_open = MagicMock()
mock_open.read.return_value = MOCK_ETC_HOSTNAME
yield mock_open
elif filename == '/etc/hosts':
raise IOError(2, "No such file or directory: '{0}'".format(filename))
else:
_unhandled_mock_read(filename)
@contextmanager
def _fopen_side_effect_etc_hosts(filename):
'''
Mock /etc/hostname not existing, and falling back to reading /etc/hosts
'''
log.debug('Mock-reading {0}'.format(filename))
if filename == '/etc/hostname':
raise IOError(2, "No such file or directory: '{0}'".format(filename))
elif filename == '/etc/hosts':
mock_open = MagicMock()
mock_open.__iter__.return_value = MOCK_ETC_HOSTS.splitlines()
yield mock_open
else:
_unhandled_mock_read(filename)
class ConfigTestCase(TestCase, integration.AdaptedConfigurationTestCaseMixIn):
def test_proper_path_joining(self):
fpath = tempfile.mktemp()
try:
salt.utils.fopen(fpath, 'w').write(
"root_dir: /\n"
"key_logfile: key\n"
)
config = sconfig.master_config(fpath)
# os.path.join behavior
self.assertEqual(config['key_logfile'], os.path.join('/', 'key'))
# os.sep.join behavior
self.assertNotEqual(config['key_logfile'], '//key')
finally:
if os.path.isfile(fpath):
os.unlink(fpath)
def test_common_prefix_stripping(self):
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
try:
root_dir = os.path.join(tempdir, 'foo', 'bar')
os.makedirs(root_dir)
fpath = os.path.join(root_dir, 'config')
salt.utils.fopen(fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(root_dir, fpath)
)
config = sconfig.master_config(fpath)
self.assertEqual(config['log_file'], fpath)
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_load_master_config_from_environ_var(self):
original_environ = os.environ.copy()
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
try:
env_root_dir = os.path.join(tempdir, 'foo', 'env')
os.makedirs(env_root_dir)
env_fpath = os.path.join(env_root_dir, 'config-env')
salt.utils.fopen(env_fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(env_root_dir, env_fpath)
)
os.environ['SALT_MASTER_CONFIG'] = env_fpath
# Should load from env variable, not the default configuration file.
config = sconfig.master_config('/etc/salt/master')
self.assertEqual(config['log_file'], env_fpath)
os.environ.clear()
os.environ.update(original_environ)
root_dir = os.path.join(tempdir, 'foo', 'bar')
os.makedirs(root_dir)
fpath = os.path.join(root_dir, 'config')
salt.utils.fopen(fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(root_dir, fpath)
)
# Let's set the environment variable, yet, since the configuration
# file path is not the default one, ie, the user has passed an
# alternative configuration file form the CLI parser, the
# environment variable will be ignored.
os.environ['SALT_MASTER_CONFIG'] = env_fpath
config = sconfig.master_config(fpath)
self.assertEqual(config['log_file'], fpath)
os.environ.clear()
os.environ.update(original_environ)
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_load_minion_config_from_environ_var(self):
original_environ = os.environ.copy()
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
try:
env_root_dir = os.path.join(tempdir, 'foo', 'env')
os.makedirs(env_root_dir)
env_fpath = os.path.join(env_root_dir, 'config-env')
salt.utils.fopen(env_fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(env_root_dir, env_fpath)
)
os.environ['SALT_MINION_CONFIG'] = env_fpath
# Should load from env variable, not the default configuration file
config = sconfig.minion_config('/etc/salt/minion')
self.assertEqual(config['log_file'], env_fpath)
os.environ.clear()
os.environ.update(original_environ)
root_dir = os.path.join(tempdir, 'foo', 'bar')
os.makedirs(root_dir)
fpath = os.path.join(root_dir, 'config')
salt.utils.fopen(fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(root_dir, fpath)
)
# Let's set the environment variable, yet, since the configuration
# file path is not the default one, ie, the user has passed an
# alternative configuration file form the CLI parser, the
# environment variable will be ignored.
os.environ['SALT_MINION_CONFIG'] = env_fpath
config = sconfig.minion_config(fpath)
self.assertEqual(config['log_file'], fpath)
os.environ.clear()
os.environ.update(original_environ)
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_load_client_config_from_environ_var(self):
original_environ = os.environ.copy()
try:
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
env_root_dir = os.path.join(tempdir, 'foo', 'env')
os.makedirs(env_root_dir)
# Let's populate a master configuration file which should not get
# picked up since the client configuration tries to load the master
# configuration settings using the provided client configuration
# file
master_config = os.path.join(env_root_dir, 'master')
salt.utils.fopen(master_config, 'w').write(
'blah: true\n'
'root_dir: {0}\n'
'log_file: {1}\n'.format(env_root_dir, master_config)
)
os.environ['SALT_MASTER_CONFIG'] = master_config
# Now the client configuration file
env_fpath = os.path.join(env_root_dir, 'config-env')
salt.utils.fopen(env_fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(env_root_dir, env_fpath)
)
os.environ['SALT_CLIENT_CONFIG'] = env_fpath
# Should load from env variable, not the default configuration file
config = sconfig.client_config(os.path.expanduser('~/.salt'))
self.assertEqual(config['log_file'], env_fpath)
self.assertTrue('blah' not in config)
os.environ.clear()
os.environ.update(original_environ)
root_dir = os.path.join(tempdir, 'foo', 'bar')
os.makedirs(root_dir)
fpath = os.path.join(root_dir, 'config')
salt.utils.fopen(fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(root_dir, fpath)
)
# Let's set the environment variable, yet, since the configuration
# file path is not the default one, ie, the user has passed an
# alternative configuration file form the CLI parser, the
# environment variable will be ignored.
os.environ['SALT_MASTER_CONFIG'] = env_fpath
config = sconfig.master_config(fpath)
self.assertEqual(config['log_file'], fpath)
os.environ.clear()
os.environ.update(original_environ)
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_issue_5970_minion_confd_inclusion(self):
try:
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
minion_config = os.path.join(tempdir, 'minion')
minion_confd = os.path.join(tempdir, 'minion.d')
os.makedirs(minion_confd)
# Let's populate a minion configuration file with some basic
# settings
salt.utils.fopen(minion_config, 'w').write(
'blah: false\n'
'root_dir: {0}\n'
'log_file: {1}\n'.format(tempdir, minion_config)
)
# Now, let's populate an extra configuration file under minion.d
# Notice that above we've set blah as False and bellow as True.
# Since the minion.d files are loaded after the main configuration
# file so overrides can happen, the final value of blah should be
# True.
extra_config = os.path.join(minion_confd, 'extra.conf')
salt.utils.fopen(extra_config, 'w').write(
'blah: true\n'
)
# Let's load the configuration
config = sconfig.minion_config(minion_config)
self.assertEqual(config['log_file'], minion_config)
# As proven by the assertion below, blah is True
self.assertTrue(config['blah'])
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_master_confd_inclusion(self):
try:
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
master_config = os.path.join(tempdir, 'master')
master_confd = os.path.join(tempdir, 'master.d')
os.makedirs(master_confd)
# Let's populate a master configuration file with some basic
# settings
salt.utils.fopen(master_config, 'w').write(
'blah: false\n'
'root_dir: {0}\n'
'log_file: {1}\n'.format(tempdir, master_config)
)
# Now, let's populate an extra configuration file under master.d
# Notice that above we've set blah as False and bellow as True.
# Since the master.d files are loaded after the main configuration
# file so overrides can happen, the final value of blah should be
# True.
extra_config = os.path.join(master_confd, 'extra.conf')
salt.utils.fopen(extra_config, 'w').write(
'blah: true\n'
)
# Let's load the configuration
config = sconfig.master_config(master_config)
self.assertEqual(config['log_file'], master_config)
# As proven by the assertion below, blah is True
self.assertTrue(config['blah'])
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_syndic_config(self):
syndic_conf_path = self.get_config_file_path('syndic')
minion_conf_path = self.get_config_file_path('minion')
syndic_opts = sconfig.syndic_config(
syndic_conf_path, minion_conf_path
)
syndic_opts.update(salt.minion.resolve_dns(syndic_opts))
root_dir = syndic_opts['root_dir']
# id & pki dir are shared & so configured on the minion side
self.assertEqual(syndic_opts['id'], 'minion')
self.assertEqual(syndic_opts['pki_dir'], os.path.join(root_dir, 'pki'))
# the rest is configured master side
self.assertEqual(syndic_opts['master_uri'], 'tcp://127.0.0.1:54506')
self.assertEqual(syndic_opts['master_port'], 54506)
self.assertEqual(syndic_opts['master_ip'], '127.0.0.1')
self.assertEqual(syndic_opts['master'], 'localhost')
self.assertEqual(syndic_opts['sock_dir'], os.path.join(root_dir, 'minion_sock'))
self.assertEqual(syndic_opts['cachedir'], os.path.join(root_dir, 'cache'))
self.assertEqual(syndic_opts['log_file'], os.path.join(root_dir, 'osyndic.log'))
self.assertEqual(syndic_opts['pidfile'], os.path.join(root_dir, 'osyndic.pid'))
# Show that the options of localclient that repub to local master
# are not merged with syndic ones
self.assertEqual(syndic_opts['_master_conf_file'], minion_conf_path)
self.assertEqual(syndic_opts['_minion_conf_file'], syndic_conf_path)
def test_issue_6714_parsing_errors_logged(self):
try:
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
test_config = os.path.join(tempdir, 'config')
# Let's populate a master configuration file with some basic
# settings
salt.utils.fopen(test_config, 'w').write(
'root_dir: {0}\n'
'log_file: {0}/foo.log\n'.format(tempdir) +
'\n\n\n'
'blah:false\n'
)
with TestsLoggingHandler() as handler:
# Let's load the configuration
config = sconfig.master_config(test_config)
for message in handler.messages:
if message.startswith('ERROR:Error parsing configuration'):
break
else:
raise AssertionError(
'No parsing error message was logged'
)
finally:
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
@patch('salt.utils.network.get_fqhostname', MagicMock(return_value='localhost'))
def test_get_id_etc_hostname(self):
'''
Test calling salt.config.get_id() and falling back to looking at
/etc/hostname.
'''
with patch('salt.utils.fopen', _fopen_side_effect_etc_hostname):
self.assertEqual(
sconfig.get_id({'root_dir': None, 'minion_id_caching': False}), (MOCK_HOSTNAME, False)
)
@patch('salt.utils.network.get_fqhostname', MagicMock(return_value='localhost'))
def test_get_id_etc_hosts(self):
'''
Test calling salt.config.get_id() and falling back all the way to
looking up data from /etc/hosts.
'''
with patch('salt.utils.fopen', _fopen_side_effect_etc_hosts):
self.assertEqual(
sconfig.get_id({'root_dir': None, 'minion_id_caching': False}), (MOCK_HOSTNAME, False)
)
# <---- Salt Cloud Configuration Tests ---------------------------------------------
# cloud_config tests
@patch('salt.config.load_config', MagicMock(return_value={}))
def test_cloud_config_double_master_path(self):
'''
Tests passing in master_config_path and master_config kwargs.
'''
self.assertRaises(SaltCloudConfigError, sconfig.cloud_config, PATH,
master_config_path='foo', master_config='bar')
@patch('salt.config.load_config', MagicMock(return_value={}))
def test_cloud_config_double_providers_path(self):
'''
Tests passing in providers_config_path and providers_config kwargs.
'''
self.assertRaises(SaltCloudConfigError, sconfig.cloud_config, PATH,
providers_config_path='foo', providers_config='bar')
@patch('salt.config.load_config', MagicMock(return_value={}))
def test_cloud_config_double_profiles_path(self):
'''
Tests passing in profiles_config_path and profiles_config kwargs.
'''
self.assertRaises(SaltCloudConfigError, sconfig.cloud_config, PATH,
profiles_config_path='foo', profiles_config='bar')
@patch('salt.config.load_config', MagicMock(return_value={}))
@patch('salt.config.apply_cloud_config',
MagicMock(return_value={'providers': 'foo'}))
def test_cloud_config_providers_in_opts(self):
'''
Tests mixing old cloud providers with pre-configured providers configurations
using the providers_config kwarg
'''
self.assertRaises(SaltCloudConfigError, sconfig.cloud_config, PATH,
providers_config='bar')
@patch('salt.config.load_config', MagicMock(return_value={}))
@patch('salt.config.apply_cloud_config',
MagicMock(return_value={'providers': 'foo'}))
@patch('os.path.isfile', MagicMock(return_value=True))
def test_cloud_config_providers_in_opts_path(self):
'''
Tests mixing old cloud providers with pre-configured providers configurations
using the providers_config_path kwarg
'''
self.assertRaises(SaltCloudConfigError, sconfig.cloud_config, PATH,
providers_config_path='bar')
# apply_cloud_config tests
def test_apply_cloud_config_no_provider_detail_list(self):
'''
Tests when the provider is not contained in a list of details
'''
overrides = {'providers': {'foo': [{'bar': 'baz'}]}}
self.assertRaises(SaltCloudConfigError, sconfig.apply_cloud_config,
overrides, defaults=DEFAULT)
def test_apply_cloud_config_no_provider_detail_dict(self):
'''
Tests when the provider is not contained in the details dictionary
'''
overrides = {'providers': {'foo': {'bar': 'baz'}}}
self.assertRaises(SaltCloudConfigError, sconfig.apply_cloud_config,
overrides, defaults=DEFAULT)
@patch('salt.config.old_to_new',
MagicMock(return_value={'default_include': 'path/to/some/cloud/conf/file',
'providers': {'foo': {'bar': {
'provider': 'foo:bar'}}}}))
def test_apply_cloud_config_success_list(self):
'''
Tests success when valid data is passed into the function as a list
'''
overrides = {'providers': {'foo': [{'provider': 'bar'}]}}
ret = {'default_include': 'path/to/some/cloud/conf/file',
'providers': {'foo': {'bar': {'provider': 'foo:bar'}}}}
self.assertEqual(sconfig.apply_cloud_config(overrides, defaults=DEFAULT), ret)
@patch('salt.config.old_to_new',
MagicMock(return_value={'default_include': 'path/to/some/cloud/conf/file',
'providers': {'foo': {'bar': {
'provider': 'foo:bar'}}}}))
def test_apply_cloud_config_success_dict(self):
'''
Tests success when valid data is passed into function as a dictionary
'''
overrides = {'providers': {'foo': {'provider': 'bar'}}}
ret = {'default_include': 'path/to/some/cloud/conf/file',
'providers': {'foo': {'bar': {'provider': 'foo:bar'}}}}
self.assertEqual(sconfig.apply_cloud_config(overrides, defaults=DEFAULT), ret)
# apply_vm_profiles_config tests
def test_apply_vm_profiles_config_bad_profile_format(self):
'''
Tests passing in a bad profile format in overrides
'''
overrides = {'foo': 'bar', 'conf_file': PATH}
self.assertRaises(SaltCloudConfigError, sconfig.apply_vm_profiles_config,
PATH, overrides, defaults=DEFAULT)
def test_apply_vm_profiles_config_success(self):
'''
Tests passing in valid provider and profile config files successfully
'''
providers = {'test-provider':
{'digital_ocean':
{'provider': 'digital_ocean', 'profiles': {}}}}
overrides = {'test-profile':
{'provider': 'test-provider',
'image': 'Ubuntu 12.10 x64',
'size': '512MB'},
'conf_file': PATH}
ret = {'test-profile':
{'profile': 'test-profile',
'provider': 'test-provider:digital_ocean',
'image': 'Ubuntu 12.10 x64',
'size': '512MB'}}
self.assertEqual(sconfig.apply_vm_profiles_config(providers,
overrides,
defaults=DEFAULT), ret)
def test_apply_vm_profiles_config_extend_success(self):
'''
Tests profile extends functionality with valid provider and profile configs
'''
providers = {'test-config': {'ec2': {'profiles': {}, 'provider': 'ec2'}}}
overrides = {'Amazon': {'image': 'test-image-1',
'extends': 'dev-instances'},
'Fedora': {'image': 'test-image-2',
'extends': 'dev-instances'},
'conf_file': PATH,
'dev-instances': {'ssh_username': 'test_user',
'provider': 'test-config'}}
ret = {'Amazon': {'profile': 'Amazon',
'ssh_username': 'test_user',
'image': 'test-image-1',
'provider': 'test-config:ec2'},
'Fedora': {'profile': 'Fedora',
'ssh_username': 'test_user',
'image': 'test-image-2',
'provider': 'test-config:ec2'},
'dev-instances': {'profile': 'dev-instances',
'ssh_username': 'test_user',
'provider': 'test-config:ec2'}}
self.assertEqual(sconfig.apply_vm_profiles_config(providers,
overrides,
defaults=DEFAULT), ret)
# apply_cloud_providers_config tests
def test_apply_cloud_providers_config_same_providers(self):
'''
Tests when two providers are given with the same provider name
'''
overrides = {'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2'},
{'apikey': 'abcdefghijklmnopqrstuvwxyz',
'password': 'supersecret',
'provider': 'ec2'}],
'conf_file': PATH}
self.assertRaises(SaltCloudConfigError,
sconfig.apply_cloud_providers_config,
overrides,
DEFAULT)
def test_apply_cloud_providers_config_extend(self):
'''
Tests the successful extension of a cloud provider
'''
overrides = {'my-production-envs':
[{'extends': 'my-dev-envs:ec2',
'location': 'us-east-1',
'user': 'ec2-user@mycorp.com'
}],
'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com',
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2'
},
{'apikey': 'abcdefghijklmnopqrstuvwxyz',
'password': 'supersecret',
'provider': 'linode'
}],
'conf_file': PATH}
ret = {'my-production-envs':
{'ec2':
{'profiles': {},
'location': 'us-east-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2',
'id': 'ABCDEFGHIJKLMNOP',
'user': 'ec2-user@mycorp.com'}},
'my-dev-envs':
{'linode':
{'apikey': 'abcdefghijklmnopqrstuvwxyz',
'password': 'supersecret',
'profiles': {},
'provider': 'linode'},
'ec2':
{'profiles': {},
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2',
'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com'}}}
self.assertEqual(ret,
sconfig.apply_cloud_providers_config(
overrides,
defaults=DEFAULT))
def test_apply_cloud_providers_config_extend_multiple(self):
'''
Tests the successful extension of two cloud providers
'''
overrides = {'my-production-envs':
[{'extends': 'my-dev-envs:ec2',
'location': 'us-east-1',
'user': 'ec2-user@mycorp.com'},
{'password': 'new-password',
'extends': 'my-dev-envs:linode',
'location': 'Salt Lake City'
}],
'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com',
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2'},
{'apikey': 'abcdefghijklmnopqrstuvwxyz',
'password': 'supersecret',
'provider': 'linode'}],
'conf_file': PATH}
ret = {'my-production-envs':
{'linode':
{'apikey': 'abcdefghijklmnopqrstuvwxyz',
'profiles': {},
'location': 'Salt Lake City',
'provider': 'linode',
'password': 'new-password'},
'ec2':
{'user': 'ec2-user@mycorp.com',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2',
'id': 'ABCDEFGHIJKLMNOP',
'profiles': {},
'location': 'us-east-1'}},
'my-dev-envs':
{'linode':
{'apikey': 'abcdefghijklmnopqrstuvwxyz',
'password': 'supersecret',
'profiles': {},
'provider': 'linode'},
'ec2':
{'profiles': {},
'user': 'user@mycorp.com',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2',
'id': 'ABCDEFGHIJKLMNOP',
'location': 'ap-southeast-1'}}}
self.assertEqual(ret, sconfig.apply_cloud_providers_config(
overrides,
defaults=DEFAULT))
def test_apply_cloud_providers_config_extends_bad_alias(self):
'''
Tests when the extension contains an alias not found in providers list
'''
overrides = {'my-production-envs':
[{'extends': 'test-alias:ec2',
'location': 'us-east-1',
'user': 'ec2-user@mycorp.com'}],
'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com',
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2'}],
'conf_file': PATH}
self.assertRaises(SaltCloudConfigError,
sconfig.apply_cloud_providers_config,
overrides,
DEFAULT)
def test_apply_cloud_providers_config_extends_bad_provider(self):
'''
Tests when the extension contains a provider not found in providers list
'''
overrides = {'my-production-envs':
[{'extends': 'my-dev-envs:linode',
'location': 'us-east-1',
'user': 'ec2-user@mycorp.com'}],
'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com',
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'ec2'}],
'conf_file': PATH}
self.assertRaises(SaltCloudConfigError,
sconfig.apply_cloud_providers_config,
overrides,
DEFAULT)
def test_apply_cloud_providers_config_extends_no_provider(self):
'''
Tests when no provider is supplied in the extends statement
'''
overrides = {'my-production-envs':
[{'extends': 'my-dev-envs',
'location': 'us-east-1',
'user': 'ec2-user@mycorp.com'}],
'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com',
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'linode'}],
'conf_file': PATH}
self.assertRaises(SaltCloudConfigError,
sconfig.apply_cloud_providers_config,
overrides,
DEFAULT)
def test_apply_cloud_providers_extends_not_in_providers(self):
'''
Tests when extends is not in the list of providers
'''
overrides = {'my-production-envs':
[{'extends': 'my-dev-envs ec2',
'location': 'us-east-1',
'user': 'ec2-user@mycorp.com'}],
'my-dev-envs':
[{'id': 'ABCDEFGHIJKLMNOP',
'user': 'user@mycorp.com',
'location': 'ap-southeast-1',
'key': 'supersecretkeysupersecretkey',
'provider': 'linode'}],
'conf_file': PATH}
self.assertRaises(SaltCloudConfigError,
sconfig.apply_cloud_providers_config,
overrides,
DEFAULT)
# is_provider_configured tests
def test_is_provider_configured_no_alias(self):
'''
Tests when provider alias is not in opts
'''
opts = {'providers': 'test'}
provider = 'foo:bar'
self.assertFalse(sconfig.is_provider_configured(opts, provider))
def test_is_provider_configured_no_driver(self):
'''
Tests when provider driver is not in opts
'''
opts = {'providers': {'foo': 'baz'}}
provider = 'foo:bar'
self.assertFalse(sconfig.is_provider_configured(opts, provider))
def test_is_provider_configured_key_is_none(self):
'''
Tests when a required configuration key is not set
'''
opts = {'providers': {'foo': {'bar': {'api_key': None}}}}
provider = 'foo:bar'
self.assertFalse(
sconfig.is_provider_configured(opts,
provider,
required_keys=('api_key',)))
def test_is_provider_configured_success(self):
'''
Tests successful cloud provider configuration
'''
opts = {'providers': {'foo': {'bar': {'api_key': 'baz'}}}}
provider = 'foo:bar'
ret = {'api_key': 'baz'}
self.assertEqual(
sconfig.is_provider_configured(opts,
provider,
required_keys=('api_key',)), ret)
def test_is_provider_configured_multiple_driver_not_provider(self):
'''
Tests when the drive is not the same as the provider when
searching through multiple providers
'''
opts = {'providers': {'foo': {'bar': {'api_key': 'baz'}}}}
provider = 'foo'
self.assertFalse(sconfig.is_provider_configured(opts, provider))
def test_is_provider_configured_multiple_key_is_none(self):
'''
Tests when a required configuration key is not set when
searching through multiple providers
'''
opts = {'providers': {'foo': {'bar': {'api_key': None}}}}
provider = 'bar'
self.assertFalse(
sconfig.is_provider_configured(opts,
provider,
required_keys=('api_key',)))
def test_is_provider_configured_multiple_success(self):
'''
Tests successful cloud provider configuration when searching
through multiple providers
'''
opts = {'providers': {'foo': {'bar': {'api_key': 'baz'}}}}
provider = 'bar'
ret = {'api_key': 'baz'}
self.assertEqual(
sconfig.is_provider_configured(opts,
provider,
required_keys=('api_key',)), ret)
# other cloud configuration tests
def test_load_cloud_config_from_environ_var(self):
original_environ = os.environ.copy()
tempdir = tempfile.mkdtemp(dir=integration.SYS_TMP_DIR)
try:
env_root_dir = os.path.join(tempdir, 'foo', 'env')
os.makedirs(env_root_dir)
env_fpath = os.path.join(env_root_dir, 'config-env')
salt.utils.fopen(env_fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(env_root_dir, env_fpath)
)
os.environ['SALT_CLOUD_CONFIG'] = env_fpath
# Should load from env variable, not the default configuration file
config = sconfig.cloud_config('/etc/salt/cloud')
self.assertEqual(config['log_file'], env_fpath)
os.environ.clear()
os.environ.update(original_environ)
root_dir = os.path.join(tempdir, 'foo', 'bar')
os.makedirs(root_dir)
fpath = os.path.join(root_dir, 'config')
salt.utils.fopen(fpath, 'w').write(
'root_dir: {0}\n'
'log_file: {1}\n'.format(root_dir, fpath)
)
# Let's set the environment variable, yet, since the configuration
# file path is not the default one, ie, the user has passed an
# alternative configuration file form the CLI parser, the
# environment variable will be ignored.
os.environ['SALT_CLOUD_CONFIG'] = env_fpath
config = sconfig.cloud_config(fpath)
self.assertEqual(config['log_file'], fpath)
finally:
# Reset the environ
os.environ.clear()
os.environ.update(original_environ)
if os.path.isdir(tempdir):
shutil.rmtree(tempdir)
def test_deploy_search_path_as_string(self):
temp_conf_dir = os.path.join(integration.TMP, 'issue-8863')
config_file_path = os.path.join(temp_conf_dir, 'cloud')
deploy_dir_path = os.path.join(temp_conf_dir, 'test-deploy.d')
try:
for directory in (temp_conf_dir, deploy_dir_path):
if not os.path.isdir(directory):
os.makedirs(directory)
default_config = sconfig.cloud_config(config_file_path)
default_config['deploy_scripts_search_path'] = deploy_dir_path
with salt.utils.fopen(config_file_path, 'w') as cfd:
cfd.write(yaml.dump(default_config))
default_config = sconfig.cloud_config(config_file_path)
# Our custom deploy scripts path was correctly added to the list
self.assertIn(
deploy_dir_path,
default_config['deploy_scripts_search_path']
)
# And it's even the first occurrence as it should
self.assertEqual(
deploy_dir_path,
default_config['deploy_scripts_search_path'][0]
)
finally:
if os.path.isdir(temp_conf_dir):
shutil.rmtree(temp_conf_dir)
def test_includes_load(self):
'''
Tests that cloud.{providers,profiles}.d directories are loaded, even if not
directly passed in through path
'''
config = sconfig.cloud_config(self.get_config_file_path('cloud'))
self.assertIn('ec2-config', config['providers'])
self.assertIn('ec2-test', config['profiles'])
# <---- Salt Cloud Configuration Tests ---------------------------------------------
if __name__ == '__main__':
from integration import run_tests
run_tests(ConfigTestCase, needs_daemon=False)
Reference in New Issue View Git Blame Copy Permalink