test.py 17.4 KB
Newer Older
1
# Copyright 2013 GRNET S.A. All rights reserved.
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
#
# Redistribution and use in source and binary forms, with or
# without modification, are permitted provided that the following
# conditions are met:
#
#   1. Redistributions of source code must retain the above
#      copyright notice, this list of conditions and the following
#      disclaimer.
#
#   2. Redistributions in binary form must reproduce the above
#      copyright notice, this list of conditions and the following
#      disclaimer in the documentation and/or other materials
#      provided with the distribution.
#
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
# POSSIBILITY OF SUCH DAMAGE.
#
# The views and conclusions contained in the software and
# documentation are those of the authors and should not be
# interpreted as representing official policies, either expressed
# or implied, of GRNET S.A.
33

34
from mock import patch, call
35
from unittest import TestCase
36
from itertools import product
37

38
from kamaki.clients import ClientError, cyclades
39
40
41
42
43
44
45
46

img_ref = "1m4g3-r3f3r3nc3"
vm_name = "my new VM"
fid = 42
vm_recv = dict(server=dict(
    status="BUILD",
    updated="2013-03-01T10:04:00.637152+00:00",
    hostId="",
47
48
    name=vm_name,
    imageRef=img_ref,
49
50
51
52
53
54
55
    created="2013-03-01T10:04:00.087324+00:00",
    flavorRef=fid,
    adminPass="n0n3sh@11p@55",
    suspended=False,
    progress=0,
    id=31173,
    metadata=dict(values=dict(os="debian", users="root"))))
56
57
58
vm_list = dict(servers=dict(values=[
    dict(name='n1', id=1),
    dict(name='n2', id=2)]))
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
net_send = dict(network=dict(dhcp=False, name='someNet'))
net_recv = dict(network=dict(
    status="PENDING",
    updated="2013-03-05T15:04:51.758780+00:00",
    name="someNet",
    created="2013-03-05T15:04:51.758728+00:00",
    cidr6=None,
    id="2130",
    gateway6=None,
    public=False,
    dhcp=False,
    cidr="192.168.1.0/24",
    type="MAC_FILTERED",
    gateway=None,
    attachments=dict(values=[dict(name='att1'), dict(name='att2')])))
74
75
76
77
net_list = dict(networks=dict(values=[
    dict(id=1, name='n1'),
    dict(id=2, name='n2'),
    dict(id=3, name='n3')]))
78
79
firewalls = dict(attachments=dict(values=[
    dict(firewallProfile='50m3_pr0f1L3', otherStuff='57uff')]))
80
81


82
83
84
85
86
87
88
89
class FR(object):
    """FR stands for Fake Response"""
    json = vm_recv
    headers = {}
    content = json
    status = None
    status_code = 200

90
rest_pkg = 'kamaki.clients.cyclades.CycladesRestClient'
91
92
93
cyclades_pkg = 'kamaki.clients.cyclades.CycladesClient'


94
class CycladesRestClient(TestCase):
95
96
97
98
99

    """Set up a Cyclades thorough test"""
    def setUp(self):
        self.url = 'http://cyclades.example.com'
        self.token = 'cyc14d3s70k3n'
100
        self.client = cyclades.CycladesRestClient(self.url, self.token)
101
102
103
104
105
106

    def tearDown(self):
        FR.json = vm_recv

    @patch('%s.set_param' % rest_pkg)
    @patch('%s.get' % rest_pkg, return_value=FR())
107
    def test_servers_get(self, get, SP):
108
        for args in product(
109
                ('', 'vm_id'),
110
111
112
113
114
                ('', 'cmd'),
                (200, 204),
                (None, '50m3-d473'),
                ({}, {'k': 'v'})):
            (srv_id, command, success, changes_since, kwargs) = args
115
            self.client.servers_get(*args[:4], **kwargs)
116
117
118
            srv_str = '/%s' % srv_id if srv_id else ''
            cmd_str = '/%s' % command if command else ''
            self.assertEqual(get.mock_calls[-1], call(
119
                '/servers%s%s' % (srv_str, cmd_str),
120
121
122
123
124
125
126
                success=success,
                **kwargs))
            if changes_since:
                self.assertEqual(
                    SP.mock_calls[-1],
                    call('changes-since', changes_since, changes_since))

