2
"""Tests for assess_public_clouds."""
4
from argparse import Namespace
5
from contextlib import contextmanager
15
from assess_public_clouds import (
16
bootstrap_cloud_regions,
25
from fakejuju import (
37
_LOCAL = 'assess_public_clouds'
40
def patch_local(target, **kwargs):
41
return patch(_LOCAL + '.' + target, **kwargs)
44
class TestParseArgs(TestCase):
46
def test_parse_args(self):
48
self.assertEqual(Namespace(
49
deadline=None, debug=False, juju_bin='/usr/bin/juju', logs=None,
53
def test_parse_args_start(self):
54
args = parse_args(['--start', '7'])
55
self.assertEqual(7, args.start)
58
class TestMain(TestCase):
61
def patch_for_test(self, test_iterator, args=[], clouds={}, creds={}):
62
fake_yaml = {'clouds': clouds, 'credentials': creds}
63
with patch_local('configure_logging', autospec=True):
64
with patch_local('parse_args', return_value=parse_args(args)):
65
with patch_local('yaml_file_load', return_value=fake_yaml):
66
with patch_local('bootstrap_cloud_regions', autospec=True,
67
return_value=test_iterator):
68
with patch_local('print') as print_mock:
72
with self.patch_for_test([]) as print_mock:
74
self.assertEqual(0, retval)
75
print_mock.assert_called_once_with('No failures!')
77
def test_main_failures(self):
78
errors = [('config', 'region', 'one'), ('config', 'region', 'two')]
79
with self.patch_for_test(errors) as print_mock:
81
self.assertEqual(1, retval)
82
print_mock.assert_has_calls([
84
call(' * config region one'),
85
call(' * config region two'),
87
self.assertEqual(3, print_mock.call_count)
90
class TestHelpers(TestCase):
92
def test_make_logging_dir(self):
93
with temp_dir() as root_dir:
94
expected_path = os.path.join(root_dir, 'config/region')
95
log_dir = make_logging_dir(root_dir, 'config', 'region')
96
self.assertTrue(os.path.isdir(expected_path))
97
self.assertEqual(expected_path, log_dir)
99
def test_yaml_file_load(self):
100
expected_data = {'data': {'alpha': 'A', 'beta': 'B'}}
101
with temp_dir() as root_dir:
102
src_file = os.path.join(root_dir, 'test.yaml')
103
with open(src_file, 'w') as yaml_file:
104
yaml.safe_dump(expected_data, yaml_file)
105
with patch_local('get_juju_home', autospec=True,
106
return_value=root_dir) as get_home_mock:
107
data = yaml_file_load('test.yaml')
108
get_home_mock.assert_called_once_with()
109
self.assertEqual(data, expected_data)
111
def test_default_log_dir(self):
112
settings = Namespace(logs=None)
114
'deploy_stack.BootstrapManager._generate_default_clean_dir',
115
return_value='/tmp12345') as clean_dir_mock:
116
default_log_dir(settings)
117
self.assertEqual('/tmp12345', settings.logs)
118
clean_dir_mock.assert_called_once_with(_LOCAL)
120
def test_default_log_dir_provided(self):
121
settings = Namespace(logs='/tmpABCDE')
123
'deploy_stack.BootstrapManager._generate_default_clean_dir',
124
autospec=True) as clean_dir_mock:
125
default_log_dir(settings)
126
self.assertEqual('/tmpABCDE', settings.logs)
127
self.assertFalse(clean_dir_mock.called)
130
class TestPrepareCloud(FakeHomeTestCase):
132
def test_prepare_cloud(self):
133
client = fake_juju_client()
135
with patch_local('make_logging_dir', autospec=True,
136
side_effect=os.path.join):
137
with patch_local('BootstrapManager', autospec=True,
138
return_value=bs_manager) as bsm_mock:
139
with patch_local('assess_cloud_combined', autospec=True):
140
prepare_cloud('config', 'region', client, 'log_dir')
141
self.assertEqual(1, bsm_mock.call_count)
142
bsm_mock.assert_called_once_with(
143
'boot-cpc-foo-region', client, client, bootstrap_host=None,
144
machines=[], series=None, agent_url=None, agent_stream=None,
145
region='region', log_dir='log_dir/config/region', keep_env=False,
146
permanent=True, jes_enabled=True)
149
class TestIterCloudRegions(TestCase):
151
def test_iter_cloud_regions(self):
152
credentials = {'aws', 'google', 'rackspace'}
154
'aws': {'regions': ['north', 'south']},
155
'google': {'regions': ['west']},
157
regions = list(iter_cloud_regions(public_clouds, credentials))
158
self.assertEqual([('default-aws', 'north'), ('default-aws', 'south'),
159
('default-gce', 'west')], regions)
161
def test_iter_cloud_regions_credential_skip(self):
163
public_clouds = {'rackspace': {'regions': 'none'}}
164
with patch('logging.warning') as warning_mock:
165
regions = list(iter_cloud_regions(public_clouds, credentials))
166
self.assertEqual(1, warning_mock.call_count)
167
self.assertEqual([], regions)
170
class TestBootstrapCloudRegions(FakeHomeTestCase):
173
def patch_for_test(self, cloud_regions, client, error=None):
174
"""Handles all the patching for testing bootstrap_cloud_regions."""
175
with patch_local('iter_cloud_regions', autospec=True,
176
return_value=cloud_regions) as iter_mock:
177
with patch_local('prepare_cloud', autospec=True,
178
side_effect=error) as bootstrap_mock:
179
with patch_local('client_from_config', autospec=True,
180
return_value=client):
181
with patch('logging.info', autospec=True) as info_mock:
182
yield (iter_mock, bootstrap_mock, info_mock)
184
def run_test_bootstrap_cloud_regions(self, start=0, error=None):
185
pc_key = 'public_clouds'
186
cred_key = 'credentials'
187
cloud_regions = [('foo.config', 'foo'), ('bar.config', 'bar')]
188
args = Namespace(start=start, debug=True, deadline=None,
189
juju_bin='juju', logs='/tmp/log')
190
fake_client = fake_juju_client()
191
expect_calls = [call('foo.config', 'foo', fake_client, '/tmp/log'),
192
call('bar.config', 'bar', fake_client, '/tmp/log')]
193
with self.patch_for_test(cloud_regions, fake_client, error) as (
194
iter_mock, bootstrap_mock, info_mock):
195
errors = list(bootstrap_cloud_regions(pc_key, cred_key, args))
197
iter_mock.assert_called_once_with(pc_key, cred_key)
198
bootstrap_mock.assert_has_calls(expect_calls[start:])
199
self.assertEqual(len(cloud_regions) - start, info_mock.call_count)
201
self.assertEqual([], errors)
203
expect_errors = [cr + (error,) for cr in cloud_regions]
204
self.assertEqual(expect_errors, errors)
206
def test_bootstrap_cloud_regions(self):
207
self.run_test_bootstrap_cloud_regions(start=0, error=None)
209
def test_bootstrap_cloud_regions_start(self):
210
self.run_test_bootstrap_cloud_regions(start=1, error=None)
212
def test_bootstrap_cloud_regions_error(self):
213
self.run_test_bootstrap_cloud_regions(error=Exception('test'))
added
removed
tests/test_assess_public_clouds.py
