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