Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / api / tests / subnets.py @ 097009bf

History | View | Annotate | Download (19.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.IPv6SubnetFactory(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_ipv4_with_slac(self):
96
        """Test create an IPv4 subnet, with a slac that will be ingored"""
97
        test_net = mf.NetworkFactory()
98
        request = {
99
            'subnet': {
100
                'network_id': test_net.id,
101
                'cidr': '10.0.3.0/24',
102
                'ip_version': 4,
103
                'enable_slac': False}
104
        }
105
        response = self.post(SUBNETS_URL, test_net.userid,
106
                             json.dumps(request), "json")
107
        self.assertSuccess(response)
108

    
109
    def test_create_subnet_success_ipv6_with_slac(self):
110
        """Test create a subnet with ipv6 and slac"""
111
        test_net = mf.NetworkFactory()
112
        request = {
113
            'subnet': {
114
                'network_id': test_net.id,
115
                'cidr': 'fdc1:4992:1130:fc0b::/64',
116
                'ip_version': 6,
117
                'enable_slac': False}
118
        }
119
        response = self.post(SUBNETS_URL, test_net.userid,
120
                             json.dumps(request), "json")
121
        self.assertSuccess(response)
122

    
123
    def test_create_subnet_with_malformed_slac(self):
124
        """Test create a subnet with ipv6 and a malformed slac"""
125
        test_net = mf.NetworkFactory()
126
        request = {
127
            'subnet': {
128
                'network_id': test_net.id,
129
                'cidr': 'fdc1:4992:1130:fc0b::/64',
130
                'ip_version': 6,
131
                'enable_slac': 'Random'}
132
        }
133
        response = self.post(SUBNETS_URL, test_net.userid,
134
                             json.dumps(request), "json")
135
        self.assertBadRequest(response)
136

    
137
    def test_create_subnet_success_ipv6(self):
138
        """Test create an IPv6 subnet successfully"""
139
        test_net = mf.NetworkFactory()
140
        request = {
141
            'subnet': {
142
                'network_id': test_net.id,
143
                'cidr': 'fdc1:4992:1130:fc0b::/64',
144
                'ip_version': 6}
145
        }
146
        response = self.post(SUBNETS_URL, test_net.userid,
147
                             json.dumps(request), "json")
148
        self.assertSuccess(response)
149

    
150
    def test_create_subnet_with_ip_pool_allocation(self):
151
        """Test create a subnet with an IP pool"""
152
        test_net = mf.NetworkFactory()
153
        request = {
154
            'subnet': {
155
                'network_id': test_net.id,
156
                'cidr': '10.0.3.0/24',
157
                'ip_version': 4,
158
                'allocation_pools': [{
159
                    'start': '10.0.3.2',
160
                    'end': '10.0.3.252'}
161
                ]}
162
        }
163
        response = self.post(SUBNETS_URL, test_net.userid,
164
                             json.dumps(request), "json")
165
        self.assertSuccess(response)
166

    
167
    def test_create_subnet_with_multiple_ip_pools(self):
168
        """Test create a subnet with multiple IP pools"""
169
        test_net = mf.NetworkFactory()
170
        request = {
171
            'subnet': {
172
                'network_id': test_net.id,
173
                'cidr': '10.0.3.0/24',
174
                'ip_version': 4,
175
                'allocation_pools': [{
176
                    'start': '10.0.3.2',
177
                    'end': '10.0.3.100'}, {
178
                    'start': '10.0.3.200',
179
                    'end': '10.0.3.220'}
180
                ]}
181
        }
182
        response = self.post(SUBNETS_URL, test_net.userid,
183
                             json.dumps(request), "json")
184
        self.assertSuccess(response)
185

    
186
    def test_create_subnet_with_gateway_inside_of_ip_pool_range(self):
187
        """Test create a subnet with an IP pool outside of network range"""
188
        test_net = mf.NetworkFactory()
189
        request = {
190
            'subnet': {
191
                'network_id': test_net.id,
192
                'cidr': '10.0.3.0/24',
193
                'ip_version': 4,
194
                'gateway_ip': '10.0.3.1',
195
                'allocation_pools': [{
196
                    'start': '10.0.3.0',
197
                    'end': '10.0.3.255'}
198
                ]}
199
        }
200
        response = self.post(SUBNETS_URL, test_net.userid,
201
                             json.dumps(request), "json")
202
        self.assertConflict(response)
203

    
204
    def test_create_subnet_with_ip_pool_outside_of_network_range(self):
205
        """Test create a subnet with an IP pool outside of network range"""
206
        test_net = mf.NetworkFactory()
207
        request = {
208
            'subnet': {
209
                'network_id': test_net.id,
210
                'cidr': '10.0.3.0/24',
211
                'ip_version': 4,
212
                'allocation_pools': [{
213
                    'start': '10.0.8.0',
214
                    'end': '10.0.1.250'}
215
                ]}
216
        }
217
        response = self.post(SUBNETS_URL, test_net.userid,
218
                             json.dumps(request), "json")
219
        self.assertConflict(response)
220

    
221
    def test_create_subnet_with_ip_pool_end_lower_than_start(self):
222
        """Test create a subnet with a pool where end is lower than start"""
223
        test_net = mf.NetworkFactory()
224
        request = {
225
            'subnet': {
226
                'network_id': test_net.id,
227
                'cidr': '10.0.3.0/24',
228
                'ip_version': 4,
229
                'allocation_pools': [{
230
                    'start': '10.0.1.10',
231
                    'end': '10.0.1.5'}
232
                ]}
233
        }
234
        response = self.post(SUBNETS_URL, test_net.userid,
235
                             json.dumps(request), "json")
236
        self.assertConflict(response)
237

    
238
    def test_create_subnet_with_ip_pool_in_a_ipv6_subnet(self):
239
        """Test create a subnet with an ip pool, in an IPv6 subnet """
240
        test_net = mf.NetworkFactory()
241
        request = {
242
            'subnet': {
243
                'network_id': test_net.id,
244
                'cidr': 'fd4b:638e:fd7a:f998::/64',
245
                'ip_version': 6,
246
                'allocation_pools': [{
247
                    'start': '10.0.1.10',
248
                    'end': '10.0.1.5'}
249
                ]}
250
        }
251
        response = self.post(SUBNETS_URL, test_net.userid,
252
                             json.dumps(request), "json")
253
        self.assertConflict(response)
254

    
255
    def test_create_subnet_with_invalid_network_id(self):
256
        """Test create a subnet with a network id that doesn't exist"""
257
        test_net = mf.NetworkFactory()
258
        request = {
259
            'subnet': {
260
                'network_id': '420000',
261
                'cidr': '10.0.3.0/24',
262
                'ip_version': 4}
263
        }
264
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
265
                             "json")
266
        self.assertItemNotFound(response)
267

    
268
    def test_create_subnet_with_malformed_ipversion(self):
269
        """Create a subnet with a malformed ip_version type"""
270
        test_net = mf.NetworkFactory()
271
        request = {
272
            'subnet': {
273
                'network_id': test_net.id,
274
                'cidr': '10.0.3.0/24',
275
                'ip_version': 8}
276
        }
277
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
278
                             "json")
