1
from copy import deepcopy
2
from contextlib import contextmanager
21
import assess_container_networking as jcnet
27
from tests.test_jujupy import fake_juju_client
33
class JujuMock(EnvJujuClient):
34
"""A mock of the parts of the Juju command that the tests hit."""
36
def __init__(self, *args, **kwargs):
37
super(JujuMock, self).__init__(*args, **kwargs)
43
def add_machine(self, args):
44
if isinstance(args, tuple) and args[0] == '-n':
45
for n in range(int(args[1])):
48
self._add_machine(args)
51
def _model_state(self):
52
return self._backend.controller_state.models[self.model_name]
54
def _add_machine(self, name):
56
name = str(self.next_machine)
57
self.next_machine += 1
59
bits = name.split(':')
63
container_type = bits[0]
66
self._model_state.add_container(container_type, machine, n)
69
self._model_state.add_machine(machine_id=name)
71
def add_service(self, name, machine=0, instance_number=1):
72
self._model_state.add_unit(name)
74
def juju(self, cmd, *rargs, **kwargs):
79
if cmd != 'bootstrap':
80
self.commands.append((cmd, args))
82
if len(self._ssh_output) == 0:
86
return self._ssh_output[self._call_number()]
88
# If we ran out of values, just return the last one
89
return self._ssh_output[-1]
91
return super(JujuMock, self).juju(cmd, *rargs, **kwargs)
93
def _call_number(self):
94
call_number = self._call_n
98
def set_ssh_output(self, ssh_output):
99
self._ssh_output = deepcopy(ssh_output)
101
def reset_calls(self):
105
class TestContainerNetworking(TestCase):
107
self.client = EnvJujuClient(
108
JujuData('foo', {'type': 'local'}), '1.234-76', None)
110
self.juju_mock = fake_juju_client(cls=JujuMock)
111
self.juju_mock.bootstrap()
112
self.ssh_mock = Mock()
115
patch.object(self.client, 'juju', self.juju_mock.juju),
116
patch.object(self.client, 'get_status', self.juju_mock.get_status),
117
patch.object(self.client, 'juju_async', self.juju_mock.juju_async),
118
patch.object(self.client, 'wait_for', lambda *args, **kw: None),
119
patch.object(self.client, 'wait_for_started',
120
self.juju_mock.get_status),
121
patch.object(self.client, 'get_juju_output', self.juju_mock.juju),
124
for patcher in patches:
126
self.addCleanup(patcher.stop)
128
def assert_ssh(self, args, machine, cmd):
129
self.assertEqual(args, [('ssh', '--proxy', machine, cmd), ])
131
def test_parse_args(self):
132
# Test a simple command line that should work
133
cmdline = ['env', '/juju', 'logs', 'ten']
134
args = jcnet.parse_args(cmdline)
135
self.assertEqual(args.machine_type, None)
136
self.assertEqual(args.juju_bin, '/juju')
137
self.assertEqual(args.env, 'env')
138
self.assertEqual(args.logs, 'logs')
139
self.assertEqual(args.temp_env_name, 'ten')
140
self.assertEqual(args.debug, False)
141
self.assertEqual(args.upload_tools, False)
142
self.assertEqual(args.clean_environment, False)
144
# check the optional arguments
145
opts = ['--machine-type', jcnet.KVM_MACHINE, '--debug',
146
'--upload-tools', '--clean-environment']
147
args = jcnet.parse_args(cmdline + opts)
148
self.assertEqual(args.machine_type, jcnet.KVM_MACHINE)
149
self.assertEqual(args.debug, True)
150
self.assertEqual(args.upload_tools, True)
151
self.assertEqual(args.clean_environment, True)
153
# Now check that we can only set machine_type to kvm or lxc
154
opts = ['--machine-type', jcnet.LXC_MACHINE]
155
args = jcnet.parse_args(cmdline + opts)
156
self.assertEqual(args.machine_type, jcnet.LXC_MACHINE)
158
# Machine type can also be lxd
159
opts = ['--machine-type', jcnet.LXD_MACHINE]
160
args = jcnet.parse_args(cmdline + opts)
161
self.assertEqual(args.machine_type, jcnet.LXD_MACHINE)
163
# Set up an error (bob is invalid)
164
opts = ['--machine-type', 'bob']
165
with parse_error(self) as stderr:
166
jcnet.parse_args(cmdline + opts)
167
self.assertRegexpMatches(
169
".*error: argument --machine-type: invalid choice: 'bob'.*")
172
machine, addr = '0', 'foobar'
173
with patch.object(self.client, 'get_juju_output',
174
autospec=True) as ssh_mock:
175
jcnet.ssh(self.client, machine, addr)
176
self.assertEqual(1, ssh_mock.call_count)
177
self.assert_ssh(ssh_mock.call_args, machine, addr)
179
def test_find_network(self):
180
machine, addr = '0', '1.1.1.1'
181
self.assertRaisesRegexp(
182
ValueError, "Unable to find route to '1.1.1.1'",
183
jcnet.find_network, self.client, machine, addr)
185
self.juju_mock.set_ssh_output([
186
'default via 192.168.0.1 dev eth3\n'
187
'1.1.1.0/24 dev eth3 proto kernel scope link src 1.1.1.22',
189
self.juju_mock.commands = []
190
jcnet.find_network(self.client, machine, addr)
191
self.assertItemsEqual(self.juju_mock.commands,
194
'ip route show to match ' + addr))])
196
def test_clean_environment(self):
197
self.juju_mock.add_machine('1')
198
self.juju_mock.add_machine('2')
199
self.juju_mock.add_service('name')
201
jcnet.clean_environment(self.client)
202
self.assertItemsEqual(self.juju_mock.commands, [
203
('remove-application', ('name',)),
204
('remove-machine', '1'),
205
('remove-machine', '2'),
208
def test_clean_environment_with_containers(self):
209
self.juju_mock.add_machine('0')
210
self.juju_mock.add_machine('1')
211
self.juju_mock.add_machine('2')
212
self.juju_mock.add_machine('lxc:0')
213
self.juju_mock.add_machine('kvm:0')
215
jcnet.clean_environment(self.client)
216
self.assertItemsEqual(self.juju_mock.commands, [
217
('remove-machine', '0/lxc/0'),
218
('remove-machine', '0/kvm/0'),
219
('remove-machine', '1'),
220
('remove-machine', '2')
223
def test_clean_environment_just_services(self):
224
self.juju_mock.add_machine('1')
225
self.juju_mock.add_machine('2')
226
self.juju_mock.add_machine('lxc:0')
227
self.juju_mock.add_machine('kvm:0')
228
self.juju_mock.add_service('name')
230
jcnet.clean_environment(self.client, True)
231
self.assertEqual(self.juju_mock.commands, [
232
('remove-application', ('name',)),
235
def test_make_machines(self):
236
hosts, containers = jcnet.make_machines(
237
self.client, [jcnet.LXC_MACHINE, jcnet.KVM_MACHINE])
238
self.assertEqual(hosts, ['0', '1'])
240
'kvm': {'0': ['0/kvm/1', '0/kvm/0'],
242
'lxc': {'0': ['0/lxc/0', '0/lxc/1'],
245
self.assertDictEqual(containers, expected)
247
hosts, containers = jcnet.make_machines(
248
self.client, [jcnet.LXC_MACHINE])
249
self.assertEqual(hosts, ['0', '1'])
251
'lxc': {'0': ['0/lxc/0', '0/lxc/1'],
254
self.assertDictEqual(containers, expected)
256
hosts, containers = jcnet.make_machines(
257
self.client, [jcnet.KVM_MACHINE])
258
self.assertEqual(hosts, ['0', '1'])
260
'kvm': {'0': ['0/kvm/1', '0/kvm/0'],
263
self.assertDictEqual(containers, expected)
265
def test_test_network_traffic(self):
266
targets = ['0/lxc/0', '0/lxc/1']
267
self.juju_mock._model_state.add_machine()
268
self.juju_mock._model_state.add_container('lxc', '0')
270
with patch('assess_container_networking.get_random_string',
271
lambda *args, **kw: 'hello'):
273
self.juju_mock.set_ssh_output(['', '', 'hello'])
274
jcnet.assess_network_traffic(self.client, targets)
276
self.juju_mock.reset_calls()
277
self.juju_mock.set_ssh_output(['', '', 'fail'])
278
self.assertRaisesRegexp(
279
ValueError, "Wrong or missing message: 'fail'",
280
jcnet.assess_network_traffic, self.client, targets)
282
def test_test_address_range(self):
283
targets = ['0/lxc/0', '0/lxc/1']
284
self.juju_mock._model_state.add_machine()
285
self.juju_mock._model_state.add_container('lxc', '0')
286
self.juju_mock._model_state.add_container('lxc', '0')
287
self.juju_mock.set_ssh_output([
288
'default via 192.168.0.1 dev eth3',
289
'2: eth3 inet 192.168.0.22/24 brd 192.168.0.255 scope '
290
'global eth3\ valid_lft forever preferred_lft forever',
294
jcnet.assess_address_range(self.client, targets)
296
def test_test_address_range_fail(self):
297
targets = ['0/lxc/0', '0/lxc/1']
298
self.juju_mock._model_state.add_machine()
299
self.juju_mock._model_state.add_container('lxc', '0')
300
self.juju_mock.set_ssh_output([
301
'default via 192.168.0.1 dev eth3',
302
'2: eth3 inet 192.168.0.22/24 brd 192.168.0.255 scope '
303
'global eth3\ valid_lft forever preferred_lft forever',
310
self.assertRaisesRegexp(
311
ValueError, '0/lxc/0 \S+ not on the same subnet as 0 \S+',
312
jcnet.assess_address_range, self.client, targets)
314
def test_test_internet_connection(self):
315
targets = ['0/lxc/0', '0/lxc/1']
316
model_state = self.juju_mock._model_state
317
model_state.add_machine(host_name='0-dns-name')
318
model_state.add_container('lxc', '0')
319
model_state.add_container('lxc', '0')
321
# Can ping default route
322
self.juju_mock.set_ssh_output([
323
'default via 192.168.0.1 dev eth3', 0,
324
'default via 192.168.0.1 dev eth3', 0])
325
jcnet.assess_internet_connection(self.client, targets)
327
# Can't ping default route
328
self.juju_mock.set_ssh_output([
329
'default via 192.168.0.1 dev eth3', 1])
330
self.juju_mock.reset_calls()
331
self.assertRaisesRegexp(
332
ValueError, "0/lxc/0 unable to ping default route",
333
jcnet.assess_internet_connection, self.client, targets)
335
# Can't find default route
336
self.juju_mock.set_ssh_output(['', 1])
337
self.juju_mock.reset_calls()
338
self.assertRaisesRegexp(
339
ValueError, "Default route not found",
340
jcnet.assess_internet_connection, self.client, targets)
342
def test_private_address(self):
343
ssh_results = ["default via 10.0.30.1 dev br-eth1",
344
"5: br-eth1 inet 10.0.30.24/24 brd "
345
"10.0.30.255 scope global br-eth1 "
346
"valid_lft forever preferred_lft forever"]
347
fake_client = object()
348
with patch("assess_container_networking.ssh",
349
autospec=True, side_effect=ssh_results) as mock_ssh:
350
result = jcnet.private_address(fake_client, "machine.test")
351
self.assertEqual(result, "10.0.30.24")
352
self.assertEqual(mock_ssh.mock_calls,
353
[call(fake_client, "machine.test",
354
"ip -4 -o route list 0/0"),
355
call(fake_client, "machine.test",
356
"ip -4 -o addr show br-eth1")])
359
class TestMain(FakeHomeTestCase):
362
def patch_main(self, argv, client, log_level, debug=False):
363
env = SimpleEnvironment(argv[0], {"type": "ec2"})
365
with patch("assess_container_networking.configure_logging",
366
autospec=True) as mock_cl:
367
with patch("deploy_stack.client_from_config",
368
return_value=client) as mock_c:
370
mock_cl.assert_called_once_with(log_level)
371
mock_c.assert_called_once_with('an-env', argv[1], debug=debug,
375
def patch_bootstrap_manager(self, runs=True):
376
with patch("deploy_stack.BootstrapManager.top_context",
377
autospec=True) as mock_tc:
378
with patch("deploy_stack.BootstrapManager.bootstrap_context",
379
autospec=True) as mock_bc:
380
with patch("deploy_stack.BootstrapManager.runtime_context",
381
autospec=True) as mock_rc:
383
self.assertEqual(mock_tc.call_count, 1)
385
self.assertEqual(mock_rc.call_count, 1)
387
def test_bootstrap_required(self):
388
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod", "--verbose"]
389
client = Mock(spec=["bootstrap", "enable_feature", "is_jes_enabled"])
390
client.supported_container_types = frozenset([KVM_MACHINE,
392
with patch("assess_container_networking.assess_container_networking",
393
autospec=True) as mock_assess:
394
with self.patch_bootstrap_manager() as mock_bc:
395
with self.patch_main(argv, client, logging.DEBUG):
396
ret = jcnet.main(argv)
397
client.bootstrap.assert_called_once_with(False)
398
self.assertEqual("", self.log_stream.getvalue())
399
self.assertEqual(mock_bc.call_count, 1)
400
mock_assess.assert_called_once_with(client, [KVM_MACHINE, "lxc"])
401
self.assertEqual(ret, 0)
403
def test_clean_existing_env(self):
404
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod",
405
"--clean-environment"]
406
client = Mock(spec=["enable_feature", "env", "get_status",
407
"is_jes_enabled", "wait_for", "wait_for_started"])
408
client.supported_container_types = frozenset([KVM_MACHINE,
410
client.get_status.return_value = Status.from_text("""
415
with patch("assess_container_networking.assess_container_networking",
416
autospec=True) as mock_assess:
417
with self.patch_bootstrap_manager() as mock_bc:
418
with self.patch_main(argv, client, logging.INFO):
419
ret = jcnet.main(argv)
420
self.assertEqual(client.env.environment, "an-env-mod")
421
self.assertEqual("", self.log_stream.getvalue())
422
self.assertEqual(mock_bc.call_count, 0)
423
mock_assess.assert_called_once_with(client, ["kvm", "lxc"])
424
self.assertEqual(ret, 0)
426
def test_clean_missing_env(self):
427
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod",
428
"--clean-environment"]
429
client = Mock(spec=["bootstrap", "enable_feature", "env", "get_status",
430
"is_jes_enabled", "wait_for", "wait_for_started"])
431
client.supported_container_types = frozenset([KVM_MACHINE,
433
client.get_status.side_effect = [
434
Exception("Timeout"),
441
with patch("assess_container_networking.assess_container_networking",
442
autospec=True) as mock_assess:
443
with self.patch_bootstrap_manager() as mock_bc:
444
with self.patch_main(argv, client, logging.INFO):
445
ret = jcnet.main(argv)
446
self.assertEqual(client.env.environment, "an-env-mod")
447
client.bootstrap.assert_called_once_with(False)
449
"INFO Could not clean existing env: Timeout\n",
450
self.log_stream.getvalue())
451
self.assertEqual(mock_bc.call_count, 1)
452
mock_assess.assert_called_once_with(client, ["kvm", "lxc"])
453
self.assertEqual(ret, 0)
455
def test_final_clean_fails(self):
456
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod",
457
"--clean-environment"]
458
client = Mock(spec=["enable_feature", "env", "get_status",
459
"is_jes_enabled", "wait_for", "wait_for_started"])
460
client.supported_container_types = frozenset([KVM_MACHINE,
462
client.get_status.side_effect = [
468
Exception("Timeout"),
470
with patch("assess_container_networking.assess_container_networking",
471
autospec=True) as mock_assess:
472
with self.patch_bootstrap_manager() as mock_bc:
473
with self.patch_main(argv, client, logging.INFO):
474
ret = jcnet.main(argv)
475
self.assertEqual(client.env.environment, "an-env-mod")
477
"INFO Could not clean existing env: Timeout\n",
478
self.log_stream.getvalue())
479
self.assertEqual(mock_bc.call_count, 0)
480
mock_assess.assert_called_once_with(client, ["kvm", "lxc"])
481
self.assertEqual(ret, 1)
483
def test_lxd_unsupported_on_juju_1(self):
484
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod", "--verbose",
485
"--machine-type=lxd"]
486
client = Mock(spec=["bootstrap", "enable_feature", "is_jes_enabled"])
487
client.version = "1.25.5"
488
client.supported_container_types = frozenset([LXC_MACHINE,
490
with self.patch_main(argv, client, logging.DEBUG):
491
with self.assertRaises(Exception) as exc_ctx:
494
str(exc_ctx.exception),
495
"no lxd support on juju 1.25.5")
496
self.assertEqual(client.bootstrap.call_count, 0)
497
self.assertEqual("", self.log_stream.getvalue())
499
def test_lxd_tested_on_juju_2(self):
500
argv = ["an-env", "/bin/juju", "/tmp/logs", "an-env-mod", "--verbose"]
501
client = Mock(spec=["bootstrap", "enable_feature", "is_jes_enabled"])
502
client.supported_container_types = frozenset([
503
LXD_MACHINE, KVM_MACHINE, LXC_MACHINE])
504
with patch("assess_container_networking.assess_container_networking",
505
autospec=True) as mock_assess:
506
with self.patch_bootstrap_manager() as mock_bc:
507
with self.patch_main(argv, client, logging.DEBUG):
508
ret = jcnet.main(argv)
509
client.bootstrap.assert_called_once_with(False)
510
self.assertEqual("", self.log_stream.getvalue())
511
self.assertEqual(mock_bc.call_count, 1)
512
mock_assess.assert_called_once_with(client, [
513
KVM_MACHINE, LXC_MACHINE, LXD_MACHINE])
514
self.assertEqual(ret, 0)
added
removed
tests/test_assess_container_networking.py
