e298ec0b2c13a6f6f81cf49c7f856d5dee711347
[kamaki] / kamaki / clients / networking / test.py
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 networking
40
41
42 class NetworkingRestClient(TestCase):
43
44     """Set up a ComputesRest thorough test"""
45     def setUp(self):
46         self.url = 'http://networking.example.com'
47         self.token = 'n2tw0rk70k3n'
48         self.client = networking.NetworkingRestClient(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=dumps(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=dumps(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=dumps(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.set_param')
150     @patch('kamaki.clients.Client.post', return_value='ret val')
151     def test_ports_post(self, post, set_param):
152         for params, kwargs in product(
153                 [p for p in product(
154                     (
155                         ('name', 'port name', 'port name'),
156                         ('name', None, None)),
157                     (
158                         ('mac_address', 'max address', 'max address'),
159                         ('mac_address', None, None)),
160                     (
161                         ('fixed_ips', 'fixed ip', 'fixed ip'),
162                         ('fixed_ips', None, None)),
163                     (
164                         ('security_groups', 'sec groups', 'sec groups'),
165                         ('security_groups', None, None))
166                 )],
167                 (dict(), dict(k1='v1'), dict(k2='v2', k3='v3'))):
168
169             callargs = dict()
170             for p in params:
171                 callargs[p[0]] = p[2]
172             callargs.update(kwargs)
173
174             self.assertEqual(self.client.ports_post(**callargs), 'ret val')
175             self._assert(
176                 post, '/ports', set_param,
177                 params=params, json=None, **kwargs)
178
179             json_data = dict(id='some id', other_param='other val')
180             callargs['json_data'] = json_data
181             self.assertEqual(self.client.ports_post(**callargs), 'ret val')
182             self._assert(
183                 post, '/ports', set_param, params,
184                 json=dumps(json_data), **kwargs)
185
186     @patch('kamaki.clients.Client.set_param')
187     @patch('kamaki.clients.Client.put', return_value='ret val')
188     def test_ports_put(self, put, set_param):
189         port_id = 'portid'
190         for params, kwargs in product(
191                 [p for p in product(
192                     (
193                         ('name', 'port name', 'port name'),
194                         ('name', None, None)),
195                     (
196                         ('mac_address', 'max address', 'max address'),
197                         ('mac_address', None, None)),
198                     (
199                         ('fixed_ips', 'fixed ip', 'fixed ip'),
200                         ('fixed_ips', None, None)),
201                     (
202                         ('security_groups', 'sec groups', 'sec groups'),
203                         ('security_groups', None, None))
204                 )],
205                 (dict(), dict(k1='v1'), dict(k2='v2', k3='v3'))):
206
207             callargs = dict()
208             for p in params:
209                 callargs[p[0]] = p[2]
210             callargs.update(kwargs)
211
212             self.assertEqual(
213                 self.client.ports_put(port_id, **callargs), 'ret val')
214             self._assert(
215                 put, '/ports/%s' % port_id, set_param,
216                 params=params, json=None, **kwargs)
217
218             json_data = dict(id='some id', other_param='other val')
219             callargs['json_data'] = json_data
220             self.assertEqual(
221                 self.client.ports_put(port_id, **callargs), 'ret val')
222             self._assert(
223                 put, '/ports/%s' % port_id, set_param, params,
224                 json=dumps(json_data), **kwargs)
225
226
227 class FakeObject(object):
228
229     json = None
230     headers = None
231
232
233 class NetworkingClient(TestCase):
234
235     """Set up a ComputesRest thorough test"""
236     def setUp(self):
237         self.url = 'http://network.example.com'
238         self.token = 'n2tw0rk70k3n'
239         self.client = networking.NetworkingClient(self.url, self.token)
240
241     def tearDown(self):
242         FakeObject.json, FakeObject.headers = None, None
243         del self.client
244
245     @patch(
246         'kamaki.clients.networking.NetworkingClient.networks_get',
247         return_value=FakeObject())
248     def test_list_networks(self, networks_get):
249         FakeObject.json = dict(networks='ret val')
250         self.assertEqual(self.client.list_networks(), 'ret val')
251         networks_get.assert_called_once_with(success=200)
252
253     @patch(
254         'kamaki.clients.networking.NetworkingClient.networks_post',
255         return_value=FakeObject())
256     def test_create_network(self, networks_post):
257         for admin_state_up, shared in product((None, True), (None, True)):
258             FakeObject.json = dict(network='ret val')
259             name = 'net name'
260             self.assertEqual(
261                 self.client.create_network(
262                     name, admin_state_up=admin_state_up, shared=shared),
263                 'ret val')
264             req = dict(name=name, admin_state_up=bool(admin_state_up))
265             if shared:
266                 req['shared'] = shared
267             expargs = dict(json_data=dict(network=req), success=201)
268             self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
269
270     @patch(
271         'kamaki.clients.networking.NetworkingClient.networks_post',
272         return_value=FakeObject())
273     def test_create_networks(self, networks_post):
274         for networks in (
275                 None, dict(name='name'), 'nets', [1, 2, 3], [{'k': 'v'}, ],
276                 [dict(admin_state_up=True, shared=True)],
277                 [dict(name='n1', invalid='mistake'), ],
278                 [dict(name='valid', shared=True), {'err': 'nop'}]):
279             self.assertRaises(
280                 ValueError, self.client.create_networks, networks)
281
282         FakeObject.json = dict(networks='ret val')
283         for networks in (
284                 [
285                     dict(name='net1'),
286                     dict(name='net 2', admin_state_up=False, shared=True)],
287                 [
288                     dict(name='net1', admin_state_up=True),
289                     dict(name='net 2', shared=False),
290                     dict(name='net-3')],
291                 (dict(name='n.e.t'), dict(name='net 2'))):
292             self.assertEqual(self.client.create_networks(networks), 'ret val')
293
294             networks = list(networks)
295             expargs = dict(json_data=dict(networks=networks), success=201)
296             self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
297
298     @patch(
299         'kamaki.clients.networking.NetworkingClient.networks_get',
300         return_value=FakeObject())
301     def test_get_network_details(self, networks_get):
302         netid, FakeObject.json = 'netid', dict(network='ret val')
303         self.assertEqual(self.client.get_network_details(netid), 'ret val')
304         networks_get.assert_called_once_with(netid, success=200)
305
306     @patch(
307         'kamaki.clients.networking.NetworkingClient.networks_put',
308         return_value=FakeObject())
309     def test_update_network(self, networks_put):
310         netid, FakeObject.json = 'netid', dict(network='ret val')
311         for name, admin_state_up, shared in product(
312                 ('net name', None), (True, None), (True, None)):
313             kwargs = dict(
314                 name=name, admin_state_up=admin_state_up, shared=shared)
315             self.assertEqual(
316                 self.client.update_network(netid, **kwargs), 'ret val')
317             if name in (None, ):
318                 kwargs.pop('name')
319             if admin_state_up in (None, ):
320                 kwargs.pop('admin_state_up')
321             if shared in (None, ):
322                 kwargs.pop('shared')
323             kwargs = dict(json_data=dict(network=kwargs), success=200)
324             self.assertEqual(
325                 networks_put.mock_calls[-1], call(netid, **kwargs))
326
327     @patch(
328         'kamaki.clients.networking.NetworkingClient.networks_delete',
329         return_value=FakeObject())
330     def test_delete_network(self, networks_delete):
331         netid, FakeObject.headers = 'netid', 'ret headers'
332         self.assertEqual(self.client.delete_network(netid), 'ret headers')
333         networks_delete.assert_called_once_with(netid, success=204)
334
335     @patch(
336         'kamaki.clients.networking.NetworkingClient.subnets_get',
337         return_value=FakeObject())
338     def test_list_subnets(self, subnets_get):
339         FakeObject.json = dict(subnets='ret val')
340         self.assertEqual(self.client.list_subnets(), 'ret val')
341         subnets_get.assert_called_once_with(success=200)
342
343     @patch(
344         'kamaki.clients.networking.NetworkingClient.subnets_post',
345         return_value=FakeObject())
346     def test_create_subnet(self, subnets_post):
347         for (
348                 name, allocation_pools, gateway_ip,
349                 subnet_id, ipv6, enable_dhcp) in product(
350                     ('name', None), ('all pools', None), ('gip', None),
351                     ('sid', None), (True, None), (True, None)):
352             kwargs = dict(
353                 name=name, allocation_pools=allocation_pools,
354                 gateway_ip=gateway_ip, subnet_id=subnet_id,
355                 ipv6=ipv6, enable_dhcp=enable_dhcp)
356             FakeObject.json, network_id, cidr = dict(subnet='rv'), 'name', 'cd'
357             self.assertEqual(
358                 self.client.create_subnet(network_id, cidr, **kwargs), 'rv')
359             req = dict(
360                 network_id=network_id, cidr=cidr,
361                 ip_version=6 if kwargs.pop('ipv6', None) else 4)
362             for k, v in kwargs.items():
363                 if v:
364                     req['id' if k == 'subnet_id' else k] = v
365             expargs = dict(json_data=dict(subnet=req), success=201)
366             self.assertEqual(subnets_post.mock_calls[-1], call(**expargs))
367
368     @patch(
369         'kamaki.clients.networking.NetworkingClient.subnets_post',
370         return_value=FakeObject())
371     def test_create_subnets(self, subnets_post):
372         for subnets in (
373                 None, dict(network_id='name'), 'nets', [1, 2, 3], [{'k': 'v'}],
374                 [dict(ipv6=True, enable_dhcp=True)],
375                 [dict(network_id='n1', cidr='dr', invalid='mistake'), ],
376                 [dict(network_id='valid', cidr='valid'), {'err': 'nop'}]):
377             self.assertRaises(
378                 ValueError, self.client.create_subnets, subnets)
379
380         FakeObject.json = dict(subnets='ret val')
381         for subnets in (
382                 [
383                     dict(network_id='n1', cidr='c1'),
384                     dict(network_id='n 2', cidr='c 2', name='name')],
385                 [
386                     dict(network_id='n1', cidr='c 6', allocation_pools='a p'),
387                     dict(network_id='n 2', cidr='c_4', gateway_ip='g ip'),
388                     dict(network_id='n 2', cidr='c_4', subnet_id='s id'),
389                     dict(network_id='n-4', cidr='c3', ipv6=True, name='w. 6'),
390                     dict(network_id='n_5', cidr='c2', enable_dhcp=True)],
391                 (
392                     dict(network_id='n.e.t', cidr='c-5'),
393                     dict(network_id='net 2', cidr='c 2'))):
394             self.assertEqual(self.client.create_subnets(subnets), 'ret val')
395
396             for subnet in subnets:
397                 subnet['ip_version'] = 6 if subnet.pop('ipv6', None) else 4
398                 if 'subnet_id' in subnet:
399                     subnet['id'] = subnet.pop('subnet_id')
400             subnets = list(subnets)
401             expargs = dict(json_data=dict(subnets=subnets), success=201)
402             self.assertEqual(subnets_post.mock_calls[-1], call(**expargs))
403
404     @patch(
405         'kamaki.clients.networking.NetworkingClient.subnets_get',
406         return_value=FakeObject())
407     def test_get_subnet_details(self, subnets_get):
408         subid, FakeObject.json = 'subid', 'ret val'
409         self.assertEqual(self.client.get_subnet_details(subid), 'ret val')
410         subnets_get.assert_called_once_with(subid, success=201)
411
412     @patch(
413         'kamaki.clients.networking.NetworkingClient.subnets_put',
414         return_value=FakeObject())
415     def test_update_subnet(self, subnets_put):
416         for (
417                 name, allocation_pools, gateway_ip,
418                 subnet_id, ipv6, enable_dhcp) in product(
419                     ('name', None), ('all pools', None), ('gip', None),
420                     ('sid', None), (True, False, None), (True, False, None)):
421             kwargs = dict(
422                 name=name, allocation_pools=allocation_pools,
423                 gateway_ip=gateway_ip, subnet_id=subnet_id,
424                 ipv6=ipv6, enable_dhcp=enable_dhcp)
425             FakeObject.json, network_id, cidr = dict(subnet='rv'), 'name', 'cd'
426             self.assertEqual(
427                 self.client.update_subnet(network_id, cidr, **kwargs), 'rv')
428             req = dict(network_id=network_id, cidr=cidr)
429             if kwargs.get('ipv6', None) not in (None, ):
430                 req['ip_version'] = 6 if kwargs.pop('ipv6') else 4
431             for k, v in kwargs.items():
432                 if v not in (None, ):
433                     req['id' if k == 'subnet_id' else k] = v
434             expargs = dict(json_data=dict(subnet=req), success=201)
435             self.assertEqual(subnets_put.mock_calls[-1], call(**expargs))
436
437     @patch(
438         'kamaki.clients.networking.NetworkingClient.subnets_delete',
439         return_value=FakeObject())
440     def test_delete_subnet(self, subnets_delete):
441         netid, FakeObject.headers = 'netid', 'ret headers'
442         self.assertEqual(self.client.delete_subnet(netid), 'ret headers')
443         subnets_delete.assert_called_once_with(netid, success=204)
444
445     @patch(
446         'kamaki.clients.networking.NetworkingClient.ports_get',
447         return_value=FakeObject())
448     def test_list_ports(self, ports_get):
449         FakeObject.json = dict(ports='ret val')
450         self.assertEqual(self.client.list_ports(), 'ret val')
451         ports_get.assert_called_once_with(success=200)
452
453     @patch(
454         'kamaki.clients.networking.NetworkingClient.ports_post',
455         return_value=FakeObject())
456     def test_create_port(self, ports_post):
457         for (
458                 name, status, admin_state_up,
459                 mac_address, fixed_ips, security_groups
460                 ) in product(
461                     ('name', None), ('status', None), (True, False, None),
462                     ('maddr', None), ('some ips', None), ([1, 2, 3], None)):
463             kwargs = dict(
464                 name=name, status=status, admin_state_up=admin_state_up,
465                 mac_address=mac_address, fixed_ips=fixed_ips,
466                 security_groups=security_groups)
467             FakeObject.json, network_id = dict(port='ret val'), 'name'
468             self.assertEqual(
469                 self.client.create_port(network_id, **kwargs), 'ret val')
470             req = dict(network_id=network_id)
471             for k, v in kwargs.items():
472                 if v not in (None, ):
473                     req[k] = v
474             expargs = dict(json_data=dict(port=req), success=201)
475             self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
476
477     @patch(
478         'kamaki.clients.networking.NetworkingClient.ports_post',
479         return_value=FakeObject())
480     def test_create_ports(self, ports_post):
481         for ports in (
482                 None, dict(network_id='name'), 'nets', [1, 2, 3], [{'k': 'v'}],
483                 [dict(name=True, mac_address='mac')],
484                 [dict(network_id='n1', invalid='mistake'), ],
485                 [dict(network_id='valid', name='valid'), {'err': 'nop'}]):
486             self.assertRaises(
487                 ValueError, self.client.create_ports, ports)
488
489         FakeObject.json = dict(ports='ret val')
490         for ports in (
491                 [dict(network_id='n1'), dict(network_id='n 2', name='name')],
492                 [
493                     dict(network_id='n1', name='n 6', status='status'),
494                     dict(network_id='n 2', admin_state_up=True, fixed_ips='f'),
495                     dict(network_id='n 2', mac_address='mc', name='n.a.m.e.'),
496                     dict(network_id='n-4', security_groups='s~G', name='w. 6'),
497                     dict(network_id='n_5', admin_state_up=False, name='f a')],
498                 (
499                     dict(network_id='n.e.t', name='c-5'),
500                     dict(network_id='net 2', status='YEAH'))):
501             self.assertEqual(self.client.create_ports(ports), 'ret val')
502             expargs = dict(json_data=dict(ports=list(ports)), success=201)
503             self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
504
505     @patch(
506         'kamaki.clients.networking.NetworkingClient.ports_get',
507         return_value=FakeObject())
508     def test_get_port_details(self, ports_get):
509         portid, FakeObject.json = 'portid', dict(ports='ret val')
510         self.assertEqual(self.client.get_port_details(portid), 'ret val')
511         ports_get.assert_called_once_with(portid, success=201)
512
513     @patch(
514         'kamaki.clients.networking.NetworkingClient.ports_delete',
515         return_value=FakeObject())
516     def test_delete_port(self, ports_delete):
517         portid, FakeObject.headers = 'portid', 'ret headers'
518         self.assertEqual(self.client.delete_port(portid), 'ret headers')
519         ports_delete.assert_called_once_with(portid, success=204)
520
521
522 if __name__ == '__main__':
523     from sys import argv
524     from kamaki.clients.test import runTestCase
525     not_found = True
526     if not argv[1:] or argv[1] == 'NetworkingClient':
527         not_found = False
528         runTestCase(NetworkingClient, 'Networking Client', argv[2:])
529     if not argv[1:] or argv[1] == 'NetworkingRest':
530         not_found = False
531         runTestCase(NetworkingRestClient, 'NetworkingRest Client', argv[2:])
532     if not_found:
533         print('TestCase %s not found' % argv[1])