Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (21.5 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
        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.assertSuccess(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.assertSuccess(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.assertSuccess(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.assertSuccess(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.assertSuccess(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.assertSuccess(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_ip_pool_end_lower_than_start(self):
254
        """Test create a subnet with a pool where end is lower than start"""
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
                'allocation_pools': [{
262
                    'start': '10.0.1.10',
263
                    'end': '10.0.1.5'}
264
                ]}
265
        }
266
        response = self.post(SUBNETS_URL, test_net.userid,
267
                             json.dumps(request), "json")
268
        self.assertConflict(response)
269

    
270
    def test_create_subnet_with_ip_pool_in_a_ipv6_subnet(self):
271
        """Test create a subnet with an ip pool, in an IPv6 subnet """
272
        test_net = mf.NetworkFactory()
273
        request = {
274
            'subnet': {
275
                'network_id': test_net.id,
276
                'cidr': 'fd4b:638e:fd7a:f998::/64',
277
                'ip_version': 6,
278
                'allocation_pools': [{
279
                    'start': '10.0.1.10',
280
                    'end': '10.0.1.5'}
281
                ]}
282
        }
283
        response = self.post(SUBNETS_URL, test_net.userid,
284
                             json.dumps(request), "json")
285
        self.assertConflict(response)
286

    
287
    def test_create_subnet_with_invalid_network_id(self):
288
        """Test create a subnet with a network id that doesn't exist"""
289
        test_net = mf.NetworkFactory()
290
        request = {
291
            'subnet': {
292
                'network_id': '420000',
293
                'cidr': '10.0.3.0/24',
294
                'ip_version': 4}
295
        }
296
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
297
                             "json")
298
        self.assertItemNotFound(response)
299

    
300
    def test_create_subnet_with_malformed_ipversion(self):
301
        """Create a subnet with a malformed ip_version type"""
302
        test_net = mf.NetworkFactory()
303
        request = {
304
            'subnet': {
305
                'network_id': test_net.id,
306
                'cidr': '10.0.3.0/24',
307
                'ip_version': 8}
308
        }
309
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
310
                             "json")
311
        self.assertBadRequest(response)
312

    
313
    def test_create_subnet_with_invalid_cidr(self):
314
        """Create a subnet with an invalid cidr"""
315
        test_net = mf.NetworkFactory()
316
        request = {
317
            'subnet': {
318
                'network_id': test_net.id,
319
                'cidr': '192.168.3.0/8'}
320
        }
321
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
322
                             "json")
323
        self.assertBadRequest(response)
324

    
325
    def test_create_subnet_with_invalid_gateway(self):
326
        """Create a subnet with a gateway outside of the subnet range"""
327
        test_net = mf.NetworkFactory()
328
        request = {
329
            'subnet': {
330
                'network_id': test_net.id,
331
                'cidr': '192.168.3.0/24',
332
                'gateway_ip': '192.168.0.1'}
333
        }
334
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
335
                             "json")
336
        self.assertBadRequest(response)
337

    
338
    def test_create_subnet_with_invalid_name(self):
339
        """Create a subnet with an invalid subnet name"""
340
        test_net = mf.NetworkFactory()
341
        request = {
342
            'subnet': {
343
                'network_id': test_net.id,
344
                'cidr': '192.168.3.0/24',
345
                'name': 'a' * 300}
346
        }
347
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
348
                             "json")
349
        self.assertBadRequest(response)
350

    
351
    def test_create_subnet_with_invalid_dhcp(self):
352
        """Create a subnet with an invalid dhcp value"""
353
        test_net = mf.NetworkFactory()
354
        request = {
355
            'subnet': {
356
                'network_id': test_net.id,
357
                'cidr': '192.168.3.0/24',
358
                'enable_dhcp': 'None'}
359
        }
360
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
361
                             "json")
362
        self.assertBadRequest(response)
363

    
364
    def test_create_subnet_with_dhcp_set_to_false(self):
365
        """Create a subnet with a dhcp set to false"""