127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
    @patch('%s.get' % rest_pkg, return_value=FR())
    def test_networks_get(self, get):
        for args in product(
                ('', 'net_id'),
                ('', 'cmd'),
                (200, 204),
                ({}, {'k': 'v'})):
            (srv_id, command, success, kwargs) = args
            self.client.networks_get(*args[:3], **kwargs)
            srv_str = '/%s' % srv_id if srv_id else ''
            cmd_str = '/%s' % command if command else ''
            self.assertEqual(get.mock_calls[-1], call(
                '/networks%s%s' % (srv_str, cmd_str),
                success=success,
                **kwargs))

    @patch('%s.delete' % rest_pkg, return_value=FR())
    def test_networks_delete(self, delete):
        for args in product(
                ('', 'net_id'),
                ('', 'cmd'),
                (202, 204),
                ({}, {'k': 'v'})):
            (srv_id, command, success, kwargs) = args
            self.client.networks_delete(*args[:3], **kwargs)
            srv_str = '/%s' % srv_id if srv_id else ''
            cmd_str = '/%s' % command if command else ''
            self.assertEqual(delete.mock_calls[-1], call(
                '/networks%s%s' % (srv_str, cmd_str),
                success=success,
                **kwargs))
158

159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
    @patch('%s.set_header' % rest_pkg)
    @patch('%s.post' % rest_pkg, return_value=FR())
    def test_networks_post(self, post, SH):
        from json import dumps
        for args in product(
                ('', 'net_id'),
                ('', 'cmd'),
                (None, [dict(json="data"), dict(data="json")]),
                (202, 204),
                ({}, {'k': 'v'})):
            (srv_id, command, json_data, success, kwargs) = args
            self.client.networks_post(*args[:4], **kwargs)
            vm_str = '/%s' % srv_id if srv_id else ''
            cmd_str = '/%s' % command if command else ''
            if json_data:
                json_data = dumps(json_data)
                self.assertEqual(SH.mock_calls[-2:], [
                    call('Content-Type', 'application/json'),
                    call('Content-Length', len(json_data))])
            self.assertEqual(post.mock_calls[-1], call(
                '/networks%s%s' % (vm_str, cmd_str),
                data=json_data, success=success,
                **kwargs))

183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
    @patch('%s.set_header' % rest_pkg)
    @patch('%s.put' % rest_pkg, return_value=FR())
    def test_networks_put(self, put, SH):
        from json import dumps
        for args in product(
                ('', 'net_id'),
                ('', 'cmd'),
                (None, [dict(json="data"), dict(data="json")]),
                (202, 204),
                ({}, {'k': 'v'})):
            (srv_id, command, json_data, success, kwargs) = args
            self.client.networks_put(*args[:4], **kwargs)
            vm_str = '/%s' % srv_id if srv_id else ''
            cmd_str = '/%s' % command if command else ''
            if json_data:
                json_data = dumps(json_data)
                self.assertEqual(SH.mock_calls[-2:], [
                    call('Content-Type', 'application/json'),
                    call('Content-Length', len(json_data))])
            self.assertEqual(put.mock_calls[-1], call(
                '/networks%s%s' % (vm_str, cmd_str),
                data=json_data, success=success,
                **kwargs))

207

208
class CycladesClient(TestCase):
209

210
    def assert_dicts_are_equal(self, d1, d2):
211
212
213
        for k, v in d1.items():
            self.assertTrue(k in d2)
            if isinstance(v, dict):
214
                self.assert_dicts_are_equal(v, d2[k])
215
216
217
            else:
                self.assertEqual(unicode(v), unicode(d2[k]))

218
219
    """Set up a Cyclades thorough test"""
    def setUp(self):
220
221
        self.url = 'http://cyclades.example.com'
        self.token = 'cyc14d3s70k3n'
222
        self.client = cyclades.CycladesClient(self.url, self.token)
223
224

    def tearDown(self):
225
226
        FR.status_code = 200
        FR.json = vm_recv
227

228
229
    @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
    def test_list_servers(self, SG):
230
        FR.json = vm_list
231
232
233
234
235
236
237
238
239
        for detail, since in ((0, 0), (True, 0), (0, 'd473'), (True, 'd473')):
            r = self.client.list_servers(detail=detail, changes_since=since)
            self.assertEqual(SG.mock_calls[-1], call(
                command='detail' if detail else '',
                changes_since=since))
            expected = vm_list['servers']['values']
            for i, vm in enumerate(r):
                self.assert_dicts_are_equal(vm, expected[i])
            self.assertEqual(i + 1, len(expected))
240

241
242
    @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
    def test_shutdown_server(self, SP):
243
        vm_id = vm_recv['server']['id']
244
        self.client.shutdown_server(vm_id)
