Statistics
| Branch: | Tag: | Revision:

root / kamaki / clients / network / test.py @ 447365fe

History | View | Annotate | Download (22.1 kB)

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

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

    
39
from kamaki.clients import network
40

    
41

    
42
class NetworkRestClient(TestCase):
43

    
44
    """Set up a ComputesRest thorough test"""
45
    def setUp(self):
46
        self.url = 'http://network.example.com'
47
        self.token = 'n2tw0rk70k3n'
48
        self.client = network.NetworkRestClient(self.url, self.token)
49

    
50
    def tearDown(self):
51
        del self.client
52

    
53
    def _assert(self, method_call, path, set_param=None, params=(), **kwargs):
54
        """Assert the REST method call is called as expected"""
55
        x0 = - len(params)
56
        for i, (k, v, c) in enumerate(params):
57
            self.assertEqual(set_param.mock_calls[x0 + i], call(k, v, iff=c))
58

    
59
        self.assertEqual(method_call.mock_calls[-1], call(path, **kwargs))
60

    
61
    @patch('kamaki.clients.Client.get', return_value='ret val')
62
    def test_networks_get(self, get):
63
        netid = 'netid'
64
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
65
            self.assertEqual(self.client.networks_get(**kwargs), 'ret val')
66
            self._assert(get, '/networks', **kwargs)
67
            self.assertEqual(
68
                self.client.networks_get(network_id=netid, **kwargs),
69
                'ret val')
70
            self._assert(get, '/networks/%s' % netid, **kwargs)
71

    
72
    @patch('kamaki.clients.Client.post', return_value='ret val')
73
    def test_networks_post(self, post):
74
        for kwargs in (
75
                dict(json_data=dict(k1='v1')),
76
                dict(json_data=dict(k2='v2'), k3='v3')):
77
            self.assertEqual(self.client.networks_post(**kwargs), 'ret val')
78
            json_data = kwargs.pop('json_data')
79
            self._assert(post, '/networks', json=json_data, **kwargs)
80

    
81
    @patch('kamaki.clients.Client.put', return_value='ret val')
82
    def test_networks_put(self, put):
83
        netid = 'netid'
84
        for kwargs in (
85
                dict(json_data=dict(k1='v1')),
86
                dict(json_data=dict(k2='v2'), k3='v3')):
87
            self.assertEqual(
88
                self.client.networks_put(netid, **kwargs), 'ret val')
89
            json_data = kwargs.pop('json_data')
90
            self._assert(
91
                put, '/networks/%s' % netid, json=json_data, **kwargs)
92

    
93
    @patch('kamaki.clients.Client.delete', return_value='ret val')
94
    def test_networks_delete(self, delete):
95
        netid = 'netid'
96
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
97
            self.assertEqual(
98
                self.client.networks_delete(netid, **kwargs), 'ret val')
99
            self._assert(delete, '/networks/%s' % netid, **kwargs)
100

    
101
    @patch('kamaki.clients.Client.get', return_value='ret val')
102
    def test_subnets_get(self, get):
103
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
104
            self.assertEqual(self.client.subnets_get(**kwargs), 'ret val')
105
            self._assert(get, '/subnets', **kwargs)
106

    
107
            subnet_id = 'subnet id'
108
            self.assertEqual(
109
                self.client.subnets_get(subnet_id=subnet_id, **kwargs),
110
                'ret val')
111
            self._assert(get, '/subnets/%s' % subnet_id, **kwargs)
112

    
113
    @patch('kamaki.clients.Client.post', return_value='ret val')
114
    def test_subnets_post(self, post):
115
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
116
            json_data = dict(subnets='some data')
117
            self.assertEqual(self.client.subnets_post(
118
                json_data=json_data, **kwargs), 'ret val')
119
            self._assert(post, '/subnets', json=json_data, **kwargs)
120

    
121
    @patch('kamaki.clients.Client.put', return_value='ret val')
122
    def test_subnets_put(self, put):
123
        subnet_id = 'subid'
124
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
125
            self.assertEqual(
126
                self.client.subnets_put(subnet_id, **kwargs), 'ret val')
127
            self._assert(put, '/subnets/%s' % subnet_id, **kwargs)
128

    
129
    @patch('kamaki.clients.Client.delete', return_value='ret val')
130
    def test_subnets_delete(self, delete):
131
        netid = 'netid'
132
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
133
            self.assertEqual(
134
                self.client.subnets_delete(netid, **kwargs), 'ret val')
