Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / api / tests / subnets.py @ 57c57b88

History | View | Annotate | Download (22.3 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
NETWORK_URL = get_service_path(cyclades_services, 'network', version='v2.0')
41
SUBNETS_URL = join_urls(NETWORK_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.assertSuccess201(response)
94
        resp = json.loads(response.content)['subnet']
95
        self.assertEqual("10.0.3.1", resp['gateway_ip'])
96
        self.assertEqual([{"start": "10.0.3.2", "end": "10.0.3.254"}],
97
                         resp['allocation_pools'])
98
        self.assertEqual(True, resp['enable_dhcp'])
99

    
100
    def test_create_subnet_success_ipv4_with_slaac(self):
101
        """Test create an IPv4 subnet, with a slaac that will be ingored"""
102
        test_net = mf.NetworkFactory()
103
        request = {
104
            'subnet': {
105
                'network_id': test_net.id,
106
                'cidr': '10.0.3.0/24',
107
                'ip_version': 4,
108
                'enable_slaac': False}
109
        }
110
        response = self.post(SUBNETS_URL, test_net.userid,
111
                             json.dumps(request), "json")
112
        self.assertSuccess201(response)
113

    
114
    def test_create_subnet_success_ipv6_with_slaac(self):
115
        """Test create a subnet with ipv6 and slaac"""
116
        test_net = mf.NetworkFactory()
117
        request = {
118
            'subnet': {
119
                'network_id': test_net.id,
120
                'cidr': 'fdc1:4992:1130:fc0b::/64',
121
                'ip_version': 6,
122
                'enable_slaac': False}
123
        }
124
        response = self.post(SUBNETS_URL, test_net.userid,
125
                             json.dumps(request), "json")
126
        self.assertSuccess201(response)
127
        resp = json.loads(response.content)['subnet']
128
        self.assertEqual("fdc1:4992:1130:fc0b::1", resp['gateway_ip'])
129
        self.assertEqual([], resp['allocation_pools'])
130
        self.assertEqual(False, resp['enable_slaac'])
131

    
132
    def test_create_subnet_with_malformed_slaac(self):
133
        """Test create a subnet with ipv6 and a malformed slaac"""
134
        test_net = mf.NetworkFactory()
135
        request = {
136
            'subnet': {
137
                'network_id': test_net.id,
138
                'cidr': 'fdc1:4992:1130:fc0b::/64',
139
                'ip_version': 6,
140
                'enable_slaac': 'Random'}
141
        }
142
        response = self.post(SUBNETS_URL, test_net.userid,
143
                             json.dumps(request), "json")
144
        self.assertBadRequest(response)
145

    
146
    def test_create_subnet_success_ipv6(self):
147
        """Test create an IPv6 subnet successfully"""
148
        test_net = mf.NetworkFactory()
149
        request = {
150
            'subnet': {
151
                'network_id': test_net.id,
152
                'cidr': 'fdc1:4992:1130:fc0b::/64',
153
                'ip_version': 6}
154
        }
155
        response = self.post(SUBNETS_URL, test_net.userid,
156
                             json.dumps(request), "json")
157
        self.assertSuccess201(response)
158

    
159
    def test_create_subnet_with_ip_pool_allocation(self):
160
        """Test create a subnet with an IP pool"""
161
        test_net = mf.NetworkFactory()
162
        request = {
163
            'subnet': {
164
                'network_id': test_net.id,
165
                'cidr': '10.0.3.0/24',
166
                'ip_version': 4,
167
                'allocation_pools': [{
168
                    'start': '10.0.3.2',
169
                    'end': '10.0.3.252'}
170
                ]}
171
        }
172
        response = self.post(SUBNETS_URL, test_net.userid,
173
                             json.dumps(request), "json")
174
        self.assertSuccess201(response)
175

    
176
    def test_create_subnet_with_multiple_ip_pools(self):
177
        """Test create a subnet with multiple IP pools"""
178
        test_net = mf.NetworkFactory()
179
        request = {
180
            'subnet': {
181
                'network_id': test_net.id,
182
                'cidr': '10.0.3.0/24',
183
                'ip_version': 4,
184
                'allocation_pools': [{
185
                    'start': '10.0.3.2',
186
                    'end': '10.0.3.100'}, {
187
                    'start': '10.0.3.200',
188
                    'end': '10.0.3.220'}
189
                ]}
190
        }
191
        response = self.post(SUBNETS_URL, test_net.userid,
192
                             json.dumps(request), "json")
193
        self.assertSuccess201(response)
194
        resp = json.loads(response.content)['subnet']
195
        self.assertEqual([{"start": "10.0.3.2", "end": "10.0.3.100"},
196
                          {"start": "10.0.3.200", "end": "10.0.3.220"}],
197
                         resp['allocation_pools'])
198

    
199
    def test_create_subnet_with_gateway(self):
200
        """Test create a subnet with a gateway"""
201
        test_net = mf.NetworkFactory()
202
        request = {
203
            'subnet': {
204
                'network_id': test_net.id,
205
                'cidr': '10.0.3.0/24',
206
                'ip_version': 4,
207
                'gateway_ip': '10.0.3.150'}
208
        }
209
        response = self.post(SUBNETS_URL, test_net.userid,
210
                             json.dumps(request), "json")
211
        self.assertSuccess201(response)
212
        resp = json.loads(response.content)['subnet']
213
        self.assertEqual("10.0.3.150", resp['gateway_ip'])
214
        self.assertEqual([{"start": "10.0.3.1", "end": "10.0.3.149"},
215
                          {"start": "10.0.3.151", "end": "10.0.3.254"}],
216
                         resp['allocation_pools'])
217

    
218
    def test_create_subnet_with_gateway_inside_of_ip_pool_range(self):
219
        """Test create a subnet with a gateway IP inside the IP pool range"""
220
        test_net = mf.NetworkFactory()
221
        request = {
222
            'subnet': {
223
                'network_id': test_net.id,
224
                'cidr': '10.0.3.0/24',
225
                'ip_version': 4,
226
                'gateway_ip': '10.0.3.1',
227
                'allocation_pools': [{
228
                    'start': '10.0.3.0',
229
                    'end': '10.0.3.255'}
230
                ]}
231
        }
232
        response = self.post(SUBNETS_URL, test_net.userid,
233
                             json.dumps(request), "json")
234
        self.assertConflict(response)
235

    
236
    def test_create_subnet_with_ip_pool_outside_of_network_range(self):
237
        """Test create a subnet with an IP pool outside of network range"""
238
        test_net = mf.NetworkFactory()
239
        request = {
240
            'subnet': {
241
                'network_id': test_net.id,
242
                'cidr': '10.0.3.0/24',
243
                'ip_version': 4,
244
                'allocation_pools': [{
245
                    'start': '10.0.8.0',
246
                    'end': '10.0.1.250'}
247
                ]}
248
        }
249
        response = self.post(SUBNETS_URL, test_net.userid,
250
                             json.dumps(request), "json")
251
        self.assertConflict(response)
252

    
253
    def test_create_subnet_with_gateway_as_the_last_ip_of_subnet(self):
254
        """Test create a subnet with a gateway, as the last IP of the subnet"""
255
        test_net = mf.NetworkFactory()
256
        request = {
257
            'subnet': {
258
                'network_id': test_net.id,
259
                'cidr': '10.0.3.0/24',
260
                'ip_version': 4,
261
                'gateway_ip': '10.0.3.254'}
262
        }
263
        response = self.post(SUBNETS_URL, test_net.userid,
264
                             json.dumps(request), "json")
265
        self.assertSuccess201(response)
266
        resp = json.loads(response.content)['subnet']
267
        self.assertEqual("10.0.3.254", resp['gateway_ip'])
268
        self.assertEqual([{"start": "10.0.3.1", "end": "10.0.3.253"}],
269
                         resp['allocation_pools'])
270

    
271
    def test_create_subnet_with_ip_pool_end_lower_than_start(self):
272
        """Test create a subnet with a pool where end is lower than start"""
273
        test_net = mf.NetworkFactory()
274
        request = {
275
            'subnet': {
276
                'network_id': test_net.id,
277
                'cidr': '10.0.3.0/24',
278
                'ip_version': 4,
279
                'allocation_pools': [{
280
                    'start': '10.0.1.10',
281
                    'end': '10.0.1.5'}
282
                ]}
283
        }
284
        response = self.post(SUBNETS_URL, test_net.userid,
285
                             json.dumps(request), "json")
286
        self.assertConflict(response)
287

    
288
    def test_create_subnet_with_ip_pool_in_a_ipv6_subnet(self):
289
        """Test create a subnet with an ip pool, in an IPv6 subnet """
290
        test_net = mf.NetworkFactory()
291
        request = {
292
            'subnet': {
293
                'network_id': test_net.id,
294
                'cidr': 'fd4b:638e:fd7a:f998::/64',
295
                'ip_version': 6,
296
                'allocation_pools': [{
297
                    'start': '10.0.1.10',
298
                    'end': '10.0.1.5'}
299
                ]}
300
        }
301
        response = self.post(SUBNETS_URL, test_net.userid,
302
                             json.dumps(request), "json")
303
        self.assertConflict(response)
304

    
305
    def test_create_subnet_with_invalid_network_id(self):
306
        """Test create a subnet with a network id that doesn't exist"""
307
        test_net = mf.NetworkFactory()
308
        request = {
309
            'subnet': {
310
                'network_id': '420000',
311
                'cidr': '10.0.3.0/24',
312
                'ip_version': 4}
313
        }
314
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
315
                             "json")
316
        self.assertItemNotFound(response)
317

    
318
    def test_create_subnet_with_malformed_ipversion(self):
319
        """Create a subnet with a malformed ip_version type"""
320
        test_net = mf.NetworkFactory()
321
        request = {
322
            'subnet': {
323
                'network_id': test_net.id,
324
                'cidr': '10.0.3.0/24',
325
                'ip_version': 8}
326
        }
327
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
328
                             "json")
