1 # Copyright 2013 GRNET S.A. All rights reserved.
3 # Redistribution and use in source and binary forms, with or
4 # without modification, are permitted provided that the following
7 # 1. Redistributions of source code must retain the above
8 # copyright notice, this list of conditions and the following
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.
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.
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.
34 from mock import patch, call
35 from unittest import TestCase
36 from itertools import product
37 from json import dumps
39 from kamaki.clients import networking
42 class NetworkingRestClient(TestCase):
44 """Set up a ComputesRest thorough test"""
46 self.url = 'http://networking.example.com'
47 self.token = 'n2tw0rk70k3n'
48 self.client = networking.NetworkingRestClient(self.url, self.token)
53 def _assert(self, method_call, path, set_param=None, params=(), **kwargs):
54 """Assert the REST method call is called as expected"""
56 for i, (k, v, c) in enumerate(params):
57 self.assertEqual(set_param.mock_calls[x0 + i], call(k, v, iff=c))
59 self.assertEqual(method_call.mock_calls[-1], call(path, **kwargs))
61 @patch('kamaki.clients.Client.get', return_value='ret val')
62 def test_networks_get(self, get):
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)
68 self.client.networks_get(network_id=netid, **kwargs),
70 self._assert(get, '/networks/%s' % netid, **kwargs)
72 @patch('kamaki.clients.Client.post', return_value='ret val')
73 def test_networks_post(self, post):
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)
81 @patch('kamaki.clients.Client.put', return_value='ret val')
82 def test_networks_put(self, put):
85 dict(json_data=dict(k1='v1')),
86 dict(json_data=dict(k2='v2'), k3='v3')):
88 self.client.networks_put(netid, **kwargs), 'ret val')
89 json_data = kwargs.pop('json_data')
91 put, '/networks/%s' % netid, json=dumps(json_data), **kwargs)
93 @patch('kamaki.clients.Client.delete', return_value='ret val')
94 def test_networks_delete(self, delete):
96 for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
98 self.client.networks_delete(netid, **kwargs), 'ret val')
99 self._assert(delete, '/networks/%s' % netid, **kwargs)
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)
107 subnet_id = 'subnet id'
109 self.client.subnets_get(subnet_id=subnet_id, **kwargs),
111 self._assert(get, '/subnets/%s' % subnet_id, **kwargs)
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)
121 @patch('kamaki.clients.Client.put', return_value='ret val')
122 def test_subnets_put(self, put):
124 for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
126 self.client.subnets_put(subnet_id, **kwargs), 'ret val')
127 self._assert(put, '/subnets/%s' % subnet_id, **kwargs)
129 @patch('kamaki.clients.Client.delete', return_value='ret val')
130 def test_subnets_delete(self, delete):
132 for kwargs in (dict(), dict(k1='v1'), dict(k2='v2', k3='v3')):
134 self.client.subnets_delete(netid, **kwargs), 'ret val')
135 self._assert(delete, '/subnets/%s' % netid, **kwargs)
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)
145 self.client.ports_get(port_id=port_id, **kwargs),
147 self._assert(get, '/ports/%s' % port_id, **kwargs)
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(
155 ('name', 'port name', 'port name'),
156 ('name', None, None)),
158 ('mac_address', 'max address', 'max address'),
159 ('mac_address', None, None)),
161 ('fixed_ips', 'fixed ip', 'fixed ip'),
162 ('fixed_ips', None, None)),
164 ('security_groups', 'sec groups', 'sec groups'),
165 ('security_groups', None, None))
167 (dict(), dict(k1='v1'), dict(k2='v2', k3='v3'))):
171 callargs[p[0]] = p[2]
172 callargs.update(kwargs)
174 self.assertEqual(self.client.ports_post(**callargs), 'ret val')
176 post, '/ports', set_param,
177 params=params, json=None, **kwargs)
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')
183 post, '/ports', set_param, params,
184 json=dumps(json_data), **kwargs)
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):
190 for params, kwargs in product(
193 ('name', 'port name', 'port name'),
194 ('name', None, None)),
196 ('mac_address', 'max address', 'max address'),
197 ('mac_address', None, None)),
199 ('fixed_ips', 'fixed ip', 'fixed ip'),
200 ('fixed_ips', None, None)),
202 ('security_groups', 'sec groups', 'sec groups'),
203 ('security_groups', None, None))
205 (dict(), dict(k1='v1'), dict(k2='v2', k3='v3'))):
209 callargs[p[0]] = p[2]
210 callargs.update(kwargs)
213 self.client.ports_put(port_id, **callargs), 'ret val')
215 put, '/ports/%s' % port_id, set_param,
216 params=params, json=None, **kwargs)
218 json_data = dict(id='some id', other_param='other val')
219 callargs['json_data'] = json_data
221 self.client.ports_put(port_id, **callargs), 'ret val')
223 put, '/ports/%s' % port_id, set_param, params,
224 json=dumps(json_data), **kwargs)
227 class FakeObject(object):
233 class NetworkingClient(TestCase):
235 """Set up a ComputesRest thorough test"""
237 self.url = 'http://network.example.com'
238 self.token = 'n2tw0rk70k3n'
239 self.client = networking.NetworkingClient(self.url, self.token)
242 FakeObject.json, FakeObject.headers = None, None
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)
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')
261 self.client.create_network(
262 name, admin_state_up=admin_state_up, shared=shared),
264 req = dict(name=name, admin_state_up=bool(admin_state_up))
266 req['shared'] = shared
267 expargs = dict(json_data=dict(network=req), success=201)
268 self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
271 'kamaki.clients.networking.NetworkingClient.networks_post',
272 return_value=FakeObject())
273 def test_create_networks(self, networks_post):
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'}]):
280 ValueError, self.client.create_networks, networks)
282 FakeObject.json = dict(networks='ret val')
286 dict(name='net 2', admin_state_up=False, shared=True)],
288 dict(name='net1', admin_state_up=True),
289 dict(name='net 2', shared=False),
291 (dict(name='n.e.t'), dict(name='net 2'))):
292 self.assertEqual(self.client.create_networks(networks), 'ret val')
294 networks = list(networks)
295 expargs = dict(json_data=dict(networks=networks), success=201)
296 self.assertEqual(networks_post.mock_calls[-1], call(**expargs))
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)
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)):
314 name=name, admin_state_up=admin_state_up, shared=shared)
316 self.client.update_network(netid, **kwargs), 'ret val')
319 if admin_state_up in (None, ):
320 kwargs.pop('admin_state_up')
321 if shared in (None, ):
323 kwargs = dict(json_data=dict(network=kwargs), success=200)
325 networks_put.mock_calls[-1], call(netid, **kwargs))
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)
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)
344 'kamaki.clients.networking.NetworkingClient.subnets_post',
345 return_value=FakeObject())
346 def test_create_subnet(self, subnets_post):
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)):
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'
358 self.client.create_subnet(network_id, cidr, **kwargs), 'rv')
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():
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))
369 'kamaki.clients.networking.NetworkingClient.subnets_post',
370 return_value=FakeObject())
371 def test_create_subnets(self, subnets_post):
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'}]):
378 ValueError, self.client.create_subnets, subnets)
380 FakeObject.json = dict(subnets='ret val')
383 dict(network_id='n1', cidr='c1'),
384 dict(network_id='n 2', cidr='c 2', name='name')],
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)],
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')
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))
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)
413 'kamaki.clients.networking.NetworkingClient.subnets_put',
414 return_value=FakeObject())
415 def test_update_subnet(self, subnets_put):
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)):
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'
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))
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)
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)
454 'kamaki.clients.networking.NetworkingClient.ports_post',
455 return_value=FakeObject())
456 def test_create_port(self, ports_post):
458 name, status, admin_state_up,
459 mac_address, fixed_ips, security_groups
461 ('name', None), ('status', None), (True, False, None),
462 ('maddr', None), ('some ips', None), ([1, 2, 3], None)):
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'
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, ):
474 expargs = dict(json_data=dict(port=req), success=201)
475 self.assertEqual(ports_post.mock_calls[-1], call(**expargs))
478 'kamaki.clients.networking.NetworkingClient.ports_post',
479 return_value=FakeObject())
480 def test_create_ports(self, ports_post):
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'}]):
487 ValueError, self.client.create_ports, ports)
489 FakeObject.json = dict(ports='ret val')
491 [dict(network_id='n1'), dict(network_id='n 2', name='name')],
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')],
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))
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)
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)
522 if __name__ == '__main__':
524 from kamaki.clients.test import runTestCase
526 if not argv[1:] or argv[1] == 'NetworkingClient':
528 runTestCase(NetworkingClient, 'Networking Client', argv[2:])
529 if not argv[1:] or argv[1] == 'NetworkingRest':
531 runTestCase(NetworkingRestClient, 'NetworkingRest Client', argv[2:])
533 print('TestCase %s not found' % argv[1])