366
        test_net = mf.NetworkFactory()
367
        request = {
368
            'subnet': {
369
                'network_id': test_net.id,
370
                'cidr': '192.168.3.0/24',
371
                'enable_dhcp': False}
372
        }
373
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
374
                             "json")
375
        self.assertSuccess(response)
376

    
377
    def test_create_subnet_with_dns_nameservers(self):
378
        """Create a subnet with dns nameservers"""
379
        test_net = mf.NetworkFactory()
380
        request = {
381
            'subnet': {
382
                'network_id': test_net.id,
383
                'cidr': '192.168.3.0/24',
384
                'dns_nameservers': ['8.8.8.8', '1.1.1.1']}
385
        }
386
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
387
                             "json")
388
        self.assertSuccess(response)
389

    
390
    def test_create_subnet_with_host_routes(self):
391
        """Create a subnet with dns nameservers"""
392
        test_net = mf.NetworkFactory()
393
        request = {
394
            'subnet': {
395
                'network_id': test_net.id,
396
                'cidr': '192.168.3.0/24',
397
                'host_routes': ['8.8.8.8', '1.1.1.1']}
398
        }
399
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
400
                             "json")
401
        self.assertSuccess(response)
402
        resp = json.loads(response.content)['subnet']
403
        self.assertEqual(["8.8.8.8", "1.1.1.1"], resp["host_routes"])
404

    
405
    def test_create_subnet_with_same_ipversion(self):
406
        """
407
        Create a subnet in a network with another subnet of the same
408
        ipversion type
409
        """
410
        test_net = mf.NetworkFactory()
411
        test_sub = mf.IPv4SubnetFactory(network=test_net)
412
        request = {
413
            'subnet': {
414
                'network_id': test_net.id,
415
                'cidr': '192.168.3.0/24'}
416
        }
417
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
418
                             "json")
419
        self.assertBadRequest(response)
420

    
421
    def test_update_subnet_ip_version(self):
422
        """Update the IP version of a subnet, raises 400 BadRequest"""
423
        test_net = mf.NetworkFactory()
424
        test_sub = mf.IPv4SubnetFactory(network=test_net)
425
        request = {
426
            'subnet': {
427
                'ip_version': '6'}
428
        }
429
        url = join_urls(SUBNETS_URL, str(test_sub.id))
430
        response = self.put(url, test_net.userid, json.dumps(request), "json")
431
        self.assertBadRequest(response)
432

    
433
    def test_update_subnet_cidr(self):
434
        """Update the cidr of a subnet, raises 400 BadRequest"""
435
        test_net = mf.NetworkFactory()
436
        test_sub = mf.IPv4SubnetFactory(network=test_net)
437
        request = {
438
            'subnet': {
439
                'cidr': '192.168.42.0/24'}
440
        }
441
        url = join_urls(SUBNETS_URL, str(test_sub.id))
442
        response = self.put(url, test_net.userid, json.dumps(request), "json")
443
        self.assertBadRequest(response)
444

    
445
    def test_update_subnet_allocation_pools(self):
446
        """Update the allocation pools of a subnet, raises 400 BadRequest"""
447
        test_net = mf.NetworkFactory()
448
        test_sub = mf.IPv4SubnetFactory(network=test_net)
449
        request = {
450
            'subnet': {
451
                'allocation_pools': [{
452
                    'start': '10.0.3.0',
453
                    'end': '10.0.3.255'}
454
                ]}
455
        }
456
        url = join_urls(SUBNETS_URL, str(test_sub.id))
457
        response = self.put(url, test_net.userid, json.dumps(request), "json")
458
        self.assertBadRequest(response)
459

    
460
    def test_update_subnet_add_dns(self):
461
        """Update the dns nameservers of a subnet, raises 400 BadRequest"""
462
        test_net = mf.NetworkFactory()
463
        test_sub = mf.IPv4SubnetFactory(network=test_net)
464
        request = {
465
            'subnet': {
466
                'dns_nameservers': ['8.8.8.8']}
467
        }