329
        self.assertBadRequest(response)
330

    
331
    def test_create_subnet_with_invalid_cidr(self):
332
        """Create a subnet with an invalid cidr"""
333
        test_net = mf.NetworkFactory()
334
        request = {
335
            'subnet': {
336
                'network_id': test_net.id,
337
                'cidr': '192.168.3.0/8'}
338
        }
339
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
340
                             "json")
341
        self.assertBadRequest(response)
342

    
343
    def test_create_subnet_with_invalid_gateway(self):
344
        """Create a subnet with a gateway outside of the subnet range"""
345
        test_net = mf.NetworkFactory()
346
        request = {
347
            'subnet': {
348
                'network_id': test_net.id,
349
                'cidr': '192.168.3.0/24',
350
                'gateway_ip': '192.168.0.1'}
351
        }
352
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
353
                             "json")
354
        self.assertBadRequest(response)
355

    
356
    def test_create_subnet_with_invalid_name(self):
357
        """Create a subnet with an invalid subnet name"""
358
        test_net = mf.NetworkFactory()
359
        request = {
360
            'subnet': {
361
                'network_id': test_net.id,
362
                'cidr': '192.168.3.0/24',
363
                'name': 'a' * 300}
364
        }
365
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
366
                             "json")
367
        self.assertBadRequest(response)