245
        SP.assert_called_once_with(
246
            vm_id, 'action',
247
            json_data=dict(shutdown=dict()), success=202)
248

249
250
    @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
    def test_start_server(self, SP):
251
        vm_id = vm_recv['server']['id']
252
        self.client.start_server(vm_id)
253
        SP.assert_called_once_with(
254
            vm_id, 'action',
255
            json_data=dict(start=dict()), success=202)
256

257
258
    @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
    def test_get_server_console(self, SP):
259
        cnsl = dict(console=dict(info1='i1', info2='i2', info3='i3'))
260
        FR.json = cnsl
261
        vm_id = vm_recv['server']['id']
262
        r = self.client.get_server_console(vm_id)
263
        SP.assert_called_once_with(
264
            vm_id, 'action',
265
            json_data=dict(console=dict(type='vnc')), success=200)
266
        self.assert_dicts_are_equal(r, cnsl['console'])
267
268

    def test_get_firewall_profile(self):
269
        vm_id = vm_recv['server']['id']
270
        v = firewalls['attachments']['values'][0]['firewallProfile']
271
        with patch.object(
272
                cyclades.CycladesClient, 'get_server_details',
273
                return_value=firewalls) as GSD:
274
            r = self.client.get_firewall_profile(vm_id)
275
            GSD.assert_called_once_with(vm_id)
276
            self.assertEqual(r, v)
277
        with patch.object(
278
                cyclades.CycladesClient, 'get_server_details',
279
                return_value=dict()):
280
            self.assertRaises(
281
282
                ClientError,
                self.client.get_firewall_profile,
283
                vm_id)
284

285
286
    @patch('%s.servers_post' % cyclades_pkg, return_value=FR())
    def test_set_firewall_profile(self, SP):
287
        vm_id = vm_recv['server']['id']
288
        v = firewalls['attachments']['values'][0]['firewallProfile']
289
        self.client.set_firewall_profile(vm_id, v)
290
        SP.assert_called_once_with(
291
            vm_id, 'action',
292
            json_data=dict(firewallProfile=dict(profile=v)), success=202)
293

294
295
    @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
    def test_get_server_stats(self, SG):
296
297
        vm_id = vm_recv['server']['id']
        stats = dict(stat1='v1', stat2='v2', stat3='v3', stat4='v4')
298
299
        FR.json = dict(stats=stats)
        r = self.client.get_server_stats(vm_id)
300
        SG.assert_called_once_with(vm_id, 'stats')
301
302
        self.assert_dicts_are_equal(stats, r)

303
304
    @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
    def test_create_network(self, NP):
305
306
307
308
309
310
311
312
313
        net_name = net_send['network']['name']
        FR.json = net_recv
        full_args = dict(
                cidr='192.168.0.0/24',
                gateway='192.168.0.1',
                type='MAC_FILTERED',
                dhcp=True)
        test_args = dict(full_args)
        test_args.update(dict(empty=None, full=None))
314
        net_exp = dict(dhcp=False, name=net_name, type='MAC_FILTERED')
315
316
317
        for arg, val in test_args.items():
            kwargs = {} if arg == 'empty' else full_args if (
                arg == 'full') else {arg: val}
318
319
            expected = dict(network=dict(net_exp))
            expected['network'].update(kwargs)
320
            r = self.client.create_network(net_name, **kwargs)
321
            self.assertEqual(
322
323
                NP.mock_calls[-1],
                call(json_data=expected, success=202))
324
            self.assert_dicts_are_equal(r, net_recv['network'])
325

326
327
    @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
    def test_connect_server(self, NP):
328
329
        vm_id = vm_recv['server']['id']
        net_id = net_recv['network']['id']
330
        self.client.connect_server(vm_id, net_id)
331
        NP.assert_called_once_with(
332
            net_id, 'action',
333
            json_data=dict(add=dict(serverRef=vm_id)))
334
335
336

    @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
    def test_disconnect_server(self, NP):
337
        net_id, vm_id = net_recv['network']['id'], vm_recv['server']['id']
338
339
340
341
342
343
        nic_id = 'nic-%s-%s' % (net_id, vm_id)
        vm_nics = [
            dict(id=nic_id, network_id=net_id),
            dict(id='another-nic-id', network_id='another-net-id'),
            dict(id=nic_id * 2, network_id=net_id * 2)]
        with patch.object(
344
                cyclades.CycladesClient,
345
                'list_server_nics',
346
347
                return_value=vm_nics) as LSN:
            r = self.client.disconnect_server(vm_id, nic_id)