279
        self.assertBadRequest(response)
280

    
281
    def test_create_subnet_with_invalid_cidr(self):
282
        """Create a subnet with an invalid cidr"""
283
        test_net = mf.NetworkFactory()
284
        request = {
285
            'subnet': {
286
                'network_id': test_net.id,
287
                'cidr': '192.168.3.0/8'}
288
        }
289
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
290
                             "json")
291
        self.assertBadRequest(response)
292

    
293
    def test_create_subnet_with_invalid_gateway(self):
294
        """Create a subnet with a gateway outside of the subnet range"""
295
        test_net = mf.NetworkFactory()
296
        request = {
297
            'subnet': {
298
                'network_id': test_net.id,
299
                'cidr': '192.168.3.0/24',
300
                'gateway_ip': '192.168.0.1'}
301
        }
302
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
303
                             "json")
304
        self.assertBadRequest(response)
305

    
306
    def test_create_subnet_with_invalid_name(self):
307
        """Create a subnet with an invalid subnet name"""
308
        test_net = mf.NetworkFactory()
309
        request = {
310
            'subnet': {
311
                'network_id': test_net.id,
312
                'cidr': '192.168.3.0/24',
313
                'name': 'a' * 300}
314
        }
315
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
316
                             "json")
317
        self.assertBadRequest(response)
