1
from collections import namedtuple
13
from tests import TestCase
14
from winazurearm import (
16
DEFAULT_RESOURCE_PREFIX,
28
'AZURE_SUBSCRIPTION_ID': 'subscription_id',
29
'AZURE_CLIENT_ID': 'client_id',
30
'AZURE_SECRET': 'secret',
31
'AZURE_TENANT': 'tenant',
34
# The azure unit test use namedtuples like these to avoid tight
35
# coupling between services. When Azure ARM is stable, we might want
36
# to use the real objects.
37
ResourceGroup = namedtuple('ResourceGroup', ['name'])
38
StorageAccount = namedtuple('StorageAccount', ['name', 'creation_time'])
39
VirtualMachine = namedtuple('VirtualMachine', ['name', 'vm_id'])
40
Network = namedtuple('Network', ['name'])
41
Address = namedtuple('Address', ['name', 'ip_address'])
46
def __init__(self, result_data=None):
48
self.result_data = result_data
55
raise Exception('It is an error to call result after done.')
57
return self.result_data
60
def fake_init_services(client):
61
"""Repurpose the lazy init to install mocks."""
62
# client.resource.resource_groups.list()
63
client.resource = Mock(resource_groups=Mock(
64
list=Mock(return_value=[]),
65
delete=Mock(return_value=None)))
66
# client.storage.storage_accounts.list_by_resource_group()
67
client.storage = Mock(
68
storage_accounts=Mock(list_by_resource_group=Mock(return_value=[])))
69
# client.compute.virtual_machines.list()
70
client.compute = Mock(virtual_machines=Mock(
71
list=Mock(return_value=[]),
72
delete=Mock(return_value=None)))
73
# client.network.public_ip_addresses.list()
74
# client.network.virtual_networks.list()
75
client.network = Mock(
76
public_ip_addresses=Mock(list=Mock(return_value=[])),
77
virtual_networks=Mock(list=Mock(return_value=[])))
80
@patch('winazurearm.ARMClient.init_services',
81
autospec=True, side_effect=fake_init_services)
82
class WinAzureARMTestCase(TestCase):
84
test_environ = AZURE_ENVIRON
86
def test_main_list_resources(self, is_mock):
87
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
88
with patch('winazurearm.list_resources', autospec=True) as lr_mock:
89
code = main(['winazurearm.py', 'list-resources', 'juju-deploy*'])
90
self.assertEqual(0, code)
91
self.assertEqual(1, is_mock.call_count)
92
lr_mock.assert_called_once_with(
93
client, glob='juju-deploy*', print_out=True, recursive=False)
95
def test_main_delete_resources(self, is_mock):
96
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
97
with patch('winazurearm.delete_resources', autospec=True) as dr_mock:
98
code = main(['winazurearm.py', 'delete-resources', 'juju-deploy*'])
99
self.assertEqual(0, code)
100
self.assertEqual(1, is_mock.call_count)
101
dr_mock.assert_called_once_with(
102
client, glob='juju-deploy*', old_age=OLD_MACHINE_AGE)
104
def test_main_delete_resources_old_age(self, is_mock):
105
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
106
with patch('winazurearm.delete_resources', autospec=True) as dr_mock:
107
code = main(['winazurearm.py', 'delete-resources',
108
'-o', '2', 'juju-deploy*'])
109
self.assertEqual(0, code)
110
self.assertEqual(1, is_mock.call_count)
111
dr_mock.assert_called_once_with(
112
client, glob='juju-deploy*', old_age=2)
114
def test_main_delete_instance_instance_id(self, is_mock):
115
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
116
with patch('winazurearm.delete_instance', autospec=True) as di_mock:
117
code = main(['winazurearm.py', 'delete-instance', 'instance-id'])
118
self.assertEqual(0, code)
119
self.assertEqual(1, is_mock.call_count)
120
di_mock.assert_called_once_with(
121
client, 'instance-id', resource_group=None)
123
def test_main_delete_instance_not_found(self, is_mock):
124
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
125
with patch('winazurearm.delete_instance', autospec=True,
126
side_effect=ValueError) as di_mock:
127
code = main(['winazurearm.py', 'delete-instance', 'instance-id'])
128
self.assertEqual(1, code)
129
self.assertEqual(1, is_mock.call_count)
130
di_mock.assert_called_once_with(
131
client, 'instance-id', resource_group=None)
133
def test_main_delete_instance_name_group(self, is_mock):
134
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
135
with patch('winazurearm.delete_instance', autospec=True) as di_mock:
136
code = main(['winazurearm.py', 'delete-instance', 'name', 'group'])
137
self.assertEqual(0, code)
138
self.assertEqual(1, is_mock.call_count)
139
di_mock.assert_called_once_with(client, 'name', resource_group='group')
141
def test_list_resources(self, is_mock):
142
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
143
client.init_services()
144
groups = [ResourceGroup('juju-foo-0'), ResourceGroup('juju-bar-1')]
145
client.resource.resource_groups.list.return_value = groups
146
result = list_resources(client, 'juju-bar*')
147
rgd = ResourceGroupDetails(client, groups[-1])
148
self.assertEqual([rgd], result)
149
client.resource.resource_groups.list.assert_called_once_with()
151
def test_list_resources_ignore_default(self, is_mock):
152
# Default resources are created by Azure. They should only be
153
# seen via the UI. A glob for everything will still ignore defaults.
154
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
155
client.init_services()
156
groups = [ResourceGroup('{}-network'.format(DEFAULT_RESOURCE_PREFIX))]
157
client.resource.resource_groups.list.return_value = groups
158
result = list_resources(client, '*')
159
self.assertEqual([], result)
161
def test_list_resources_recursive(self, is_mock):
162
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
163
client.init_services()
164
# For the call to find many groups.
165
a_group = ResourceGroup('juju-bar-1')
166
b_group = ResourceGroup('juju-foo-0')
167
client.resource.resource_groups.list.return_value = [a_group, b_group]
168
# For the call to load a ResourceGroupDetails instance.
169
storage_account = StorageAccount('abcd-12', datetime.now(tz=pytz.UTC))
170
client.storage.storage_accounts.list_by_resource_group.return_value = [
172
virtual_machine = VirtualMachine('admin-machine-0', 'bcde-1234')
173
client.compute.virtual_machines.list.return_value = [virtual_machine]
174
address = Address('machine-0-public-ip', '1.2.3.4')
175
client.network.public_ip_addresses.list.return_value = [address]
176
network = Network('juju-bar-network-1')
177
client.network.virtual_networks.list.return_value = [network]
178
# The addition of recursive=True will get the details of the
179
# subordinate resources and set is_loaded to True.
180
result = list_resources(client, 'juju-bar*', recursive=True)
181
rgd = ResourceGroupDetails(
182
client, a_group, storage_accounts=[storage_account],
183
vms=[virtual_machine], addresses=[address], networks=[network])
185
self.assertEqual([rgd], result)
187
def test_delete_resources_found_old(self, is_mock):
188
now = datetime.now(tz=pytz.utc)
189
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
190
client.init_services()
191
group = ResourceGroup('juju-bar-1')
192
client.resource.resource_groups.list.return_value = [group]
193
# The resource_groups's storage_account is 4 hours old.
194
storage_account = StorageAccount('abcd-12', now - timedelta(hours=4))
195
client.storage.storage_accounts.list_by_resource_group.return_value = [
197
poller = FakePoller()
198
client.resource.resource_groups.delete.return_value = poller
199
# Delete resource groups that are 2 hours old.
200
count = delete_resources(client, 'juju-bar*', old_age=2, now=now)
201
self.assertEqual(1, count)
202
client.resource.resource_groups.delete.assert_called_once_with(
204
self.assertIs(True, poller.is_done)
206
def test_delete_resources_not_found_old(self, is_mock):
207
now = datetime.now(tz=pytz.utc)
208
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
209
client.init_services()
210
group = ResourceGroup('juju-bar-1')
211
client.resource.resource_groups.list.return_value = [group]
212
# The resource_groups's storage_account is 2 hours old.
213
storage_account = StorageAccount('abcd-12', now - timedelta(hours=2))
214
client.storage.storage_accounts.list_by_resource_group.return_value = [
216
# Delete resource groups that are 8 hours old.
217
count = delete_resources(client, 'juju-bar*', old_age=8, now=now)
218
self.assertEqual(0, count)
219
self.assertEqual(0, client.resource.resource_groups.delete.call_count)
221
def test_delete_resources_read_only(self, is_mock):
222
now = datetime.now(tz=pytz.utc)
224
'subscription_id', 'client_id', 'secret', 'tenant', read_only=True)
225
client.init_services()
226
group = ResourceGroup('juju-bar-1')
227
client.resource.resource_groups.list.return_value = [group]
228
# The resource_groups's storage_account is 4 hours old.
229
storage_account = StorageAccount('abcd-12', now - timedelta(hours=4))
230
client.storage.storage_accounts.list_by_resource_group.return_value = [
232
count = delete_resources(client, 'juju-bar*', old_age=2, now=now)
233
self.assertEqual(0, count)
234
self.assertEqual(0, client.resource.resource_groups.delete.call_count)
236
def test_delete_resources_poller_already_done(self, is_mock):
237
now = datetime.now(tz=pytz.utc)
238
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
239
client.init_services()
240
group = ResourceGroup('juju-bar-1')
241
client.resource.resource_groups.list.return_value = [group]
242
# The resource_groups's storage_account is 4 hours old.
243
storage_account = StorageAccount('abcd-12', now - timedelta(hours=4))
244
client.storage.storage_accounts.list_by_resource_group.return_value = [
246
poller = FakePoller()
247
poller.is_done = True
248
client.resource.resource_groups.delete.return_value = poller
249
count = delete_resources(client, 'juju-bar*', old_age=2, now=now)
250
self.assertEqual(1, count)
251
self.assertEqual(1, client.resource.resource_groups.delete.call_count)
253
def test_delete_resources_poller_is_none(self, is_mock):
254
now = datetime.now(tz=pytz.utc)
255
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
256
client.init_services()
257
group = ResourceGroup('juju-bar-1')
258
client.resource.resource_groups.list.return_value = [group]
259
# The resource_groups's storage_account is 4 hours old.
260
storage_account = StorageAccount('abcd-12', now - timedelta(hours=4))
261
client.storage.storage_accounts.list_by_resource_group.return_value = [
263
count = delete_resources(client, 'juju-bar*', old_age=2, now=now)
264
self.assertEqual(1, count)
265
self.assertEqual(1, client.resource.resource_groups.delete.call_count)
267
def test_delete_resources_old_age_0(self, is_mock):
268
now = datetime.now(tz=pytz.utc)
269
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
270
client.init_services()
271
a_group = ResourceGroup('juju-bar-1')
272
b_group = ResourceGroup('juju-foo-0')
273
client.resource.resource_groups.list.return_value = [a_group, b_group]
274
poller = FakePoller()
275
client.resource.resource_groups.delete.return_value = poller
276
# Delete resource groups that are 0 hours old.
277
# All matched resource_groups are deleted
278
count = delete_resources(client, 'juju-bar*', old_age=0, now=now)
279
self.assertEqual(1, count)
280
self.assertEqual(1, client.resource.resource_groups.delete.call_count)
281
self.assertIs(True, poller.is_done)
283
def test_delete_resources_only_network(self, is_mock):
284
now = datetime.now(tz=pytz.utc)
285
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
286
client.init_services()
287
group = ResourceGroup('juju-bar-1')
288
client.resource.resource_groups.list.return_value = [group]
289
# The resource_group has a network, but nothing else
290
network = Network('juju-bar-network-1')
291
client.network.virtual_networks.list.return_value = [network]
292
poller = FakePoller()
293
client.resource.resource_groups.delete.return_value = poller
294
count = delete_resources(client, 'juju-bar*', old_age=2, now=now)
295
self.assertEqual(1, count)
296
self.assertEqual(1, client.resource.resource_groups.delete.call_count)
298
def test_find_vm_instance(self, is_mock):
299
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
300
client.init_services()
301
rgd1 = ResourceGroupDetails(
302
client, ResourceGroup('one'),
303
vms=[VirtualMachine('name-0', 'id-a'),
304
VirtualMachine('name-1', 'id-b')])
305
rgd2 = ResourceGroupDetails(
306
client, ResourceGroup('two'),
307
vms=[VirtualMachine('name-0', 'id-y'),
308
VirtualMachine('name-1', 'id-z')])
309
resources = [rgd1, rgd2]
311
(None, None), find_vm_instance(resources, 'id-q', None))
313
(rgd2, rgd2.vms[0]), find_vm_instance(resources, 'id-y', None))
315
(rgd2, rgd2.vms[0]), find_vm_instance(resources, 'name-0', 'two'))
317
(None, None), find_vm_instance(resources, 'name-0', 'three'))
319
(None, None), find_vm_instance(resources, 'name-9', 'two'))
321
def test_delete_instance_with_name_and_resource_group(self, is_mock):
322
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
323
client.init_services()
324
poller = FakePoller()
325
client.compute.virtual_machines.delete.return_value = poller
326
vm1 = VirtualMachine('name-0', 'id-a')
327
rgd1 = ResourceGroupDetails(client, ResourceGroup('one'), vms=[vm1])
328
with patch('winazurearm.list_resources', autospec=True,
329
return_value=[rgd1]) as lr_mock:
330
# Passing just a name and group will take the group glob path.
331
delete_instance(client, 'name-0', 'one')
332
lr_mock.assert_called_once_with(client, glob='one', recursive=True)
333
client.compute.virtual_machines.delete.assert_called_once_with(
335
self.assertIs(True, poller.is_done)
337
def test_delete_instance_with_id(self, is_mock):
338
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
339
client.init_services()
340
poller = FakePoller()
341
client.compute.virtual_machines.delete.return_value = poller
342
vm1 = VirtualMachine('name-0', 'id-a')
343
rgd1 = ResourceGroupDetails(client, ResourceGroup('one'), vms=[vm1])
344
vm2 = VirtualMachine('name-0', 'id-b')
345
rgd2 = ResourceGroupDetails(client, ResourceGroup('two'), vms=[vm2])
346
with patch('winazurearm.list_resources', autospec=True,
347
return_value=[rgd1, rgd2]) as lr_mock:
348
# Passing just an id will take the * glob path.
349
delete_instance(client, 'id-a')
350
lr_mock.assert_called_once_with(client, glob='*', recursive=True)
351
client.compute.virtual_machines.delete.assert_called_once_with(
353
self.assertIs(True, poller.is_done)
355
def test_delete_instance_without_match(self, is_mock):
356
client = ARMClient('subscription_id', 'client_id', 'secret', 'tenant')
357
client.init_services()
358
vm1 = VirtualMachine('name-0', 'id-a')
359
rgd1 = ResourceGroupDetails(client, ResourceGroup('one'), vms=[vm1])
360
with patch('winazurearm.list_resources', autospec=True,
361
return_value=[rgd1]):
362
# Passing an non-existent id bypasses the call to delete.
363
with self.assertRaises(ValueError):
364
delete_instance(client, 'id-z')
365
self.assertEqual(0, client.compute.virtual_machines.delete.call_count)
added
removed
tests/test_winazurearm.py