348
349
            LSN.assert_called_once_with(vm_id)
            NP.assert_called_once_with(
350
                net_id, 'action',
351
352
                json_data=dict(remove=dict(attachment=nic_id)))
            self.assertEqual(r, 1)
353

354
355
    @patch('%s.servers_get' % cyclades_pkg, return_value=FR())
    def test_list_server_nics(self, SG):
356
357
358
359
        vm_id = vm_recv['server']['id']
        nics = dict(addresses=dict(values=[dict(id='nic1'), dict(id='nic2')]))
        FR.json = nics
        r = self.client.list_server_nics(vm_id)
360
        SG.assert_called_once_with(vm_id, 'ips')
361
362
363
        expected = nics['addresses']['values']
        for i in range(len(r)):
            self.assert_dicts_are_equal(r[i], expected[i])
364
        self.assertEqual(i + 1, len(r))
365

366
367
    @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
    def test_list_networks(self, NG):
368
369
        FR.json = net_list
        expected = net_list['networks']['values']
370
371
372
373
374
375
        for detail in ('', 'detail'):
            r = self.client.list_networks(detail=True if detail else False)
            self.assertEqual(NG.mock_calls[-1], call(command=detail))
            for i, net in enumerate(expected):
                self.assert_dicts_are_equal(r[i], net)
            self.assertEqual(i + 1, len(r))
376

377
378
    @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
    def test_list_network_nics(self, NG):
379
        net_id = net_recv['network']['id']
380
381
        FR.json = net_recv
        r = self.client.list_network_nics(net_id)
382
        NG.assert_called_once_with(network_id=net_id)
383
384
385
386
387
388
        expected = net_recv['network']['attachments']['values']
        for i in range(len(r)):
            self.assert_dicts_are_equal(r[i], expected[i])

    @patch('%s.networks_post' % cyclades_pkg, return_value=FR())
    def test_disconnect_network_nics(self, NP):
389
390
391
        net_id = net_recv['network']['id']
        nics = ['nic1', 'nic2', 'nic3']
        with patch.object(
392
                cyclades.CycladesClient,
393
                'list_network_nics',
394
                return_value=nics) as LNN:
395
            self.client.disconnect_network_nics(net_id)
396
            LNN.assert_called_once_with(net_id)
397
398
399
400
401
            for i in range(len(nics)):
                expected = call(net_id, 'action', json_data=dict(
                    remove=dict(attachment=nics[i])))
                self.assertEqual(expected, NP.mock_calls[i])

402
    @patch('%s.networks_get' % cyclades_pkg, return_value=FR())
403
    def test_get_network_details(self, NG):
404
        FR.json = net_recv
405
        net_id = net_recv['network']['id']
406
        r = self.client.get_network_details(net_id)
407
        NG.assert_called_once_with(network_id=net_id)
408
409
        self.assert_dicts_are_equal(r, net_recv['network'])

410
411
    @patch('%s.networks_put' % cyclades_pkg, return_value=FR())
    def test_update_network_name(self, NP):
412
413
        net_id = net_recv['network']['id']
        new_name = '%s_new' % net_id
414
        self.client.update_network_name(net_id, new_name)
415
416
417
        NP.assert_called_once_with(
            network_id=net_id,
            json_data=dict(network=dict(name=new_name)))
418

419
    def test_delete_network(self):
420
        net_id = net_recv['network']['id']
421
        with patch.object(
422
                cyclades.CycladesClient, 'networks_delete',
423
424
425
426
                return_value=FR()) as ND:
            self.client.delete_network(net_id)
            ND.assert_called_once_with(net_id)
        with patch.object(
427
                cyclades.CycladesClient, 'networks_delete',
428
429
430
431
432
433
434
                side_effect=ClientError('A 421 Error', 421)):
            try:
                self.client.delete_network(421)
            except ClientError as err:
                self.assertEqual(err.status, 421)
                self.assertEqual(err.details, [
                    'Network may be still connected to at least one server'])
435

436

437
438
439
if __name__ == '__main__':
    from sys import argv
    from kamaki.clients.test import runTestCase
440
    not_found = True
441
    if not argv[1:] or argv[1] == 'CycladesClient':
442
        not_found = False
443
444
        runTestCase(CycladesClient, 'Cyclades Client', argv[2:])
    if not argv[1:] or argv[1] == 'CycladesRestClient':
445
        not_found = False
446
        runTestCase(CycladesRestClient, 'CycladesRest Client', argv[2:])
447
448
    if not_found:
        print('TestCase %s not found' % argv[1])