368

    
369
    def test_create_subnet_with_invalid_dhcp(self):
370
        """Create a subnet with an invalid dhcp value"""
371
        test_net = mf.NetworkFactory()
372
        request = {
373
            'subnet': {
374
                'network_id': test_net.id,
375
                'cidr': '192.168.3.0/24',
376
                'enable_dhcp': 'None'}
377
        }
378
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
379
                             "json")
380
        self.assertBadRequest(response)
381

    
382
    def test_create_subnet_with_dhcp_set_to_false(self):
383
        """Create a subnet with a dhcp set to false"""
384
        test_net = mf.NetworkFactory()
385
        request = {
386
            'subnet': {
387
                'network_id': test_net.id,
388
                'cidr': '192.168.3.0/24',
389
                'enable_dhcp': False}
390
        }
391
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
392
                             "json")
393
        self.assertSuccess201(response)
394

    
395
    def test_create_subnet_with_dns_nameservers(self):
396
        """Create a subnet with dns nameservers"""
397
        test_net = mf.NetworkFactory()
398
        request = {
399
            'subnet': {
400
                'network_id': test_net.id,
401
                'cidr': '192.168.3.0/24',
402
                'dns_nameservers': ['8.8.8.8', '1.1.1.1']}
403
        }
404
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
405
                             "json")