318

    
319
    def test_create_subnet_with_invalid_dhcp(self):
320
        """Create a subnet with an invalid dhcp value"""
321
        test_net = mf.NetworkFactory()
322
        request = {
323
            'subnet': {
324
                'network_id': test_net.id,
325
                'cidr': '192.168.3.0/24',
326
                'enable_dhcp': 'None'}
327
        }
328
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
329
                             "json")
330
        self.assertBadRequest(response)
331

    
332
    def test_create_subnet_with_dhcp_set_to_false(self):
333
        """Create a subnet with a dhcp set to false"""
334
        test_net = mf.NetworkFactory()
335
        request = {
336
            'subnet': {
337
                'network_id': test_net.id,
338
                'cidr': '192.168.3.0/24',
339
                'enable_dhcp': False}
340
        }
341
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
342
                             "json")
343
        self.assertSuccess(response)
344

    
345
    def test_create_subnet_with_dns_nameservers(self):
346
        """Create a subnet with dns nameservers"""
347
        test_net = mf.NetworkFactory()
348
        request = {
349
            'subnet': {
350
                'network_id': test_net.id,
351
                'cidr': '192.168.3.0/24',
352
                'dns_nameservers': ['8.8.8.8', '1.1.1.1']}
353
        }
354
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
355
                             "json")
356
        self.assertSuccess(response)
357

    
358
    def test_create_subnet_with_host_routes(self):
359
        """Create a subnet with dns nameservers"""
360
        test_net = mf.NetworkFactory()
361
        request = {
362
            'subnet': {
363
                'network_id': test_net.id,
364
                'cidr': '192.168.3.0/24',
365
                'host_routes': ['8.8.8.8', '1.1.1.1']}
366
        }
367
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
368
                             "json")
369
        self.assertSuccess(response)
370

    
371
    def test_create_subnet_with_same_ipversion(self):
372
        """
373
        Create a subnet in a network with another subnet of the same
374
        ipversion type
375
        """
376
        test_net = mf.NetworkFactory()
377
        test_sub = mf.IPv4SubnetFactory(network=test_net)
378
        request = {
379
            'subnet': {
380
                'network_id': test_net.id,
381
                'cidr': '192.168.3.0/24'}
382
        }
383
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
384
                             "json")
385
        self.assertBadRequest(response)
386

    
387
    def test_update_subnet_ip_version(self):
388
        """Update the IP version of a subnet, raises 400 BadRequest"""
389
        test_net = mf.NetworkFactory()
390
        test_sub = mf.IPv4SubnetFactory(network=test_net)
391
        request = {
392
            'subnet': {
393
                'ip_version': '6'}
394
        }
395
        url = join_urls(SUBNETS_URL, str(test_sub.id))
396
        response = self.put(url, test_net.userid, json.dumps(request), "json")
397
        self.assertBadRequest(response)
398

    
399
    def test_update_subnet_cidr(self):
400
        """Update the cidr of a subnet, raises 400 BadRequest"""
401
        test_net = mf.NetworkFactory()
402
        test_sub = mf.IPv4SubnetFactory(network=test_net)
403
        request = {
404
            'subnet': {
405
                'cidr': '192.168.42.0/24'}
406
        }
407
        url = join_urls(SUBNETS_URL, str(test_sub.id))
408
        response = self.put(url, test_net.userid, json.dumps(request), "json")
409
        self.assertBadRequest(response)
410

    
411
    def test_update_subnet_allocation_pools(self):
412
        """Update the allocation pools of a subnet, raises 400 BadRequest"""
413
        test_net = mf.NetworkFactory()
414
        test_sub = mf.IPv4SubnetFactory(network=test_net)
415
        request = {
416
            'subnet': {
417
                'allocation_pools': [{
418
                    'start': '10.0.3.0',
419
                    'end': '10.0.3.255'}
420
                ]}
421
        }
422
        url = join_urls(SUBNETS_URL, str(test_sub.id))
423
        response = self.put(url, test_net.userid, json.dumps(request), "json")
424
        self.assertBadRequest(response)
425

    
426
    def test_update_subnet_add_dns(self):
