1
"""Tests for assess_constraints module."""
3
from contextlib import contextmanager
5
from mock import call, Mock, patch
9
from assess_constraints import (
11
assess_cores_constraints,
12
assess_cpu_power_constraints,
13
assess_instance_type_constraints,
14
assess_root_disk_constraints,
15
assess_virt_type_constraints,
18
deploy_charm_constraint,
19
juju_show_machine_hardware,
29
from tests.test_jujupy import fake_juju_client
36
class TestParseArgs(TestCase):
38
def test_common_args(self):
39
args = parse_args(["an-env", "/bin/juju", "/tmp/logs", "an-env-mod"])
40
self.assertEqual("an-env", args.env)
41
self.assertEqual("/bin/juju", args.juju_bin)
42
self.assertEqual("/tmp/logs", args.logs)
43
self.assertEqual("an-env-mod", args.temp_env_name)
44
self.assertEqual(False, args.debug)
47
fake_stdout = StringIO.StringIO()
48
with parse_error(self) as fake_stderr:
49
with patch("sys.stdout", fake_stdout):
50
parse_args(["--help"])
51
self.assertEqual("", fake_stderr.getvalue())
54
class TestConstraints(TestCase):
56
def test_mem_to_int(self):
57
self.assertEqual(1, mem_to_int('1'))
58
self.assertEqual(1, mem_to_int('1M'))
59
self.assertEqual(1024, mem_to_int('1G'))
60
self.assertEqual(4096, mem_to_int('4G'))
61
with self.assertRaises(JujuAssertionError):
64
def test_str_operator(self):
65
constraints = Constraints(mem='2G', root_disk='4G', virt_type='lxd')
66
self.assertEqual('mem=2G virt-type=lxd root-disk=4G',
69
def test_str_operator_none(self):
70
self.assertEqual('', str(Constraints()))
71
self.assertEqual('', str(Constraints(arch=None)))
73
def test__meets_string(self):
74
meets_string = Constraints._meets_string
75
self.assertTrue(meets_string(None, 'amd64'))
76
self.assertTrue(meets_string('amd64', 'amd64'))
77
self.assertFalse(meets_string('i32', 'amd64'))
79
def test__meets_min_int(self):
80
meets_min_int = Constraints._meets_min_int
81
self.assertTrue(meets_min_int(None, '2'))
82
self.assertFalse(meets_min_int('2', '1'))
83
self.assertTrue(meets_min_int('2', '2'))
84
self.assertTrue(meets_min_int('2', '3'))
86
def test__meets_min_mem(self):
87
meets_min_mem = Constraints._meets_min_mem
88
self.assertTrue(meets_min_mem(None, '64'))
89
self.assertFalse(meets_min_mem('1G', '512M'))
90
self.assertTrue(meets_min_mem('1G', '1024'))
91
self.assertTrue(meets_min_mem('1G', '1G'))
92
self.assertTrue(meets_min_mem('1G', '2G'))
94
def test_meets_root_disk(self):
95
constraints = Constraints(root_disk='8G')
96
self.assertTrue(constraints.meets_root_disk('8G'))
97
self.assertFalse(constraints.meets_root_disk('4G'))
99
def test_meets_cores(self):
100
constraints = Constraints(cores='2')
101
self.assertTrue(constraints.meets_cores('3'))
102
self.assertFalse(constraints.meets_cores('1'))
104
def test_meets_cpu_power(self):
105
constraints = Constraints(cpu_power='20')
106
self.assertTrue(constraints.meets_cpu_power('30'))
107
self.assertFalse(constraints.meets_cpu_power('10'))
109
def test_meets_arch(self):
110
constraints = Constraints(arch='amd64')
111
self.assertTrue(constraints.meets_arch('amd64'))
112
self.assertFalse(constraints.meets_arch('i386'))
114
def test_meets_instance_type(self):
115
constraints = Constraints(instance_type='t2.micro')
116
data1 = {'mem': '1G', 'cpu-power': '10', 'cores': '1'}
117
self.assertTrue(constraints.meets_instance_type(data1))
118
data2 = {'mem': '8G', 'cpu-power': '20', 'cores': '1'}
119
self.assertFalse(constraints.meets_instance_type(data2))
120
data3 = dict(data1, arch='amd64')
121
self.assertTrue(constraints.meets_instance_type(data3))
122
data4 = {'root-disk': '1G', 'cpu-power': '10', 'cores': '1'}
123
with self.assertRaises(JujuAssertionError):
124
constraints.meets_instance_type(data4)
126
def test_meets_instance_type_fix(self):
127
constraints = Constraints(instance_type='t2.micro')
128
data = {'mem': '1G', 'cpu-power': '10', 'cpu-cores': '1'}
129
self.assertTrue(constraints.meets_instance_type(data))
132
class TestMain(TestCase):
135
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod", "--verbose"]
136
client = Mock(spec=["is_jes_enabled"])
137
with patch("assess_constraints.configure_logging",
138
autospec=True) as mock_cl:
139
with patch("assess_constraints.BootstrapManager.booted_context",
140
autospec=True) as mock_bc:
141
with patch('deploy_stack.client_from_config',
142
return_value=client) as mock_cfc:
143
with patch("assess_constraints.assess_constraints",
144
autospec=True) as mock_assess:
146
mock_cl.assert_called_once_with(logging.DEBUG)
147
mock_cfc.assert_called_once_with('an-env', "/bin/juju", debug=False,
149
self.assertEqual(mock_bc.call_count, 1)
150
mock_assess.assert_called_once_with(client, False)
153
class TestAssess(TestCase):
156
def prepare_deploy_mock(self):
157
# Using fake_client means that deploy and get_status have plausible
158
# results. Wrapping it in a Mock causes every call to be recorded, and
159
# allows assertions to be made about calls. Mocks and the fake client
160
# can also be used separately.
161
"""Mock a client and the deploy function."""
162
fake_client = Mock(wraps=fake_juju_client())
163
fake_client.bootstrap()
164
with patch('jujupy.EnvJujuClient.deploy',
165
autospec=True) as deploy_mock:
166
yield fake_client, deploy_mock
168
def gather_constraint_args(self, mock):
169
"""Create a list of the constraint arguments passed to the mock."""
171
args[1]["constraints"] for args in mock.call_args_list]
172
return constraint_args
175
def patch_hardware(self, return_values):
176
"""Patch juju_show_machine_hardware with a series of return values."""
177
def pop_hardware(client, machine):
178
return return_values.pop(0)
179
with patch('assess_constraints.juju_show_machine_hardware',
180
autospec=True, side_effect=pop_hardware) as hardware_mock:
183
def test_virt_type_constraints_with_kvm(self):
184
assert_constraints_calls = ["virt-type=lxd", "virt-type=kvm"]
185
with self.prepare_deploy_mock() as (fake_client, deploy_mock):
186
assess_virt_type_constraints(fake_client, True)
187
constraints_calls = self.gather_constraint_args(deploy_mock)
188
self.assertEqual(constraints_calls, assert_constraints_calls)
190
def test_virt_type_constraints_without_kvm(self):
191
assert_constraints_calls = ["virt-type=lxd"]
192
with self.prepare_deploy_mock() as (fake_client, deploy_mock):
193
assess_virt_type_constraints(fake_client, False)
194
constraints_calls = self.gather_constraint_args(deploy_mock)
195
self.assertEqual(constraints_calls, assert_constraints_calls)
198
def patch_instance_spec(self, fake_provider, passing=True):
199
fake_instance_types = ['bar', 'baz']
200
bar_data = {'cpu-power': '20'}
201
baz_data = {'cpu-power': '30'}
203
def mock_get_instance_spec(instance_type):
204
if 'bar' == instance_type:
206
elif 'baz' == instance_type and passing:
208
elif 'baz' == instance_type:
209
return {'cpu-power': '40'}
211
raise ValueError('instance-type not in mock.')
212
with patch.dict(INSTANCE_TYPES, {fake_provider: fake_instance_types}):
213
with patch('assess_constraints.get_instance_spec', autospec=True,
214
side_effect=mock_get_instance_spec) as spec_mock:
215
with self.patch_hardware([bar_data, baz_data]):
216
with patch('assess_constraints.application_machines',
217
autospec=True, return_value=['0']):
220
def test_instance_type_constraints(self):
221
assert_constraints_calls = ['instance-type=bar', 'instance-type=baz']
222
with self.prepare_deploy_mock() as (fake_client, deploy_mock):
223
fake_provider = fake_client.env.config.get('type')
224
with self.patch_instance_spec(fake_provider) as spec_mock:
225
assess_instance_type_constraints(fake_client)
226
constraints_calls = self.gather_constraint_args(deploy_mock)
227
self.assertEqual(constraints_calls, assert_constraints_calls)
228
self.assertEqual(spec_mock.call_args_list, [call('bar'), call('baz')])
230
def test_instance_type_constraints_fail(self):
231
with self.prepare_deploy_mock() as (fake_client, deploy_mock):
232
fake_provider = fake_client.env.config.get('type')
233
with self.patch_instance_spec(fake_provider, False):
234
with self.assertRaisesRegexp(
236
'Test Failed: on {} with constraints "{}"'.format(
237
fake_provider, 'instance-type=baz')):
238
assess_instance_type_constraints(fake_client)
240
def test_instance_type_constraints_missing(self):
241
fake_client = Mock(wraps=fake_juju_client())
242
with self.prepare_deploy_mock() as (fake_client, deploy_mock):
243
assess_instance_type_constraints(fake_client)
244
self.assertFalse(deploy_mock.called)
246
def test_root_disk_constraints(self):
247
fake_client = Mock(wraps=fake_juju_client())
248
with patch('assess_constraints.prepare_constraint_test',
249
autospec=True, return_value={'root-disk': '8G'}
251
with self.assertRaises(JujuAssertionError):
252
assess_root_disk_constraints(fake_client, ['8G', '16G'])
253
self.assertEqual(2, prepare_mock.call_count)
255
def test_cores_constraints(self):
256
fake_client = Mock(wraps=fake_juju_client())
257
with patch('assess_constraints.prepare_constraint_test',
258
autospec=True, return_value={'cores': '2'}
260
with self.assertRaises(JujuAssertionError):
261
assess_cores_constraints(fake_client, ['2', '4'])
262
self.assertEqual(2, prepare_mock.call_count)
264
def test_cpu_power_constraints(self):
265
fake_client = Mock(wraps=fake_juju_client())
266
with patch('assess_constraints.prepare_constraint_test',
267
autospec=True, return_value={'cpu-power': '20'}
269
with self.assertRaises(JujuAssertionError):
270
assess_cpu_power_constraints(fake_client, ['10', '30'])
271
self.assertEqual(2, prepare_mock.call_count)
274
class TestDeploy(TestCase):
276
def test_deploy_constraint(self):
277
fake_client = Mock(wraps=fake_juju_client())
278
fake_client.attach_mock(Mock(), 'deploy')
279
fake_client.attach_mock(Mock(), 'wait_for_workloads')
280
charm_name = 'test-constraint'
281
charm_series = 'xenial'
282
constraints = Constraints(mem='10GB')
283
with temp_dir() as charm_dir:
284
charm = os.path.join(charm_dir, charm_series, charm_name)
285
deploy_constraint(fake_client, constraints, charm, charm_series,
287
fake_client.deploy.assert_called_once_with(
288
charm, series=charm_series, repository=charm_dir,
289
constraints=str(constraints))
290
fake_client.wait_for_workloads.assert_called_once_with()
292
def test_deploy_charm_constraint(self):
293
fake_client = Mock(wraps=fake_juju_client())
294
charm_name = 'test-constraint'
295
charm_series = 'xenial'
296
constraints = Constraints(mem='10GB')
297
with temp_dir() as charm_dir:
298
with patch('assess_constraints.deploy_constraint',
299
autospec=True) as deploy_mock:
300
deploy_charm_constraint(fake_client, constraints, charm_name,
301
charm_series, charm_dir)
302
charm = os.path.join(charm_dir, charm_series, charm_name)
303
deploy_mock.assert_called_once_with(fake_client, constraints, charm,
304
charm_series, charm_dir)
307
class TestJujuWrappers(TestCase):
309
# Dictionaries showing plausable, but reduced and pre-loaded, output.
310
SAMPLE_SHOW_MACHINE_OUTPUT = {
311
'model': {'name': 'controller'},
312
'machines': {'0': {'hardware': 'arch=amd64 cpu-cores=0 mem=0M'}},
316
SAMPLE_SHOW_MODEL_OUTPUT = {
318
'machines': {'0': 'UNUSED', '1': 'UNUSED'},
320
{'wiki': {'units': {'wiki/0': {'machine': '0'}}},
321
'mysql': {'units': {'mysql/0': {'machine': '1'}}},
324
def test_juju_show_machine_hardware(self):
325
"""Check the juju_show_machine_hardware data translation."""
326
output_mock = Mock(return_value=self.SAMPLE_SHOW_MACHINE_OUTPUT)
327
fake_client = Mock(get_juju_output=output_mock)
328
with patch('yaml.load', side_effect=lambda x: x):
329
data = juju_show_machine_hardware(fake_client, '0')
330
output_mock.assert_called_once_with('show-machine', '0',
332
self.assertEqual({'arch': 'amd64', 'cpu-cores': '0', 'mem': '0M'},
335
def test_application_machines(self):
336
output_mock = Mock(return_value=self.SAMPLE_SHOW_MODEL_OUTPUT)
337
fake_client = Mock(get_juju_output=output_mock)
338
with patch('yaml.load', side_effect=lambda x: x):
339
data = application_machines(fake_client, 'wiki')
340
output_mock.assert_called_once_with('status', '--format', 'yaml')
341
self.assertEquals(['0'], data)
added
removed
tests/test_assess_constraints.py
