root / snf-cyclades-app / synnefo / api / tests / subnets.py @ 01adca6b
History | View | Annotate | Download (17.8 kB)
1 |
# Copyright 2011-2013 GRNET S.A. All rights reserved.
|
---|---|
2 |
#
|
3 |
# Redistribution and use in source and binary forms, with or without
|
4 |
# modification, are permitted provided that the following conditions
|
5 |
# are met:
|
6 |
#
|
7 |
# 1. Redistributions of source code must retain the above copyright
|
8 |
# notice, this list of conditions and the following disclaimer.
|
9 |
#
|
10 |
# 2. Redistributions in binary form must reproduce the above copyright
|
11 |
# notice, this list of conditions and the following disclaimer in the
|
12 |
# documentation and/or other materials provided with the distribution.
|
13 |
#
|
14 |
# THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND
|
15 |
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
16 |
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
17 |
# ARE DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
|
18 |
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
|
19 |
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
|
20 |
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
|
21 |
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
|
22 |
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
|
23 |
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
|
24 |
# SUCH DAMAGE.
|
25 |
#
|
26 |
# The views and conclusions contained in the software and documentation are
|
27 |
# those of the authors and should not be interpreted as representing official
|
28 |
# policies, either expressed or implied, of GRNET S.A.
|
29 |
|
30 |
from snf_django.utils.testing import BaseAPITest |
31 |
from django.utils import simplejson as json |
32 |
from synnefo.cyclades_settings import cyclades_services |
33 |
from synnefo.lib.services import get_service_path |
34 |
from synnefo.lib import join_urls |
35 |
from ipaddr import IPv4Network |
36 |
import json |
37 |
import synnefo.db.models_factory as mf |
38 |
|
39 |
|
40 |
COMPUTE_URL = get_service_path(cyclades_services, 'compute', version='v2.0') |
41 |
SUBNETS_URL = join_urls(COMPUTE_URL, "subnets/")
|
42 |
|
43 |
|
44 |
class SubnetTest(BaseAPITest): |
45 |
def test_list_subnets(self): |
46 |
"""Test list subnets without data"""
|
47 |
response = self.get(SUBNETS_URL)
|
48 |
self.assertSuccess(response)
|
49 |
subnets = json.loads(response.content) |
50 |
self.assertEqual(subnets, {'subnets': []}) |
51 |
|
52 |
def test_list_subnets_data(self): |
53 |
"""Test list subnets with data"""
|
54 |
test_net = mf.NetworkFactory() |
55 |
test_subnet_ipv4 = mf.IPv4SubnetFactory(network=test_net) |
56 |
test_subnet_ipv6 = mf.IPv4SubnetFactory(network=test_net, ipversion=6,
|
57 |
cidr= |
58 |
'fd4b:638e:fd7a:f998::/64')
|
59 |
response = self.get(SUBNETS_URL, user=test_net.userid)
|
60 |
self.assertSuccess(response)
|
61 |
|
62 |
def test_get_subnet(self): |
63 |
"""Test get info of a single subnet"""
|
64 |
test_net = mf.NetworkFactory() |
65 |
test_subnet = mf.IPv4SubnetFactory(network=test_net) |
66 |
url = join_urls(SUBNETS_URL, str(test_subnet.id))
|
67 |
response = self.get(url, user=test_net.userid)
|
68 |
self.assertSuccess(response)
|
69 |
|
70 |
def test_get_subnet_404(self): |
71 |
"""Test get info of a subnet that doesn't exist"""
|
72 |
url = join_urls(SUBNETS_URL, '42')
|
73 |
response = self.get(url)
|
74 |
self.assertItemNotFound(response)
|
75 |
|
76 |
def test_subnet_delete(self): |
77 |
"""Test delete a subnet -- not supported"""
|
78 |
url = join_urls(SUBNETS_URL, '42')
|
79 |
response = self.delete(url)
|
80 |
self.assertBadRequest(response)
|
81 |
|
82 |
def test_create_subnet_success_ipv4(self): |
83 |
"""Test create a subnet successfully"""
|
84 |
test_net = mf.NetworkFactory() |
85 |
request = { |
86 |
'subnet': {
|
87 |
'network_id': test_net.id,
|
88 |
'cidr': '10.0.3.0/24', |
89 |
'ip_version': 4} |
90 |
} |
91 |
response = self.post(SUBNETS_URL, test_net.userid,
|
92 |
json.dumps(request), "json")
|
93 |
self.assertSuccess(response)
|
94 |
|
95 |
def test_create_subnet_success_ipv6(self): |
96 |
"""Test create an IPv6 subnet successfully"""
|
97 |
test_net = mf.NetworkFactory() |
98 |
request = { |
99 |
'subnet': {
|
100 |
'network_id': test_net.id,
|
101 |
'cidr': 'fdc1:4992:1130:fc0b::/64', |
102 |
'ip_version': 6} |
103 |
} |
104 |
response = self.post(SUBNETS_URL, test_net.userid,
|
105 |
json.dumps(request), "json")
|
106 |
self.assertSuccess(response)
|
107 |
|
108 |
def test_create_subnet_with_ip_pool_allocation(self): |
109 |
"""Test create a subnet with an IP pool"""
|
110 |
test_net = mf.NetworkFactory() |
111 |
request = { |
112 |
'subnet': {
|
113 |
'network_id': test_net.id,
|
114 |
'cidr': '10.0.3.0/24', |
115 |
'ip_version': 4, |
116 |
'allocation_pools': [{
|
117 |
'start': '10.0.3.2', |
118 |
'end': '10.0.3.252'} |
119 |
]} |
120 |
} |
121 |
response = self.post(SUBNETS_URL, test_net.userid,
|
122 |
json.dumps(request), "json")
|
123 |
self.assertSuccess(response)
|
124 |
|
125 |
def test_create_subnet_with_multiple_ip_pools(self): |
126 |
"""Test create a subnet with multiple IP pools"""
|
127 |
test_net = mf.NetworkFactory() |
128 |
request = { |
129 |
'subnet': {
|
130 |
'network_id': test_net.id,
|
131 |
'cidr': '10.0.3.0/24', |
132 |
'ip_version': 4, |
133 |
'allocation_pools': [{
|
134 |
'start': '10.0.3.2', |
135 |
'end': '10.0.3.100'}, { |
136 |
'start': '10.0.3.200', |
137 |
'end': '10.0.3.220'} |
138 |
]} |
139 |
} |
140 |
response = self.post(SUBNETS_URL, test_net.userid,
|
141 |
json.dumps(request), "json")
|
142 |
self.assertSuccess(response)
|
143 |
|
144 |
def test_create_subnet_with_gateway_inside_of_ip_pool_range(self): |
145 |
"""Test create a subnet with an IP pool outside of network range"""
|
146 |
test_net = mf.NetworkFactory() |
147 |
request = { |
148 |
'subnet': {
|
149 |
'network_id': test_net.id,
|
150 |
'cidr': '10.0.3.0/24', |
151 |
'ip_version': 4, |
152 |
'gateway_ip': '10.0.3.1', |
153 |
'allocation_pools': [{
|
154 |
'start': '10.0.3.0', |
155 |
'end': '10.0.3.255'} |
156 |
]} |
157 |
} |
158 |
response = self.post(SUBNETS_URL, test_net.userid,
|
159 |
json.dumps(request), "json")
|
160 |
self.assertConflict(response)
|
161 |
|
162 |
def test_create_subnet_with_ip_pool_outside_of_network_range(self): |
163 |
"""Test create a subnet with an IP pool outside of network range"""
|
164 |
test_net = mf.NetworkFactory() |
165 |
request = { |
166 |
'subnet': {
|
167 |
'network_id': test_net.id,
|
168 |
'cidr': '10.0.3.0/24', |
169 |
'ip_version': 4, |
170 |
'allocation_pools': [{
|
171 |
'start': '10.0.8.0', |
172 |
'end': '10.0.1.250'} |
173 |
]} |
174 |
} |
175 |
response = self.post(SUBNETS_URL, test_net.userid,
|
176 |
json.dumps(request), "json")
|
177 |
self.assertConflict(response)
|
178 |
|
179 |
def test_create_subnet_with_ip_pool_end_lower_than_start(self): |
180 |
"""Test create a subnet with a pool where end is lower than start"""
|
181 |
test_net = mf.NetworkFactory() |
182 |
request = { |
183 |
'subnet': {
|
184 |
'network_id': test_net.id,
|
185 |
'cidr': '10.0.3.0/24', |
186 |
'ip_version': 4, |
187 |
'allocation_pools': [{
|
188 |
'start': '10.0.1.10', |
189 |
'end': '10.0.1.5'} |
190 |
]} |
191 |
} |
192 |
response = self.post(SUBNETS_URL, test_net.userid,
|
193 |
json.dumps(request), "json")
|
194 |
self.assertConflict(response)
|
195 |
|
196 |
def test_create_subnet_with_ip_pool_in_a_ipv6_subnet(self): |
197 |
"""Test create a subnet with an ip pool, in an IPv6 subnet """
|
198 |
test_net = mf.NetworkFactory() |
199 |
request = { |
200 |
'subnet': {
|
201 |
'network_id': test_net.id,
|
202 |
'cidr': 'fd4b:638e:fd7a:f998::/64', |
203 |
'ip_version': 6, |
204 |
'allocation_pools': [{
|
205 |
'start': '10.0.1.10', |
206 |
'end': '10.0.1.5'} |
207 |
]} |
208 |
} |
209 |
response = self.post(SUBNETS_URL, test_net.userid,
|
210 |
json.dumps(request), "json")
|
211 |
self.assertConflict(response)
|
212 |
|
213 |
def test_create_subnet_with_invalid_network_id(self): |
214 |
"""Test create a subnet with a network id that doesn't exist"""
|
215 |
test_net = mf.NetworkFactory() |
216 |
request = { |
217 |
'subnet': {
|
218 |
'network_id': '420000', |
219 |
'cidr': '10.0.3.0/24', |
220 |
'ip_version': 4} |
221 |
} |
222 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
223 |
"json")
|
224 |
self.assertItemNotFound(response)
|
225 |
|
226 |
def test_create_subnet_with_malformed_ipversion(self): |
227 |
"""Create a subnet with a malformed ip_version type"""
|
228 |
test_net = mf.NetworkFactory() |
229 |
request = { |
230 |
'subnet': {
|
231 |
'network_id': test_net.id,
|
232 |
'cidr': '10.0.3.0/24', |
233 |
'ip_version': 8} |
234 |
} |
235 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
236 |
"json")
|
237 |
self.assertBadRequest(response)
|
238 |
|
239 |
def test_create_subnet_with_invalid_cidr(self): |
240 |
"""Create a subnet with an invalid cidr"""
|
241 |
test_net = mf.NetworkFactory() |
242 |
request = { |
243 |
'subnet': {
|
244 |
'network_id': test_net.id,
|
245 |
'cidr': '192.168.3.0/8'} |
246 |
} |
247 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
248 |
"json")
|
249 |
self.assertBadRequest(response)
|
250 |
|
251 |
def test_create_subnet_with_invalid_gateway(self): |
252 |
"""Create a subnet with a gateway outside of the subnet range"""
|
253 |
test_net = mf.NetworkFactory() |
254 |
request = { |
255 |
'subnet': {
|
256 |
'network_id': test_net.id,
|
257 |
'cidr': '192.168.3.0/24', |
258 |
'gateway_ip': '192.168.0.1'} |
259 |
} |
260 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
261 |
"json")
|
262 |
self.assertBadRequest(response)
|
263 |
|
264 |
def test_create_subnet_with_invalid_name(self): |
265 |
"""Create a subnet with an invalid subnet name"""
|
266 |
test_net = mf.NetworkFactory() |
267 |
request = { |
268 |
'subnet': {
|
269 |
'network_id': test_net.id,
|
270 |
'cidr': '192.168.3.0/24', |
271 |
'name': 'a' * 300} |
272 |
} |
273 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
274 |
"json")
|
275 |
self.assertBadRequest(response)
|
276 |
|
277 |
def test_create_subnet_with_invalid_dhcp(self): |
278 |
"""Create a subnet with an invalid dhcp value"""
|
279 |
test_net = mf.NetworkFactory() |
280 |
request = { |
281 |
'subnet': {
|
282 |
'network_id': test_net.id,
|
283 |
'cidr': '192.168.3.0/24', |
284 |
'enable_dhcp': 'None'} |
285 |
} |
286 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
287 |
"json")
|
288 |
self.assertBadRequest(response)
|
289 |
|
290 |
def test_create_subnet_with_dns_nameservers(self): |
291 |
"""Create a subnet with dns nameservers"""
|
292 |
test_net = mf.NetworkFactory() |
293 |
request = { |
294 |
'subnet': {
|
295 |
'network_id': test_net.id,
|
296 |
'cidr': '192.168.3.0/24', |
297 |
'dns_nameservers': ['8.8.8.8', '1.1.1.1']} |
298 |
} |
299 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
300 |
"json")
|
301 |
self.assertSuccess(response)
|
302 |
|
303 |
def test_create_subnet_with_host_routes(self): |
304 |
"""Create a subnet with dns nameservers"""
|
305 |
test_net = mf.NetworkFactory() |
306 |
request = { |
307 |
'subnet': {
|
308 |
'network_id': test_net.id,
|
309 |
'cidr': '192.168.3.0/24', |
310 |
'host_routes': ['8.8.8.8', '1.1.1.1']} |
311 |
} |
312 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
313 |
"json")
|
314 |
self.assertSuccess(response)
|
315 |
|
316 |
def test_create_subnet_with_same_ipversion(self): |
317 |
"""
|
318 |
Create a subnet in a network with another subnet of the same
|
319 |
ipversion type
|
320 |
"""
|
321 |
test_net = mf.NetworkFactory() |
322 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
323 |
request = { |
324 |
'subnet': {
|
325 |
'network_id': test_net.id,
|
326 |
'cidr': '192.168.3.0/24'} |
327 |
} |
328 |
response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
|
329 |
"json")
|
330 |
self.assertBadRequest(response)
|
331 |
|
332 |
def test_update_subnet_ip_version(self): |
333 |
"""Update the IP version of a subnet, raises 400 BadRequest"""
|
334 |
test_net = mf.NetworkFactory() |
335 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
336 |
request = { |
337 |
'subnet': {
|
338 |
'ip_version': '6'} |
339 |
} |
340 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
341 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
342 |
self.assertBadRequest(response)
|
343 |
|
344 |
def test_update_subnet_cidr(self): |
345 |
"""Update the cidr of a subnet, raises 400 BadRequest"""
|
346 |
test_net = mf.NetworkFactory() |
347 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
348 |
request = { |
349 |
'subnet': {
|
350 |
'cidr': '192.168.42.0/24'} |
351 |
} |
352 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
353 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
354 |
self.assertBadRequest(response)
|
355 |
|
356 |
def test_update_subnet_allocation_pools(self): |
357 |
"""Update the allocation pools of a subnet, raises 400 BadRequest"""
|
358 |
test_net = mf.NetworkFactory() |
359 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
360 |
request = { |
361 |
'subnet': {
|
362 |
'allocation_pools': [{
|
363 |
'start': '10.0.3.0', |
364 |
'end': '10.0.3.255'} |
365 |
]} |
366 |
} |
367 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
368 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
369 |
self.assertBadRequest(response)
|
370 |
|
371 |
def test_update_subnet_add_dns(self): |
372 |
"""Update the dns nameservers of a subnet, raises 400 BadRequest"""
|
373 |
test_net = mf.NetworkFactory() |
374 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
375 |
request = { |
376 |
'subnet': {
|
377 |
'dns_nameservers': ['8.8.8.8']} |
378 |
} |
379 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
380 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
381 |
self.assertBadRequest(response)
|
382 |
|
383 |
def test_update_subnet_add_host_routes(self): |
384 |
"""Update the host routes of a subnet, raises 400 BadRequest"""
|
385 |
test_net = mf.NetworkFactory() |
386 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
387 |
request = { |
388 |
'subnet': {
|
389 |
'host_routes': ['8.8.8.8']} |
390 |
} |
391 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
392 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
393 |
self.assertBadRequest(response)
|
394 |
|
395 |
def test_update_subnet_with_invalid_dhcp_value(self): |
396 |
"""Update a subnet with an invalid dhcp value"""
|
397 |
test_net = mf.NetworkFactory() |
398 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
399 |
request = { |
400 |
'subnet': {
|
401 |
'enable_dhcp': 'Random'} |
402 |
} |
403 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
404 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
405 |
self.assertBadRequest(response)
|
406 |
|
407 |
def test_update_subnet_with_invalid_name(self): |
408 |
"""Update a subnet with an invalid name value"""
|
409 |
test_net = mf.NetworkFactory() |
410 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
411 |
request = { |
412 |
'subnet': {
|
413 |
'name': 'a' * 300} |
414 |
} |
415 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
416 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
417 |
self.assertBadRequest(response)
|
418 |
|
419 |
def test_update_subnet_with_invalid_gateway(self): |
420 |
"""Update a subnet with an invalid gateway value"""
|
421 |
test_net = mf.NetworkFactory() |
422 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
423 |
request = { |
424 |
'subnet': {
|
425 |
'gateway_ip': '192.168.200.0/24'} |
426 |
} |
427 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
428 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
429 |
self.assertBadRequest(response)
|
430 |
|
431 |
def test_update_subnet_gateway(self): |
432 |
"""Update the gateway of a subnet"""
|
433 |
test_net = mf.NetworkFactory() |
434 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
435 |
request = { |
436 |
'subnet': {
|
437 |
'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)} |
438 |
} |
439 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
440 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
441 |
self.assertBadRequest(response)
|
442 |
|
443 |
def test_update_subnet_name(self): |
444 |
"""Update the name of a subnet"""
|
445 |
test_net = mf.NetworkFactory() |
446 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
447 |
request = { |
448 |
'subnet': {
|
449 |
'name': 'Updated Name'} |
450 |
} |
451 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
452 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
453 |
self.assertSuccess(response)
|
454 |
|
455 |
def test_update_subnet_dhcp(self): |
456 |
"""Update the dhcp flag of a subnet"""
|
457 |
test_net = mf.NetworkFactory() |
458 |
test_sub = mf.IPv4SubnetFactory(network=test_net) |
459 |
request = { |
460 |
'subnet': {
|
461 |
'enable_dhcp': False} |
462 |
} |
463 |
url = join_urls(SUBNETS_URL, str(test_sub.id))
|
464 |
response = self.put(url, test_net.userid, json.dumps(request), "json") |
465 |
self.assertBadRequest(response)
|