427
        """Update the dns nameservers of a subnet, raises 400 BadRequest"""
428
        test_net = mf.NetworkFactory()
429
        test_sub = mf.IPv4SubnetFactory(network=test_net)
430
        request = {
431
            'subnet': {
432
                'dns_nameservers': ['8.8.8.8']}
433
        }
434
        url = join_urls(SUBNETS_URL, str(test_sub.id))
435
        response = self.put(url, test_net.userid, json.dumps(request), "json")
436
        self.assertBadRequest(response)
437

    
438
    def test_update_subnet_add_host_routes(self):
439
        """Update the host routes of a subnet, raises 400 BadRequest"""
440
        test_net = mf.NetworkFactory()
441
        test_sub = mf.IPv4SubnetFactory(network=test_net)
442
        request = {
443
            'subnet': {
444
                'host_routes': ['8.8.8.8']}
445
        }
446
        url = join_urls(SUBNETS_URL, str(test_sub.id))
447
        response = self.put(url, test_net.userid, json.dumps(request), "json")
448
        self.assertBadRequest(response)
449

    
450
    def test_update_subnet_with_invalid_dhcp_value(self):
451
        """Update a subnet with an invalid dhcp value"""
452
        test_net = mf.NetworkFactory()
453
        test_sub = mf.IPv4SubnetFactory(network=test_net)
454
        request = {
455
            'subnet': {
456
                'enable_dhcp': 'Random'}
457
        }
458
        url = join_urls(SUBNETS_URL, str(test_sub.id))
459
        response = self.put(url, test_net.userid, json.dumps(request), "json")
460
        self.assertBadRequest(response)
461

    
462
    def test_update_subnet_with_invalid_name(self):
463
        """Update a subnet with an invalid name value"""
464
        test_net = mf.NetworkFactory()
465
        test_sub = mf.IPv4SubnetFactory(network=test_net)
466
        request = {
467
            'subnet': {
468
                'name': 'a' * 300}
469
        }
470
        url = join_urls(SUBNETS_URL, str(test_sub.id))
471
        response = self.put(url, test_net.userid, json.dumps(request), "json")
472
        self.assertBadRequest(response)
473

    
474
    def test_update_subnet_with_invalid_gateway(self):
475
        """Update a subnet with an invalid gateway value"""
476
        test_net = mf.NetworkFactory()
477
        test_sub = mf.IPv4SubnetFactory(network=test_net)
478
        request = {
479
            'subnet': {
480
                'gateway_ip': '192.168.200.0/24'}
481
        }
482
        url = join_urls(SUBNETS_URL, str(test_sub.id))
483
        response = self.put(url, test_net.userid, json.dumps(request), "json")
484
        self.assertBadRequest(response)
485

    
486
    def test_update_subnet_gateway(self):
487
        """Update the gateway of a subnet"""
488
        test_net = mf.NetworkFactory()
489
        test_sub = mf.IPv4SubnetFactory(network=test_net)
490
        request = {
491
            'subnet': {
492
                'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)}
493
        }
494
        url = join_urls(SUBNETS_URL, str(test_sub.id))
495
        response = self.put(url, test_net.userid, json.dumps(request), "json")
496
        self.assertBadRequest(response)
497

    
498
    def test_update_subnet_name(self):
499
        """Update the name of a subnet"""
500
        test_net = mf.NetworkFactory()
501
        test_sub = mf.IPv4SubnetFactory(network=test_net)
502
        request = {
503
            'subnet': {
504
                'name': 'Updated Name'}
505
        }
506
        url = join_urls(SUBNETS_URL, str(test_sub.id))
507
        response = self.put(url, test_net.userid, json.dumps(request), "json")
508
        self.assertSuccess(response)
509

    
510
    def test_update_subnet_dhcp(self):
511
        """Update the dhcp flag of a subnet"""
512
        test_net = mf.NetworkFactory()
513
        test_sub = mf.IPv4SubnetFactory(network=test_net)
514
        request = {
515
            'subnet': {
516
                'enable_dhcp': False}
517
        }
518
        url = join_urls(SUBNETS_URL, str(test_sub.id))
519
        response = self.put(url, test_net.userid, json.dumps(request), "json")
520
        self.assertBadRequest(response)