135
            self._assert(delete, '/subnets/%s' % netid, **kwargs)
136

    
137
    @patch('kamaki.clients.Client.get', return_value='ret val')
138
    def test_ports_get(self, get):
139
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
140
            self.assertEqual(self.client.ports_get(**kwargs), 'ret val')
141
            self._assert(get, '/ports', **kwargs)
142

    
143
            port_id = 'port id'
144
            self.assertEqual(
145
                self.client.ports_get(port_id=port_id, **kwargs),
146
                'ret val')
147
            self._assert(get, '/ports/%s' % port_id, **kwargs)
148

    
149
    @patch('kamaki.clients.Client.post', return_value='ret val')
150
    def test_ports_post(self, post):
151
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
152
            self.assertEqual(self.client.ports_post(**kwargs), 'ret val')
153
            self._assert(post, '/ports', json=None, **kwargs)
154

    
155
            json_data = dict(id='some id', other_param='other val')
156
            self.assertEqual(
157
                self.client.ports_post(json_data=json_data, **kwargs),
158
                'ret val')
159
            self._assert(post, '/ports', json=json_data, **kwargs)
160

    
161
    @patch('kamaki.clients.Client.put', return_value='ret val')
162
    def test_ports_put(self, put):
163
        port_id = 'portid'
164
        for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
165
            self.assertEqual(
166
                self.client.ports_put(port_id, **kwargs), 'ret val')
167
            self._assert(put, '/ports/%s' % port_id, json=None, **kwargs)
168

    
169
            json_data = dict(id='some id', other_param='other val')
170
            self.assertEqual(
171
                self.client.ports_put(port_id, json_data=json_data, **kwargs),
172
                'ret val')
173
            self._assert(put, '/ports/%s' % port_id, json=json_data, **kwargs)
174

    
175

    
176
class FakeObject(object):
177

    
178
    json = None
179
    headers = None
180

    
181

    
182
class NetworkClient(TestCase):
183

    
184
    """Set up a ComputesRest thorough test"""
185
    def setUp(self):
186
        self.url = 'http://network.example.com'
187
        self.token = 'n2tw0rk70k3n'
188
        self.client = network.NetworkClient(self.url, self.token)
189

    
190
    def tearDown(self):
191
        FakeObject.json, FakeObject.headers = None, None
192
        del self.client
193

    
194
    @patch(
195
        'kamaki.clients.network.NetworkClient.networks_get',
196
        return_value=FakeObject())
197
    def test_list_networks(self, networks_get):
198
        FakeObject.json = dict(networks='ret val')
199
        self.assertEqual(self.client.list_networks(), 'ret val')
200
        networks_get.assert_called_once_with(success=200)
201

    
202
    @patch(
203
        'kamaki.clients.network.NetworkClient.networks_post',
204
        return_value=FakeObject())
205
    def test_create_network(self, networks_post):
206
        for admin_state_up, shared in product((None, True), (None, True)):
207
            FakeObject.json = dict(network='ret val')
208
            name = 'net name'
209
            self.assertEqual(
210
                self.client.create_network(
211
                    name, admin_state_up=admin_state_up, shared=shared),
212
                'ret val')
213
            req = dict(name=name, admin_state_up=bool(admin_state_up))
214
            if shared:
215
                req['shared'] = shared
216
            expargs = dict(json_data=dict(network=req), success=201)
217
            self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
218

    
219
    @patch(
220
        'kamaki.clients.network.NetworkClient.networks_post',
221
        return_value=FakeObject())
222
    def test_create_networks(self, networks_post):
223
        for networks in (
224
                None, dict(name='name'), 'nets', [1, 2, 3], [{'k': 'v'}, ],
225
                [dict(admin_state_up=True, shared=True)],
226
                [dict(name='n1', invalid='mistake'), ],
227
                [dict(name='valid', shared=True), {'err': 'nop'}]):
228
            self.assertRaises(
229
                ValueError, self.client.create_networks, networks)
230

    
231
        FakeObject.json = dict(networks='ret val')
232
        for networks in (
233
                [
234
                    dict(name='net1'),
235
                    dict(name='net 2', admin_state_up=False, shared=True)],
236
                [
237
                    dict(name='net1', admin_state_up=True),
238
                    dict(name='net 2', shared=False),
239
                    dict(name='net-3')],
240
                (dict(name='n.e.t'), dict(name='net 2'))):
