Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (20.7 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_slac(self):
101
        """Test create an IPv4 subnet, with a slac 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_slac': 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_slac(self):
115
        """Test create a subnet with ipv6 and slac"""
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_slac': 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_slac'])
131

    
132
    def test_create_subnet_with_malformed_slac(self):
133
        """Test create a subnet with ipv6 and a malformed slac"""
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_slac': '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_inside_of_ip_pool_range(self):
200
        """Test create a subnet with an IP pool outside of network range"""
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.1',
208
                'allocation_pools': [{
209
                    'start': '10.0.3.0',
210
                    'end': '10.0.3.255'}
211
                ]}
212
        }
213
        response = self.post(SUBNETS_URL, test_net.userid,
214
                             json.dumps(request), "json")
215
        self.assertConflict(response)
216

    
217
    def test_create_subnet_with_ip_pool_outside_of_network_range(self):
218
        """Test create a subnet with an IP pool outside of network range"""
219
        test_net = mf.NetworkFactory()
220
        request = {
221
            'subnet': {
222
                'network_id': test_net.id,
223
                'cidr': '10.0.3.0/24',
224
                'ip_version': 4,
225
                'allocation_pools': [{
226
                    'start': '10.0.8.0',
227
                    'end': '10.0.1.250'}
228
                ]}
229
        }
230
        response = self.post(SUBNETS_URL, test_net.userid,
231
                             json.dumps(request), "json")
232
        self.assertConflict(response)
233

    
234
    def test_create_subnet_with_ip_pool_end_lower_than_start(self):
235
        """Test create a subnet with a pool where end is lower than start"""
236
        test_net = mf.NetworkFactory()
237
        request = {
238
            'subnet': {
239
                'network_id': test_net.id,
240
                'cidr': '10.0.3.0/24',
241
                'ip_version': 4,
242
                'allocation_pools': [{
243
                    'start': '10.0.1.10',
244
                    'end': '10.0.1.5'}
245
                ]}
246
        }
247
        response = self.post(SUBNETS_URL, test_net.userid,
248
                             json.dumps(request), "json")
249
        self.assertConflict(response)
250

    
251
    def test_create_subnet_with_ip_pool_in_a_ipv6_subnet(self):
252
        """Test create a subnet with an ip pool, in an IPv6 subnet """
253
        test_net = mf.NetworkFactory()
254
        request = {
255
            'subnet': {
256
                'network_id': test_net.id,
257
                'cidr': 'fd4b:638e:fd7a:f998::/64',
258
                'ip_version': 6,
259
                'allocation_pools': [{
260
                    'start': '10.0.1.10',
261
                    'end': '10.0.1.5'}
262
                ]}
263
        }
264
        response = self.post(SUBNETS_URL, test_net.userid,
265
                             json.dumps(request), "json")
266
        self.assertConflict(response)
267

    
268
    def test_create_subnet_with_invalid_network_id(self):
269
        """Test create a subnet with a network id that doesn't exist"""
270
        test_net = mf.NetworkFactory()
271
        request = {
272
            'subnet': {
273
                'network_id': '420000',
274
                'cidr': '10.0.3.0/24',
275
                'ip_version': 4}
276
        }
277
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
278
                             "json")
279
        self.assertItemNotFound(response)
280

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

    
294
    def test_create_subnet_with_invalid_cidr(self):
295
        """Create a subnet with an invalid cidr"""
296
        test_net = mf.NetworkFactory()