406
        self.assertSuccess201(response)
407

    
408
    def test_create_subnet_with_host_routes(self):
409
        """Create a subnet with dns nameservers"""
410
        test_net = mf.NetworkFactory()
411
        request = {
412
            'subnet': {
413
                'network_id': test_net.id,
414
                'cidr': '192.168.3.0/24',
415
                'host_routes': ['8.8.8.8', '1.1.1.1']}
416
        }
417
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
418
                             "json")
419
        self.assertSuccess201(response)
420
        resp = json.loads(response.content)['subnet']
421
        self.assertEqual(["8.8.8.8", "1.1.1.1"], resp["host_routes"])
422

    
423
    def test_create_subnet_with_same_ipversion(self):
424
        """
425
        Create a subnet in a network with another subnet of the same
426
        ipversion type
427
        """
428
        test_net = mf.NetworkFactory()
429
        test_sub = mf.IPv4SubnetFactory(network=test_net)
430
        request = {
431
            'subnet': {
432
                'network_id': test_net.id,
433
                'cidr': '192.168.3.0/24'}
434
        }
435
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
436
                             "json")
437
        self.assertBadRequest(response)
438

    
439
    def test_update_subnet_ip_version(self):
440
        """Update the IP version of a subnet, raises 400 BadRequest"""
441
        test_net = mf.NetworkFactory()
442
        test_sub = mf.IPv4SubnetFactory(network=test_net)
443
        request = {
444
            'subnet': {
445
                'ip_version': '6'}
446
        }
447
        url = join_urls(SUBNETS_URL, str(test_sub.id))
448
        response = self.put(url, test_net.userid, json.dumps(request), "json")
449
        self.assertBadRequest(response)
450

    
451
    def test_update_subnet_cidr(self):
452
        """Update the cidr of a subnet, raises 400 BadRequest"""
453
        test_net = mf.NetworkFactory()
454
        test_sub = mf.IPv4SubnetFactory(network=test_net)
455
        request = {
456
            'subnet': {
457
                'cidr': '192.168.42.0/24'}
458
        }
459
        url = join_urls(SUBNETS_URL, str(test_sub.id))
460
        response = self.put(url, test_net.userid, json.dumps(request), "json")
461
        self.assertBadRequest(response)
462

    
463
    def test_update_subnet_allocation_pools(self):
464
        """Update the allocation pools of a subnet, raises 400 BadRequest"""
465
        test_net = mf.NetworkFactory()
466
        test_sub = mf.IPv4SubnetFactory(network=test_net)
467
        request = {
468
            'subnet': {
469
                'allocation_pools': [{
470
                    'start': '10.0.3.0',
471
                    'end': '10.0.3.255'}
472
                ]}
473
        }
474
        url = join_urls(SUBNETS_URL, str(test_sub.id))
475
        response = self.put(url, test_net.userid, json.dumps(request), "json")
476
        self.assertBadRequest(response)
477

    
478
    def test_update_subnet_add_dns(self):
479
        """Update the dns nameservers of a subnet, raises 400 BadRequest"""
480
        test_net = mf.NetworkFactory()
481
        test_sub = mf.IPv4SubnetFactory(network=test_net)