241
            self.assertEqual(self.client.create_networks(networks), 'ret val')
242

    
243
            networks = list(networks)
244
            expargs = dict(json_data=dict(networks=networks), success=201)
245
            self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
246

    
247
    @patch(
248
        'kamaki.clients.network.NetworkClient.networks_get',
249
        return_value=FakeObject())
250
    def test_get_network_details(self, networks_get):
251
        netid, FakeObject.json = 'netid', dict(network='ret val')
252
        self.assertEqual(self.client.get_network_details(netid), 'ret val')
253
        networks_get.assert_called_once_with(netid, success=200)
254

    
255
    @patch(
256
        'kamaki.clients.network.NetworkClient.networks_put',
257
        return_value=FakeObject())
258
    def test_update_network(self, networks_put):
259
        netid, FakeObject.json = 'netid', dict(network='ret val')
260
        for name, admin_state_up, shared in product(
261
                ('net name', None), (True, None), (True, None)):
262
            kwargs = dict(
263
                name=name, admin_state_up=admin_state_up, shared=shared)
264
            self.assertEqual(
265
                self.client.update_network(netid, **kwargs), 'ret val')
266
            if name in (None, ):
267
                kwargs.pop('name')
268
            if admin_state_up in (None, ):
269
                kwargs.pop('admin_state_up')
270
            if shared in (None, ):
271
                kwargs.pop('shared')
272
            kwargs = dict(json_data=dict(network=kwargs), success=200)
273
            self.assertEqual(
274
                networks_put.mock_calls[-1], call(netid, **kwargs))
275

    
276
    @patch(
277
        'kamaki.clients.network.NetworkClient.networks_delete',
278
        return_value=FakeObject())
279
    def test_delete_network(self, networks_delete):
280
        netid, FakeObject.headers = 'netid', 'ret headers'
281
        self.assertEqual(self.client.delete_network(netid), 'ret headers')
282
        networks_delete.assert_called_once_with(netid, success=204)
283

    
284
    @patch(
285
        'kamaki.clients.network.NetworkClient.subnets_get',
286
        return_value=FakeObject())
287
    def test_list_subnets(self, subnets_get):
288
        FakeObject.json = dict(subnets='ret val')
289
        self.assertEqual(self.client.list_subnets(), 'ret val')
290
        subnets_get.assert_called_once_with(success=200)
291

    
292
    @patch(
293
        'kamaki.clients.network.NetworkClient.subnets_post',
294
        return_value=FakeObject())
295
    def test_create_subnet(self, subnets_post):
296
        for (
297
                name, allocation_pools, gateway_ip,
298
                subnet_id, ipv6, enable_dhcp) in product(
299
                    ('name', None), ('all pools', None), ('gip', None),
300
                    ('sid', None), (True, None), (True, None)):
301
            kwargs = dict(
302
                name=name, allocation_pools=allocation_pools,
303
                gateway_ip=gateway_ip, subnet_id=subnet_id,
304
                ipv6=ipv6, enable_dhcp=enable_dhcp)
305
            FakeObject.json, network_id, cidr = dict(subnet='rv'), 'name', 'cd'
306
            self.assertEqual(
307
                self.client.create_subnet(network_id, cidr, **kwargs), 'rv')
308
            req = dict(
309
                network_id=network_id, cidr=cidr,
310
                ip_version=6 if kwargs.pop('ipv6', None) else 4)
311
            for k, v in kwargs.items():
312
                if v:
313
                    req['id' if k == 'subnet_id' else k] = v
314
            expargs = dict(json_data=dict(subnet=req), success=201)
315
            self.assertEqual(subnets_post.mock_calls[-1], call(**expargs))
316

    
317
    @patch(
318
        'kamaki.clients.network.NetworkClient.subnets_post',
319
        return_value=FakeObject())
320
    def test_create_subnets(self, subnets_post):
321
        for subnets in (
322
                None, dict(network_id='name'), 'nets', [1, 2, 3], [{'k': 'v'}],
323
                [dict(ipv6=True, enable_dhcp=True)],
324
                [dict(network_id='n1', cidr='dr', invalid='mistake'), ],
325
                [dict(network_id='valid', cidr='valid'), {'err': 'nop'}]):
326
            self.assertRaises(
327
                ValueError, self.client.create_subnets, subnets)
328

    
329
        FakeObject.json = dict(subnets='ret val')
