root / snf-cyclades-app / synnefo / neutron / tests / api.py @ 633f6952
History | View | Annotate | Download (29.3 kB)
1 |
from snf_django.utils.testing import BaseAPITest |
---|---|
2 |
from django.utils import simplejson as json |
3 |
from synnefo.cyclades_settings import cyclades_services |
4 |
from synnefo.lib.services import get_service_path |
5 |
from synnefo.lib import join_urls |
6 |
from ipaddr import IPv4Network |
7 |
import json |
8 |
import synnefo.neutron.models_factory as mf |
9 |
import synnefo.db.models_factory as dbmf |
10 |
|
11 |
|
12 |
NEUTRON_URL = get_service_path(cyclades_services, "neutron", "v2.0") |
13 |
NETWORKS_URL = join_urls(NEUTRON_URL, "networks/")
|
14 |
SUBNETS_URL = join_urls(NEUTRON_URL, "subnets/")
|
15 |
ROUTERS_URL = join_urls(NEUTRON_URL, "routers/")
|
16 |
PORTS_URL = join_urls(NEUTRON_URL, "ports/")
|
17 |
|
18 |
class NetworkTest(BaseAPITest): |
19 |
|
20 |
def test_list_networks(self): |
21 |
response = self.get(NETWORKS_URL)
|
22 |
self.assertSuccess(response)
|
23 |
networks = json.loads(response.content) |
24 |
self.assertEqual(networks, {"networks": []}) |
25 |
|
26 |
def test_create_network(self): |
27 |
request = { |
28 |
"network": {
|
29 |
"name": "sample_network" |
30 |
} |
31 |
} |
32 |
response = self.post(NETWORKS_URL, params=json.dumps(request))
|
33 |
code = response.status_code |
34 |
self.assertEqual(code, 201) |
35 |
|
36 |
def test_create_network_no_name(self): |
37 |
response = self.post(NETWORKS_URL, params="") |
38 |
code = response.status_code |
39 |
self.assertEqual(code, 201) |
40 |
|
41 |
def test_get_unfound_network(self): |
42 |
url = join_urls(NETWORKS_URL, "123")
|
43 |
response = self.get(url)
|
44 |
self.assertItemNotFound(response)
|
45 |
|
46 |
def test_get_network(self): |
47 |
test_net = mf.NetworkFactory.create() |
48 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
49 |
response = self.get(url, user=test_net.userid)
|
50 |
# validate response
|
51 |
res = json.loads(response.content) |
52 |
net = res['network']
|
53 |
keys = net.keys() |
54 |
must_send = set(["status", "subnets", "name", "admin_state_up", |
55 |
"tenant_id", "id"]) |
56 |
self.assertEqual(set(keys).issuperset(must_send), True) |
57 |
self.assertEqual(response.status_code, 200) |
58 |
|
59 |
def test_get_deleted_network(self): |
60 |
test_net = mf.NetworkFactory.create() |
61 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
62 |
self.delete(url, user=test_net.userid)
|
63 |
response = self.get(url, user=test_net.userid)
|
64 |
self.assertEqual(response.status_code, 400) |
65 |
|
66 |
def test_delete_unfound_network(self): |
67 |
url = join_urls(NETWORKS_URL, "123")
|
68 |
response = self.delete(url)
|
69 |
self.assertItemNotFound(response)
|
70 |
|
71 |
def test_delete_network(self): |
72 |
test_net = mf.NetworkFactory.create() |
73 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
74 |
response = self.delete(url, user=test_net.userid)
|
75 |
self.assertEqual(response.status_code, 204) |
76 |
|
77 |
def test_delete_network_in_use(self): |
78 |
test_net = mf.NetworkFactory.create() |
79 |
test_iface = mf.NetworkInterfaceFactory.create(network=test_net) |
80 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
81 |
response = self.delete(url, user=test_net.userid)
|
82 |
self.assertEqual(response.status_code, 409) |
83 |
|
84 |
def test_put_unfound_network(self): |
85 |
url = join_urls(NETWORKS_URL, "123")
|
86 |
response = self.delete(url)
|
87 |
self.assertItemNotFound(response)
|
88 |
|
89 |
def test_put_network(self): |
90 |
test_net = mf.NetworkFactory.create() |
91 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
92 |
request = { |
93 |
"network": {
|
94 |
"name": "new_name"} |
95 |
} |
96 |
response = self.put(url, params=json.dumps(request),
|
97 |
user=test_net.userid) |
98 |
self.assertEqual(response.status_code, 200) |
99 |
|
100 |
def test_put_network_wrong_data(self): |
101 |
test_net = mf.NetworkFactory.create() |
102 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
103 |
request = { |
104 |
"network": {
|
105 |
"wrong_field": "new_name"} |
106 |
} |
107 |
response = self.put(url, params=json.dumps(request),
|
108 |
user=test_net.userid) |
109 |
self.assertEqual(response.status_code, 400) |
110 |
|
111 |
def test_put_no_data(self): |
112 |
test_net = mf.NetworkFactory.create() |
113 |
url = join_urls(NETWORKS_URL, str(test_net.id))
|
114 |
response = self.put(url, params="", user=test_net.userid) |
115 |
self.assertEqual(response.status_code, 400) |
116 |
|
117 |
|
118 |
class PortTest(BaseAPITest): |
119 |
def test_get_ports(self): |
120 |
url = join_urls(PORTS_URL) |
121 |
response = self.get(url)
|
122 |
self.assertEqual(response.status_code, 200) |
123 |
ports = json.loads(response.content) |
124 |
self.assertEqual(ports, {"ports": []}) |
125 |
|
126 |
def test_get_port(self): |
127 |
nic = mf.NetworkInterfaceFactory.create() |
128 |
url = join_urls(PORTS_URL, str(nic.id))
|
129 |
response = self.get(url, user=nic.network.userid)
|
130 |
self.assertEqual(response.status_code, 200) |
131 |
|
132 |
def test_delete_port(self): |
133 |
nic = mf.NetworkInterfaceFactory.create() |
134 |
url = join_urls(PORTS_URL, str(nic.id))
|
135 |
response = self.delete(url, user=nic.network.userid)
|
136 |
self.assertEqual(response.status_code, 204) |
137 |
|
138 |
def test_update_port_name(self): |
139 |
nic = mf.NetworkInterfaceFactory.create() |
140 |
url = join_urls(PORTS_URL, str(nic.id))
|
141 |
request = {'port': {"name":"test-name"}} |
142 |
response = self.put(url, params=json.dumps(request), user=nic.network.userid)
|
143 |
self.assertEqual(response.status_code, 200) |
144 |
res = json.loads(response.content) |
145 |
self.assertEqual(res['port']['name'], "test-name") |
146 |
|
147 |
def test_update_port_sg_unfound(self): |
148 |
sg1 = mf.SecurityGroupFactory.create() |
149 |
nic = mf.NetworkInterfaceFactory.create() |
150 |
nic.security_groups.add(sg1) |
151 |
nic.save() |
152 |
url = join_urls(PORTS_URL, str(nic.id))
|
153 |
request = {'port': {"security_groups":["123"]}} |
154 |
response = self.put(url, params=json.dumps(request), user=nic.network.userid)
|
155 |
self.assertEqual(response.status_code, 404) |
156 |
|
157 |
def test_update_port_sg(self): |
158 |
sg1 = mf.SecurityGroupFactory.create() |
159 |
sg2 = mf.SecurityGroupFactory.create() |
160 |
sg3 = mf.SecurityGroupFactory.create() |
161 |
nic = mf.NetworkInterfaceFactory.create() |
162 |
nic.security_groups.add(sg1) |
163 |
nic.save() |
164 |
url = join_urls(PORTS_URL, str(nic.id))
|
165 |
request = {'port': {"security_groups":[str(sg2.id), str(sg3.id)]}} |
166 |
response = self.put(url, params=json.dumps(request), user=nic.network.userid)
|
167 |
res = json.loads(response.content) |
168 |
self.assertEqual(res['port']['security_groups'], [str(sg2.id), str(sg3.id)]) |
169 |
|
170 |
class RouterTest(BaseAPITest): |
171 |
def test_list_empty_routers(self): |
172 |
response = self.get(ROUTERS_URL)
|
173 |
self.assertSuccess(response)
|
174 |
routers = json.loads(response.content) |
175 |
self.assertEqual(routers, {"routers": []}) |
176 |
|
177 |
def test_create_router(self): |
178 |
pass
|
179 |
|
180 |
def test_get_router(self): |
181 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
182 |
net = mf.NetworkFactory.create(public=True)
|
183 |
nic = mf.NetworkInterfaceFactory.create(network=net, machine=router) |
184 |
response = self.get(ROUTERS_URL, user=router.userid)
|
185 |
self.assertSuccess(response)
|
186 |
|
187 |
def test_delete_router(self): |
188 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
189 |
url = join_urls(ROUTERS_URL, str(router.id))
|
190 |
response = self.delete(url, user=router.userid)
|
191 |
self.assertEqual(response.status_code, 204) |
192 |
|
193 |
def test_delete_router_with_private_net(self): |
194 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
195 |
net = mf.NetworkFactory.create(public=False)
|
196 |
nic = mf.NetworkInterfaceFactory.create(network=net, machine=router) |
197 |
url = join_urls(ROUTERS_URL, str(router.id))
|
198 |
response = self.delete(url, user=router.userid)
|
199 |
self.assertEqual(response.status_code, 409) |
200 |
|
201 |
def test_update_router(self): |
202 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
203 |
net1 = mf.NetworkFactory.create(public=True)
|
204 |
net2 = mf.NetworkFactory.create(public=True)
|
205 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router) |
206 |
request = { |
207 |
"router": {
|
208 |
"name": "new_name", |
209 |
"external_gateway_info": {"network_id": net2.id} |
210 |
} |
211 |
} |
212 |
url = join_urls(ROUTERS_URL, str(router.id))
|
213 |
response = self.put(url, params=json.dumps(request),
|
214 |
user=router.userid) |
215 |
info = json.loads(response.content) |
216 |
self.assertEqual(info['router']['external_gateway_info']['network_id'], |
217 |
net2.id) |
218 |
self.assertEqual(info['router']['name'], "new_name") |
219 |
|
220 |
def test_remove_interface_no_body(self): |
221 |
url = join_urls(join_urls(ROUTERS_URL, "123"),
|
222 |
"remove_router_interface")
|
223 |
|
224 |
response = self.put(url, params="") |
225 |
self.assertEqual(response.status_code, 400) |
226 |
|
227 |
def test_remove_interface_unfound_port(self): |
228 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
229 |
request = {"port_id": "123"} |
230 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
231 |
"remove_router_interface")
|
232 |
response = self.put(url, params=json.dumps(request),
|
233 |
user=router.userid) |
234 |
self.assertEqual(response.status_code, 404) |
235 |
|
236 |
def test_remove_interface_unfound_subnet(self): |
237 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
238 |
request = {"subnet_id": "123"} |
239 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
240 |
"remove_router_interface")
|
241 |
response = self.put(url, params=json.dumps(request),
|
242 |
user=router.userid) |
243 |
self.assertEqual(response.status_code, 404) |
244 |
|
245 |
def test_remove_interface_no_info(self): |
246 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
247 |
request = {"wrong": "123"} |
248 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
249 |
"remove_router_interface")
|
250 |
response = self.put(url, params=json.dumps(request),
|
251 |
user=router.userid) |
252 |
self.assertEqual(response.status_code, 400) |
253 |
|
254 |
def test_remove_interface_both_info_wrong(self): |
255 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
256 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
257 |
subnet = mf.SubnetFactory.create(network=net1) |
258 |
subnet2 = mf.SubnetFactory.create(network=net1) |
259 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router, |
260 |
subnet=subnet2) |
261 |
request = {"subnet_id": subnet.id, "port_id": nic.id} |
262 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
263 |
"remove_router_interface")
|
264 |
response = self.put(url, params=json.dumps(request),
|
265 |
user=router.userid) |
266 |
self.assertEqual(response.status_code, 409) |
267 |
|
268 |
def test_remove_interface_subnet(self): |
269 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
270 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
271 |
subnet = mf.SubnetFactory.create(network=net1) |
272 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router, |
273 |
subnet=subnet) |
274 |
request = {"subnet_id": subnet.id}
|
275 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
276 |
"remove_router_interface")
|
277 |
response = self.put(url, params=json.dumps(request),
|
278 |
user=router.userid) |
279 |
self.assertEqual(response.status_code, 200) |
280 |
|
281 |
def test_remove_interface_port(self): |
282 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
283 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
284 |
subnet = mf.SubnetFactory.create(network=net1) |
285 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router, |
286 |
subnet=subnet) |
287 |
request = {"port_id": nic.id}
|
288 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
289 |
"remove_router_interface")
|
290 |
response = self.put(url, params=json.dumps(request),
|
291 |
user=router.userid) |
292 |
self.assertEqual(response.status_code, 200) |
293 |
|
294 |
def test_remove_interface_both_info(self): |
295 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
296 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
297 |
subnet = mf.SubnetFactory.create(network=net1) |
298 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router, |
299 |
subnet=subnet) |
300 |
request = {"subnet_id": subnet.id, "port_id": nic.id} |
301 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
302 |
"remove_router_interface")
|
303 |
response = self.put(url, params=json.dumps(request),
|
304 |
user=router.userid) |
305 |
self.assertEqual(response.status_code, 200) |
306 |
|
307 |
def test_add_interface_no_info(self): |
308 |
url = join_urls(join_urls(ROUTERS_URL, "123"), "add_router_interface") |
309 |
response = self.put(url, params="") |
310 |
self.assertEqual(response.status_code, 400) |
311 |
|
312 |
def test_add_interface_wrong_info(self): |
313 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
314 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
315 |
"add_router_interface")
|
316 |
request = {} |
317 |
response = self.put(url, params=json.dumps(request),
|
318 |
user=router.userid) |
319 |
self.assertEqual(response.status_code, 400) |
320 |
|
321 |
def test_add_interface_unfound_port(self): |
322 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
323 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
324 |
"add_router_interface")
|
325 |
request = {"port_id": "123"} |
326 |
response = self.put(url, params=json.dumps(request),
|
327 |
user=router.userid) |
328 |
self.assertEqual(response.status_code, 404) |
329 |
|
330 |
def test_add_interface_unfound_subnet(self): |
331 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
332 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
333 |
"add_router_interface")
|
334 |
request = {"subnet_id": "123"} |
335 |
response = self.put(url, params=json.dumps(request),
|
336 |
user=router.userid) |
337 |
self.assertEqual(response.status_code, 404) |
338 |
|
339 |
def test_add_interface_both(self): |
340 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
341 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
342 |
subnet = mf.SubnetFactory.create(network=net1) |
343 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
344 |
"add_router_interface")
|
345 |
request = {"subnet_id": subnet.id, "port_id": "123"} |
346 |
response = self.put(url, params=json.dumps(request),
|
347 |
user=router.userid) |
348 |
self.assertEqual(response.status_code, 400) |
349 |
|
350 |
def test_add_interface_subnet(self): |
351 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
352 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
353 |
subnet = mf.SubnetFactory.create(network=net1) |
354 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
355 |
"add_router_interface")
|
356 |
request = {"subnet_id": subnet.id}
|
357 |
response = self.put(url, params=json.dumps(request),
|
358 |
user=router.userid) |
359 |
self.assertEqual(response.status_code, 200) |
360 |
|
361 |
def test_add_interface_port_no_ip(self): |
362 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
363 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
364 |
subnet = mf.SubnetFactory.create(network=net1) |
365 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router, |
366 |
subnet=subnet) |
367 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
368 |
"add_router_interface")
|
369 |
request = {"port_id": nic.id}
|
370 |
response = self.put(url, params=json.dumps(request),
|
371 |
user=router.userid) |
372 |
self.assertEqual(response.status_code, 400) |
373 |
|
374 |
def test_add_interface_port_connected(self): |
375 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
376 |
vm = dbmf.VirtualMachineFactory.create() |
377 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
378 |
subnet = mf.SubnetFactory.create(network=net1) |
379 |
nic = mf.NetworkInterfaceFactory.create(network=net1, machine=vm, |
380 |
subnet=subnet, |
381 |
ipv4="192.168.0.1")
|
382 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
383 |
"add_router_interface")
|
384 |
request = {"port_id": nic.id}
|
385 |
response = self.put(url, params=json.dumps(request),
|
386 |
user=router.userid) |
387 |
self.assertEqual(response.status_code, 409) |
388 |
|
389 |
def test_add_interface_port(self): |
390 |
router = dbmf.VirtualMachineFactory.create(router=True)
|
391 |
net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
|
392 |
subnet = mf.SubnetFactory.create(network=net1) |
393 |
nic = mf.NetworkInterfaceFactory.create(network=net1, |
394 |
machine=None,
|
395 |
subnet=subnet, |
396 |
ipv4="192.168.0.1")
|
397 |
url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
|
398 |
"add_router_interface")
|
399 |
request = {"port_id": nic.id}
|
400 |
response = self.put(url, params=json.dumps(request),
|
401 |
user=router.userid) |
402 |
self.assertEqual(response.status_code, 200) |
403 |
|
404 |
|
405 |
class SubnetTest(BaseAPITest): |
406 |
def test_list_subnets(self): |
407 |
'''Test list subnets without data'''
|
408 |
response = self.get(SUBNETS_URL)
|
409 |
self.assertSuccess(response)
|
410 |
subnets = json.loads(response.content) |
411 |
self.assertEqual(subnets, {'subnets': []}) |
412 |
|
413 |
def test_list_subnets_data(self): |
414 |
'''Test list subnets with data'''
|
415 |
test_net = mf.NetworkFactory() |
416 |
test_subnet_ipv4 = mf.SubnetFactory(network=test_net) |
417 |
test_subnet_ipv6 = mf.SubnetFactory(network=test_net, ipversion=6,
|
418 |
cidr='2620:0:2d0:200::7/32')
|
419 |
response = self.get(SUBNETS_URL, user=test_net.userid)
|
420 |
self.assertSuccess(response)
|
421 |
|
422 |
def test_get_subnet(self): |
423 |
'''Test get info of a single subnet'''
|
424 |
test_net = mf.NetworkFactory() |
425 |
test_subnet = mf.SubnetFactory(network=test_net) |
426 |
url = join_urls(SUBNETS_URL, str(test_subnet.id))
|
427 |
response = self.get(url, user=test_net.userid)
|
428 |
self.assertSuccess(response)
|
429 |
|
430 |
def test_get_subnet_404(self): |
431 |
'''Test get info of a subnet that doesn't exist'''
|
432 |
url = join_urls(SUBNETS_URL, '42')
|
433 |
response = self.get(url)
|
434 |
self.assertItemNotFound(response)
|
435 |
|
436 |
def test_subnet_delete(self): |
437 |
'''Test delete a subnet -- not supported'''
|
438 |
url = join_urls(SUBNETS_URL, '42')
|
439 |
response = self.delete(url)
|
440 |
self.assertBadRequest(response)
|
441 |
|
442 |
def test_create_subnet_success_ipv4(self): |
443 |
'''Test create a subnet successfully'''
|
444 |
test_net = mf.NetworkFactory() |
445 |
request = { |
446 |
'subnet': {
|
447 |
'network_id': test_net.id,
|
448 |
'cidr': '10.0.3.0/24', |
449 |
'ip_version': 4} |
450 |
} |
451 |
response = self.post(SUBNETS_URL, test_net.userid,
|
452 |
json.dumps(request), "json")
|
453 |
self.assertSuccess(response)
|
454 |
|
455 |
def test_create_subnet_success_ipv6(self): |
456 |
'''Test create an IPv6 subnet successfully'''
|
457 |
test_net = mf.NetworkFactory() |
458 |
request = { |
459 |
'subnet': {
|
460 |
'network_id': test_net.id,
|
461 |
'cidr': 'fdc1:4992:1130:fc0b::/64', |
462 |
'ip_version': 6} |
463 |
} |
464 |
response = self.post(SUBNETS_URL, test_net.userid,
|
465 |
json.dumps(request), "json")
|
466 |
self.assertSuccess(response)
|
467 |
|
468 |
def test_create_subnet_with_invalid_network_id(self): |
469 |
'''Test create a subnet with a network id that doesn't exist'''
|
470 |
test_net = mf.NetworkFactory() |
471 |
request = { |
472 |
'subnet': {
|
473 |
'network_id': '42', |
474 |
'cidr': '10.0.3.0/24', |
475 |
'ip_version': 4} |
476 |
} |
477 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
478 |
"json")
|
479 |
self.assertItemNotFound(response)
|
480 |
|
481 |
def test_create_subnet_with_malformed_ipversion(self): |
482 |
'''Create a subnet with a malformed ip_version type'''
|
483 |
test_net = mf.NetworkFactory() |
484 |
request = { |
485 |
'subnet': {
|
486 |
'network_id': test_net.id,
|
487 |
'cidr': '10.0.3.0/24', |
488 |
'ip_version': 8} |
489 |
} |
490 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
491 |
"json")
|
492 |
self.assertBadRequest(response)
|
493 |
|
494 |
def test_create_subnet_with_invalid_cidr(self): |
495 |
'''Create a subnet with an invalid cidr'''
|
496 |
test_net = mf.NetworkFactory() |
497 |
request = { |
498 |
'subnet': {
|
499 |
'network_id': test_net.id,
|
500 |
'cidr': '192.168.3.0/8'} |
501 |
} |
502 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
503 |
"json")
|
504 |
self.assertBadRequest(response)
|
505 |
|
506 |
def test_create_subnet_with_invalid_gateway(self): |
507 |
'''Create a subnet with a gateway outside of the subnet range'''
|
508 |
test_net = mf.NetworkFactory() |
509 |
request = { |
510 |
'subnet': {
|
511 |
'network_id': test_net.id,
|
512 |
'cidr': '192.168.3.0/24', |
513 |
'gateway_ip': '192.168.0.1'} |
514 |
} |
515 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
516 |
"json")
|
517 |
self.assertBadRequest(response)
|
518 |
|
519 |
def test_create_subnet_with_invalid_name(self): |
520 |
'''Create a subnet with an invalid subnet name'''
|
521 |
test_net = mf.NetworkFactory() |
522 |
request = { |
523 |
'subnet': {
|
524 |
'network_id': test_net.id,
|
525 |
'cidr': '192.168.3.0/24', |
526 |
'name': 'a' * 300} |
527 |
} |
528 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
529 |
"json")
|
530 |
self.assertBadRequest(response)
|
531 |
|
532 |
def test_create_subnet_with_invalid_dhcp(self): |
533 |
'''Create a subnet with an invalid dhcp value'''
|
534 |
test_net = mf.NetworkFactory() |
535 |
request = { |
536 |
'subnet': {
|
537 |
'network_id': test_net.id,
|
538 |
'cidr': '192.168.3.0/24', |
539 |
'enable_dhcp': 'None'} |
540 |
} |
541 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
542 |
"json")
|
543 |
self.assertBadRequest(response)
|
544 |
|
545 |
def test_create_subnet_with_dns_nameservers(self): |
546 |
'''Create a subnet with dns nameservers -- raises 400 BadRequest'''
|
547 |
test_net = mf.NetworkFactory() |
548 |
request = { |
549 |
'subnet': {
|
550 |
'network_id': test_net.id,
|
551 |
'cidr': '192.168.3.0/24', |
552 |
'dns_nameservers': '[]'} |
553 |
} |
554 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
555 |
"json")
|
556 |
self.assertBadRequest(response)
|
557 |
|
558 |
def test_create_subnet_with_host_routes(self): |
559 |
'''Create a subnet with host routes -- raises 400 BadRequest'''
|
560 |
test_net = mf.NetworkFactory() |
561 |
request = { |
562 |
'subnet': {
|
563 |
'network_id': test_net.id,
|
564 |
'cidr': '192.168.3.0/24', |
565 |
'host_routes': '[]'} |
566 |
} |
567 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
568 |
"json")
|
569 |
self.assertBadRequest(response)
|
570 |
|
571 |
def test_create_subnet_with_same_ipversion(self): |
572 |
'''
|
573 |
Create a subnet in a network with another subnet of the same
|
574 |
ipversion type
|
575 |
'''
|
576 |
test_net = mf.NetworkFactory() |
577 |
test_sub = mf.SubnetFactory(network=test_net) |
578 |
request = { |
579 |
'subnet': {
|
580 |
'network_id': test_net.id,
|
581 |
'cidr': '192.168.3.0/24'} |
582 |
} |
583 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
584 |
"json")
|
585 |
self.assertBadRequest(response)
|
586 |
|
587 |
def test_update_subnet_ip_version(self): |
588 |
'''Update the IP version of a subnet, raises 400 BadRequest'''
|
589 |
test_net = mf.NetworkFactory() |
590 |
test_sub = mf.SubnetFactory(network=test_net) |
591 |
request = { |
592 |
'subnet': {
|
593 |
'ip_version': '6'} |
594 |
} |
595 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
596 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
597 |
self.assertBadRequest(response)
|
598 |
|
599 |
def test_update_subnet_cidr(self): |
600 |
'''Update the cidr of a subnet, raises 400 BadRequest'''
|
601 |
test_net = mf.NetworkFactory() |
602 |
test_sub = mf.SubnetFactory(network=test_net) |
603 |
request = { |
604 |
'subnet': {
|
605 |
'cidr': '192.168.42.0/24'} |
606 |
} |
607 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
608 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
609 |
self.assertBadRequest(response)
|
610 |
|
611 |
def test_update_subnet_allocation_pools(self): |
612 |
'''Update the allocation pools of a subnet, raises 400 BadRequest'''
|
613 |
test_net = mf.NetworkFactory() |
614 |
test_sub = mf.SubnetFactory(network=test_net) |
615 |
request = { |
616 |
'subnet': {
|
617 |
'allocation_pools': '[]'} |
618 |
} |
619 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
620 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
621 |
self.assertBadRequest(response)
|
622 |
|
623 |
def test_update_subnet_add_dns(self): |
624 |
'''Update the dns nameservers of a subnet, raises 400 BadRequest'''
|
625 |
test_net = mf.NetworkFactory() |
626 |
test_sub = mf.SubnetFactory(network=test_net) |
627 |
request = { |
628 |
'subnet': {
|
629 |
'dns_nameservers': '[]'} |
630 |
} |
631 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
632 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
633 |
self.assertBadRequest(response)
|
634 |
|
635 |
def test_update_subnet_add_host_routes(self): |
636 |
'''Update the host routes of a subnet, raises 400 BadRequest'''
|
637 |
test_net = mf.NetworkFactory() |
638 |
test_sub = mf.SubnetFactory(network=test_net) |
639 |
request = { |
640 |
'subnet': {
|
641 |
'host_routes': '[]'} |
642 |
} |
643 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
644 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
645 |
self.assertBadRequest(response)
|
646 |
|
647 |
def test_update_subnet_with_invalid_dhcp_value(self): |
648 |
'''Update a subnet with an invalid dhcp value'''
|
649 |
test_net = mf.NetworkFactory() |
650 |
test_sub = mf.SubnetFactory(network=test_net) |
651 |
request = { |
652 |
'subnet': {
|
653 |
'enable_dhcp': 'Random'} |
654 |
} |
655 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
656 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
657 |
self.assertBadRequest(response)
|
658 |
|
659 |
def test_update_subnet_with_invalid_name(self): |
660 |
'''Update a subnet with an invalid name value'''
|
661 |
test_net = mf.NetworkFactory() |
662 |
test_sub = mf.SubnetFactory(network=test_net) |
663 |
request = { |
664 |
'subnet': {
|
665 |
'name': 'a' * 300} |
666 |
} |
667 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
668 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
669 |
self.assertBadRequest(response)
|
670 |
|
671 |
def test_update_subnet_with_invalid_gateway(self): |
672 |
'''Update a subnet with an invalid gateway value'''
|
673 |
test_net = mf.NetworkFactory() |
674 |
test_sub = mf.SubnetFactory(network=test_net) |
675 |
request = { |
676 |
'subnet': {
|
677 |
'gateway_ip': '192.168.200.0/24'} |
678 |
} |
679 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
680 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
681 |
self.assertBadRequest(response)
|
682 |
|
683 |
def test_update_subnet_gateway(self): |
684 |
'''Update the gateway of a subnet'''
|
685 |
test_net = mf.NetworkFactory() |
686 |
test_sub = mf.SubnetFactory(network=test_net) |
687 |
request = { |
688 |
'subnet': {
|
689 |
'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)} |
690 |
} |
691 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
692 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
693 |
self.assertSuccess(response)
|
694 |
|
695 |
def test_update_subnet_name(self): |
696 |
'''Update the name of a subnet'''
|
697 |
test_net = mf.NetworkFactory() |
698 |
test_sub = mf.SubnetFactory(network=test_net) |
699 |
request = { |
700 |
'subnet': {
|
701 |
'name': 'Updated Name'} |
702 |
} |
703 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
704 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
705 |
self.assertSuccess(response)
|
706 |
|
707 |
def test_update_subnet_dhcp(self): |
708 |
'''Update the dhcp flag of a subnet'''
|
709 |
test_net = mf.NetworkFactory() |
710 |
test_sub = mf.SubnetFactory(network=test_net) |
711 |
request = { |
712 |
'subnet': {
|
713 |
'enable_dhcp': False} |
714 |
} |
715 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
716 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
717 |
self.assertSuccess(response)
|