482
        request = {
483
            'subnet': {
484
                'dns_nameservers': ['8.8.8.8']}
485
        }
486
        url = join_urls(SUBNETS_URL, str(test_sub.id))
487
        response = self.put(url, test_net.userid, json.dumps(request), "json")
488
        self.assertBadRequest(response)
489

    
490
    def test_update_subnet_add_host_routes(self):
491
        """Update the host routes of a subnet, raises 400 BadRequest"""
492
        test_net = mf.NetworkFactory()
493
        test_sub = mf.IPv4SubnetFactory(network=test_net)
494
        request = {
495
            'subnet': {
496
                'host_routes': ['8.8.8.8']}
497
        }
498
        url = join_urls(SUBNETS_URL, str(test_sub.id))
499
        response = self.put(url, test_net.userid, json.dumps(request), "json")
500
        self.assertBadRequest(response)
501

    
502
    def test_update_subnet_with_invalid_dhcp_value(self):
503
        """Update a subnet with an invalid dhcp value"""
504
        test_net = mf.NetworkFactory()
505
        test_sub = mf.IPv4SubnetFactory(network=test_net)
506
        request = {
507
            'subnet': {
508
                'enable_dhcp': 'Random'}
509
        }
510
        url = join_urls(SUBNETS_URL, str(test_sub.id))
511
        response = self.put(url, test_net.userid, json.dumps(request), "json")
512
        self.assertBadRequest(response)
513

    
514
    def test_update_subnet_with_invalid_name(self):
515
        """Update a subnet with an invalid name value"""
516
        test_net = mf.NetworkFactory()
517
        test_sub = mf.IPv4SubnetFactory(network=test_net)
518
        request = {
519
            'subnet': {
520
                'name': 'a' * 300}
521
        }
522
        url = join_urls(SUBNETS_URL, str(test_sub.id))
523
        response = self.put(url, test_net.userid, json.dumps(request), "json")
524
        self.assertBadRequest(response)
525

    
526
    def test_update_subnet_with_invalid_gateway(self):
527
        """Update a subnet with an invalid gateway value"""
528
        test_net = mf.NetworkFactory()
529
        test_sub = mf.IPv4SubnetFactory(network=test_net)
530
        request = {
531
            'subnet': {
532
                'gateway_ip': '192.168.200.0/24'}
533
        }
534
        url = join_urls(SUBNETS_URL, str(test_sub.id))
535
        response = self.put(url, test_net.userid, json.dumps(request), "json")
536
        self.assertBadRequest(response)
537

    
538
    def test_update_subnet_gateway(self):
539
        """Update the gateway of a subnet"""
540
        test_net = mf.NetworkFactory()
541
        test_sub = mf.IPv4SubnetFactory(network=test_net)
542
        request = {
543
            'subnet': {
544
                'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)}
545
        }
546
        url = join_urls(SUBNETS_URL, str(test_sub.id))
547
        response = self.put(url, test_net.userid, json.dumps(request), "json")
548
        self.assertBadRequest(response)
549

    
550
    def test_update_subnet_name(self):
551
        """Update the name of a subnet"""
552
        test_net = mf.NetworkFactory()
553
        test_sub = mf.IPv4SubnetFactory(network=test_net)
554
        request = {
555
            'subnet': {
556
                'name': 'Updated Name'}
557
        }
558
        url = join_urls(SUBNETS_URL, str(test_sub.id))
559
        response = self.put(url, test_net.userid, json.dumps(request), "json")
560
        self.assertSuccess(response)
561

    
562
    def test_update_subnet_dhcp(self):
563
        """Update the dhcp flag of a subnet"""
564
        test_net = mf.NetworkFactory()
565
        test_sub = mf.IPv4SubnetFactory(network=test_net)
566
        request = {
567
            'subnet': {
568
                'enable_dhcp': False}
569
        }
570
        url = join_urls(SUBNETS_URL, str(test_sub.id))
571
        response = self.put(url, test_net.userid, json.dumps(request), "json")
572
        self.assertBadRequest(response)