330
        for subnets in (
331
                [
332
                    dict(network_id='n1', cidr='c1'),
333
                    dict(network_id='n 2', cidr='c 2', name='name')],
334
                [
335
                    dict(network_id='n1', cidr='c 6', allocation_pools='a p'),
336
                    dict(network_id='n 2', cidr='c_4', gateway_ip='g ip'),
337
                    dict(network_id='n 2', cidr='c_4', subnet_id='s id'),
338
                    dict(network_id='n-4', cidr='c3', ipv6=True, name='w. 6'),
339
                    dict(network_id='n_5', cidr='c2', enable_dhcp=True)],
340
                (
341
                    dict(network_id='n.e.t', cidr='c-5'),
342
                    dict(network_id='net 2', cidr='c 2'))):
343
            self.assertEqual(self.client.create_subnets(subnets), 'ret val')
344

    
345
            for subnet in subnets:
346
                subnet['ip_version'] = 6 if subnet.pop('ipv6', None) else 4
347
                if 'subnet_id' in subnet:
348
                    subnet['id'] = subnet.pop('subnet_id')
349
            subnets = list(subnets)
350
            expargs = dict(json_data=dict(subnets=subnets), success=201)
351
            self.assertEqual(subnets_post.mock_calls[-1], call(**expargs))
352

    
353
    @patch(
354
        'kamaki.clients.network.NetworkClient.subnets_get',
355
        return_value=FakeObject())
356
    def test_get_subnet_details(self, subnets_get):
357
        subid, FakeObject.json = 'subid', 'ret val'
358
        self.assertEqual(self.client.get_subnet_details(subid), 'ret val')
359
        subnets_get.assert_called_once_with(subid, success=201)
360

    
361
    @patch(
362
        'kamaki.clients.network.NetworkClient.subnets_put',
363
        return_value=FakeObject())
364
    def test_update_subnet(self, subnets_put):
365
        for (
366
                name, allocation_pools, gateway_ip,
367
                subnet_id, ipv6, enable_dhcp) in product(
368
                    ('name', None), ('all pools', None), ('gip', None),
369
                    ('sid', None), (True, False, None), (True, False, None)):
370
            kwargs = dict(
371
                name=name, allocation_pools=allocation_pools,
372
                gateway_ip=gateway_ip, subnet_id=subnet_id,
373
                ipv6=ipv6, enable_dhcp=enable_dhcp)
374
            FakeObject.json, network_id, cidr = dict(subnet='rv'), 'name', 'cd'
375
            self.assertEqual(
376
                self.client.update_subnet(network_id, cidr, **kwargs), 'rv')
377
            req = dict(network_id=network_id, cidr=cidr)
378
            if kwargs.get('ipv6', None) not in (None, ):
379
                req['ip_version'] = 6 if kwargs.pop('ipv6') else 4
380
            for k, v in kwargs.items():
381
                if v not in (None, ):
382
                    req['id' if k == 'subnet_id' else k] = v
383
            expargs = dict(json_data=dict(subnet=req), success=201)
384
            self.assertEqual(subnets_put.mock_calls[-1], call(**expargs))
385

    
386
    @patch(
387
        'kamaki.clients.network.NetworkClient.subnets_delete',
388
        return_value=FakeObject())
389
    def test_delete_subnet(self, subnets_delete):
390
        netid, FakeObject.headers = 'netid', 'ret headers'
391
        self.assertEqual(self.client.delete_subnet(netid), 'ret headers')
392
        subnets_delete.assert_called_once_with(netid, success=204)
393

    
394
    @patch(
395
        'kamaki.clients.network.NetworkClient.ports_get',
396
        return_value=FakeObject())
397
    def test_list_ports(self, ports_get):
398
        FakeObject.json = dict(ports='ret val')
399
        self.assertEqual(self.client.list_ports(), 'ret val')
400
        ports_get.assert_called_once_with(success=200)
401

    
402
    @patch(
403
        'kamaki.clients.network.NetworkClient.ports_post',
404
        return_value=FakeObject())
405
    def test_create_port(self, ports_post):
406
        for (
407
                name, status, admin_state_up,
408
                mac_address, fixed_ips, security_groups
409
                ) in product(
410
                    ('name', None), ('status', None), (True, False, None),
411
                    ('maddr', None), ('some ips', None), ([1, 2, 3], None)):
412
            kwargs = dict(
413
                name=name, status=status, admin_state_up=admin_state_up,
414
                mac_address=mac_address, fixed_ips=fixed_ips,
415
                security_groups=security_groups)