468
        url = join_urls(SUBNETS_URL, str(test_sub.id))
469
        response = self.put(url, test_net.userid, json.dumps(request), "json")
470
        self.assertBadRequest(response)
471

    
472
    def test_update_subnet_add_host_routes(self):
473
        """Update the host routes of a subnet, raises 400 BadRequest"""
474
        test_net = mf.NetworkFactory()
475
        test_sub = mf.IPv4SubnetFactory(network=test_net)
476
        request = {
477
            'subnet': {
478
                'host_routes': ['8.8.8.8']}
479
        }
480
        url = join_urls(SUBNETS_URL, str(test_sub.id))
481
        response = self.put(url, test_net.userid, json.dumps(request), "json")
482
        self.assertBadRequest(response)
483

    
484
    def test_update_subnet_with_invalid_dhcp_value(self):
485
        """Update a subnet with an invalid dhcp value"""
486
        test_net = mf.NetworkFactory()
487
        test_sub = mf.IPv4SubnetFactory(network=test_net)
488
        request = {
489
            'subnet': {
490
                'enable_dhcp': 'Random'}
491
        }
492
        url = join_urls(SUBNETS_URL, str(test_sub.id))
493
        response = self.put(url, test_net.userid, json.dumps(request), "json")
494
        self.assertBadRequest(response)
495

    
496
    def test_update_subnet_with_invalid_name(self):
497
        """Update a subnet with an invalid name value"""
498
        test_net = mf.NetworkFactory()
499
        test_sub = mf.IPv4SubnetFactory(network=test_net)
500
        request = {
501
            'subnet': {
502
                'name': 'a' * 300}
503
        }
504
        url = join_urls(SUBNETS_URL, str(test_sub.id))
505
        response = self.put(url, test_net.userid, json.dumps(request), "json")
506
        self.assertBadRequest(response)
507

    
508
    def test_update_subnet_with_invalid_gateway(self):
509
        """Update a subnet with an invalid gateway value"""
510
        test_net = mf.NetworkFactory()
511
        test_sub = mf.IPv4SubnetFactory(network=test_net)
512
        request = {
513
            'subnet': {
514
                'gateway_ip': '192.168.200.0/24'}
515
        }
516
        url = join_urls(SUBNETS_URL, str(test_sub.id))
517
        response = self.put(url, test_net.userid, json.dumps(request), "json")
518
        self.assertBadRequest(response)
519

    
520
    def test_update_subnet_gateway(self):
521
        """Update the gateway of a subnet"""
522
        test_net = mf.NetworkFactory()
523
        test_sub = mf.IPv4SubnetFactory(network=test_net)
524
        request = {
525
            'subnet': {
526
                'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)}
527
        }
528
        url = join_urls(SUBNETS_URL, str(test_sub.id))
529
        response = self.put(url, test_net.userid, json.dumps(request), "json")
530
        self.assertBadRequest(response)
531

    
532
    def test_update_subnet_name(self):
533
        """Update the name of a subnet"""
534
        test_net = mf.NetworkFactory()
535
        test_sub = mf.IPv4SubnetFactory(network=test_net)
536
        request = {
537
            'subnet': {
538
                'name': 'Updated Name'}
539
        }
540
        url = join_urls(SUBNETS_URL, str(test_sub.id))
541
        response = self.put(url, test_net.userid, json.dumps(request), "json")
542
        self.assertSuccess(response)
543

    
544
    def test_update_subnet_dhcp(self):
545
        """Update the dhcp flag of a subnet"""
546
        test_net = mf.NetworkFactory()
547
        test_sub = mf.IPv4SubnetFactory(network=test_net)
548
        request = {
549
            'subnet': {
550
                'enable_dhcp': False}
551
        }
552
        url = join_urls(SUBNETS_URL, str(test_sub.id))
553
        response = self.put(url, test_net.userid, json.dumps(request), "json")
554
        self.assertBadRequest(response)