297
        request = {
298
            'subnet': {
299
                'network_id': test_net.id,
300
                'cidr': '192.168.3.0/8'}
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_gateway(self):
307
        """Create a subnet with a gateway outside of the subnet range"""
308
        test_net = mf.NetworkFactory()
309
        request = {
310
            'subnet': {
311
                'network_id': test_net.id,
312
                'cidr': '192.168.3.0/24',
313
                'gateway_ip': '192.168.0.1'}
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_name(self):
320
        """Create a subnet with an invalid subnet name"""
321
        test_net = mf.NetworkFactory()
322
        request = {
323
            'subnet': {
324
                'network_id': test_net.id,
325
                'cidr': '192.168.3.0/24',
326
                'name': 'a' * 300}
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_invalid_dhcp(self):
333
        """Create a subnet with an invalid dhcp value"""
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': 'None'}
340
        }
341
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
342
                             "json")
343
        self.assertBadRequest(response)
344

    
345
    def test_create_subnet_with_dhcp_set_to_false(self):
346
        """Create a subnet with a dhcp set to false"""
347
        test_net = mf.NetworkFactory()
348
        request = {
349
            'subnet': {
350
                'network_id': test_net.id,
351
                'cidr': '192.168.3.0/24',
352
                'enable_dhcp': False}
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_dns_nameservers(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
                'dns_nameservers': ['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_host_routes(self):
372
        """Create a subnet with dns nameservers"""
373
        test_net = mf.NetworkFactory()
374
        request = {
375
            'subnet': {
376
                'network_id': test_net.id,
377
                'cidr': '192.168.3.0/24',
378
                'host_routes': ['8.8.8.8', '1.1.1.1']}
379
        }
380
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
381
                             "json")
382
        self.assertSuccess(response)
383
        resp = json.loads(response.content)['subnet']
384
        self.assertEqual(["8.8.8.8", "1.1.1.1"], resp["host_routes"])
385

    
386
    def test_create_subnet_with_same_ipversion(self):
387
        """
388
        Create a subnet in a network with another subnet of the same
389
        ipversion type
390
        """
391
        test_net = mf.NetworkFactory()
392
        test_sub = mf.IPv4SubnetFactory(network=test_net)
393
        request = {
394
            'subnet': {
395
                'network_id': test_net.id,
396
                'cidr': '192.168.3.0/24'}
397
        }
398
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
399
                             "json")
400
        self.assertBadRequest(response)
401

    
402
    def test_update_subnet_ip_version(self):
403
        """Update the IP version of a subnet, raises 400 BadRequest"""
404
        test_net = mf.NetworkFactory()
405
        test_sub = mf.IPv4SubnetFactory(network=test_net)
406
        request = {
407
            'subnet': {
408
                'ip_version': '6'}
409
        }
410
        url = join_urls(SUBNETS_URL, str(test_sub.id))
411
        response = self.put(url, test_net.userid, json.dumps(request), "json")
412
        self.assertBadRequest(response)
413

    
414
    def test_update_subnet_cidr(self):
415
        """Update the cidr of a subnet, raises 400 BadRequest"""
416
        test_net = mf.NetworkFactory()
417
        test_sub = mf.IPv4SubnetFactory(network=test_net)
418
        request = {
419
            'subnet': {
420
                'cidr': '192.168.42.0/24'}
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_allocation_pools(self):
427
        """Update the allocation pools of a subnet, raises 400 BadRequest"""
428
        test_net = mf.NetworkFactory()
429
        test_sub = mf.IPv4SubnetFactory(network=test_net)
430
        request = {
431
            'subnet': {
432
                'allocation_pools': [{
433
                    'start': '10.0.3.0',
434
                    'end': '10.0.3.255'}
435
                ]}
436
        }
437
        url = join_urls(SUBNETS_URL, str(test_sub.id))
438
        response = self.put(url, test_net.userid, json.dumps(request), "json")
439
        self.assertBadRequest(response)
440

    
441
    def test_update_subnet_add_dns(self):
442
        """Update the dns nameservers of a subnet, raises 400 BadRequest"""
443
        test_net = mf.NetworkFactory()
444
        test_sub = mf.IPv4SubnetFactory(network=test_net)
445
        request = {
446
            'subnet': {
447
                'dns_nameservers': ['8.8.8.8']}
448
        }
449
        url = join_urls(SUBNETS_URL, str(test_sub.id))
450
        response = self.put(url, test_net.userid, json.dumps(request), "json")
451
        self.assertBadRequest(response)
452

    
453
    def test_update_subnet_add_host_routes(self):
454
        """Update the host routes of a subnet, raises 400 BadRequest"""
455
        test_net = mf.NetworkFactory()
456
        test_sub = mf.IPv4SubnetFactory(network=test_net)
457
        request = {
458
            'subnet': {
459
                'host_routes': ['8.8.8.8']}
460
        }
461
        url = join_urls(SUBNETS_URL, str(test_sub.id))
462
        response = self.put(url, test_net.userid, json.dumps(request), "json")
463
        self.assertBadRequest(response)
464

    
465
    def test_update_subnet_with_invalid_dhcp_value(self):
466
        """Update a subnet with an invalid dhcp value"""
467
        test_net = mf.NetworkFactory()
468
        test_sub = mf.IPv4SubnetFactory(network=test_net)
469
        request = {
470
            'subnet': {
471
                'enable_dhcp': 'Random'}
472
        }
473
        url = join_urls(SUBNETS_URL, str(test_sub.id))
474
        response = self.put(url, test_net.userid, json.dumps(request), "json")
475
        self.assertBadRequest(response)
476

    
477
    def test_update_subnet_with_invalid_name(self):
478
        """Update a subnet with an invalid name value"""
479
        test_net = mf.NetworkFactory()
480
        test_sub = mf.IPv4SubnetFactory(network=test_net)
481
        request = {
482
            'subnet': {
483
                'name': 'a' * 300}
484
        }
485
        url = join_urls(SUBNETS_URL, str(test_sub.id))
486
        response = self.put(url, test_net.userid, json.dumps(request), "json")
487
        self.assertBadRequest(response)
488

    
489
    def test_update_subnet_with_invalid_gateway(self):
490
        """Update a subnet with an invalid gateway value"""
491
        test_net = mf.NetworkFactory()
492
        test_sub = mf.IPv4SubnetFactory(network=test_net)
493
        request = {
494
            'subnet': {
495
                'gateway_ip': '192.168.200.0/24'}
496
        }
497
        url = join_urls(SUBNETS_URL, str(test_sub.id))
498
        response = self.put(url, test_net.userid, json.dumps(request), "json")
499
        self.assertBadRequest(response)
500

    
501
    def test_update_subnet_gateway(self):
502
        """Update the gateway of a subnet"""
503
        test_net = mf.NetworkFactory()
504
        test_sub = mf.IPv4SubnetFactory(network=test_net)
505
        request = {
506
            'subnet': {
507
                'gateway_ip': str(IPv4Network(test_sub.gateway).network + 1)}
508
        }
509
        url = join_urls(SUBNETS_URL, str(test_sub.id))
510
        response = self.put(url, test_net.userid, json.dumps(request), "json")
511
        self.assertBadRequest(response)
512

    
513
    def test_update_subnet_name(self):
514
        """Update the name of a subnet"""
515
        test_net = mf.NetworkFactory()
516
        test_sub = mf.IPv4SubnetFactory(network=test_net)
517
        request = {
518
            'subnet': {
519
                'name': 'Updated Name'}
520
        }
521
        url = join_urls(SUBNETS_URL, str(test_sub.id))
522
        response = self.put(url, test_net.userid, json.dumps(request), "json")
523
        self.assertSuccess(response)
524

    
525
    def test_update_subnet_dhcp(self):
526
        """Update the dhcp flag of a subnet"""
527
        test_net = mf.NetworkFactory()
528
        test_sub = mf.IPv4SubnetFactory(network=test_net)
529
        request = {
530
            'subnet': {
531
                'enable_dhcp': False}
532
        }
533
        url = join_urls(SUBNETS_URL, str(test_sub.id))
534
        response = self.put(url, test_net.userid, json.dumps(request), "json")
535
        self.assertBadRequest(response)