416
            FakeObject.json, network_id = dict(port='ret val'), 'name'
417
            self.assertEqual(
418
                self.client.create_port(network_id, **kwargs), 'ret val')
419
            req = dict(network_id=network_id)
420
            for k, v in kwargs.items():
421
                if v not in (None, ):
422
                    req[k] = v
423
            expargs = dict(json_data=dict(port=req), success=201)
424
            self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
425

    
426
    @patch(
427
        'kamaki.clients.network.NetworkClient.ports_post',
428
        return_value=FakeObject())
429
    def test_create_ports(self, ports_post):
430
        for ports in (
431
                None, dict(network_id='name'), 'nets', [1, 2, 3], [{'k': 'v'}],
432
                [dict(name=True, mac_address='mac')],
433
                [dict(network_id='n1', invalid='mistake'), ],
434
                [dict(network_id='valid', name='valid'), {'err': 'nop'}]):
435
            self.assertRaises(
436
                ValueError, self.client.create_ports, ports)
437

    
438
        FakeObject.json = dict(ports='ret val')
439
        for ports in (
440
                [dict(network_id='n1'), dict(network_id='n 2', name='name')],
441
                [
442
                    dict(network_id='n1', name='n 6', status='status'),
443
                    dict(network_id='n 2', admin_state_up=True, fixed_ips='f'),
444
                    dict(network_id='n 2', mac_address='mc', name='n.a.m.e.'),
445
                    dict(network_id='n-4', security_groups='s~G', name='w. 6'),
446
                    dict(network_id='n_5', admin_state_up=False, name='f a')],
447
                (
448
                    dict(network_id='n.e.t', name='c-5'),
449
                    dict(network_id='net 2', status='YEAH'))):
450
            self.assertEqual(self.client.create_ports(ports), 'ret val')
451
            expargs = dict(json_data=dict(ports=list(ports)), success=201)
452
            self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
453

    
454
    @patch(
455
        'kamaki.clients.network.NetworkClient.ports_get',
456
        return_value=FakeObject())
457
    def test_get_port_details(self, ports_get):
458
        portid, FakeObject.json = 'portid', dict(ports='ret val')
459
        self.assertEqual(self.client.get_port_details(portid), 'ret val')
460
        ports_get.assert_called_once_with(portid, success=201)
461

    
462
    @patch(
463
        'kamaki.clients.network.NetworkClient.ports_delete',
464
        return_value=FakeObject())
465
    def test_delete_port(self, ports_delete):
466
        portid, FakeObject.headers = 'portid', 'ret headers'
467
        self.assertEqual(self.client.delete_port(portid), 'ret headers')
468
        ports_delete.assert_called_once_with(portid, success=204)
469

    
470
    @patch(
471
        'kamaki.clients.network.NetworkClient.ports_put',
472
        return_value=FakeObject())
473
    def test_update_port(self, ports_put):
474
        for (
475
                name, status, admin_state_up, mac_address, fixed_ips,
476
                security_groups) in product(
477
                    ('name', None), ('st', None), (True, None), ('mc', None),
478
                    ('fps', None), ('sg', None)):
479
            FakeObject.json = dict(port='rv')
480
            port_id, network_id = 'pid', 'nid'
481
            kwargs = dict(
482
                network_id=network_id, name=name, status=status,
483
                admin_state_up=admin_state_up, mac_address=mac_address,
484
                fixed_ips=fixed_ips, security_groups=security_groups)
485
            self.assertEqual(
486
                self.client.update_port(port_id, **kwargs), 'rv')
487
            req = dict()
488
            for k, v in kwargs.items():
489
                if v:
490
                    req[k] = v
491
            expargs = dict(json_data=dict(port=req), success=201)
492
            self.assertEqual(
493
                ports_put.mock_calls[-1], call(port_id, **expargs))
494

    
495

    
496
if __name__ == '__main__':
497
    from sys import argv
498
    from kamaki.clients.test import runTestCase
499
    not_found = True
500
    if not argv[1:] or argv[1] == 'NetworkClient':
501
        not_found = False
502
        runTestCase(NetworkClient, 'Network Client', argv[2:])
503
    if not argv[1:] or argv[1] == 'NetworkRestClient':
504
        not_found = False
505
        runTestCase(NetworkRestClient, 'NetworkRest Client', argv[2:])
506
    if not_found:
507
        print('TestCase %s not found' % argv[1])