Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / neutron / tests / api.py @ 5cf968ab

History | View | Annotate | Download (20 kB)

1
from snf_django.utils.testing import BaseAPITest
2
from django.utils import simplejson as json
3
from synnefo.cyclades_settings import cyclades_services
4
from synnefo.lib.services import get_service_path
5
from synnefo.lib import join_urls
6
import json
7
import synnefo.neutron.models_factory as mf
8
import synnefo.db.models_factory as dbmf
9

    
10
NEUTRON_URL = get_service_path(cyclades_services, "neutron", "v2.0")
11
NETWORKS_URL = join_urls(NEUTRON_URL, "networks/")
12
SUBNETS_URL = join_urls(NEUTRON_URL, "subnets/")
13
ROUTERS_URL = join_urls(NEUTRON_URL, "routers/")
14

    
15

    
16
class NetworkTest(BaseAPITest):
17

    
18
    def test_list_networks(self):
19
        response = self.get(NETWORKS_URL)
20
        self.assertSuccess(response)
21
        networks = json.loads(response.content)
22
        self.assertEqual(networks, {"networks": []})
23

    
24
    def test_create_network(self):
25
        request = {
26
            "network": {
27
                "name": "sample_network"
28
                }
29
        }
30
        response = self.post(NETWORKS_URL, params=json.dumps(request))
31
        code = response.status_code
32
        self.assertEqual(code, 201)
33

    
34
    def test_create_network_no_name(self):
35
        response = self.post(NETWORKS_URL, params="")
36
        code = response.status_code
37
        self.assertEqual(code, 201)
38

    
39
    def test_get_unfound_network(self):
40
        url = join_urls(NETWORKS_URL, "123")
41
        response = self.get(url)
42
        self.assertItemNotFound(response)
43

    
44
    def test_get_network(self):
45
        test_net = mf.NetworkFactory.create()
46
        url = join_urls(NETWORKS_URL, str(test_net.id))
47
        response = self.get(url, user=test_net.userid)
48
        # validate response
49
        res = json.loads(response.content)
50
        net = res['network']
51
        keys = net.keys()
52
        must_send = set(["status", "subnets", "name", "admin_state_up",
53
                        "tenant_id", "id"])
54
        self.assertEqual(set(keys).issuperset(must_send), True)
55
        self.assertEqual(response.status_code, 200)
56

    
57
    def test_get_deleted_network(self):
58
        test_net = mf.NetworkFactory.create()
59
        url = join_urls(NETWORKS_URL, str(test_net.id))
60
        self.delete(url, user=test_net.userid)
61
        response = self.get(url, user=test_net.userid)
62
        self.assertEqual(response.status_code, 400)
63

    
64
    def test_delete_unfound_network(self):
65
        url = join_urls(NETWORKS_URL, "123")
66
        response = self.delete(url)
67
        self.assertItemNotFound(response)
68

    
69
    def test_delete_network(self):
70
        test_net = mf.NetworkFactory.create()
71
        url = join_urls(NETWORKS_URL, str(test_net.id))
72
        response = self.delete(url, user=test_net.userid)
73
        self.assertEqual(response.status_code, 204)
74

    
75
    def test_delete_network_in_use(self):
76
        test_net = mf.NetworkFactory.create()
77
        test_iface = mf.NetworkInterfaceFactory.create(network=test_net)
78
        url = join_urls(NETWORKS_URL, str(test_net.id))
79
        response = self.delete(url, user=test_net.userid)
80
        self.assertEqual(response.status_code, 409)
81

    
82
    def test_put_unfound_network(self):
83
        url = join_urls(NETWORKS_URL, "123")
84
        response = self.delete(url)
85
        self.assertItemNotFound(response)
86

    
87
    def test_put_network(self):
88
        test_net = mf.NetworkFactory.create()
89
        url = join_urls(NETWORKS_URL, str(test_net.id))
90
        request = {
91
            "network": {
92
                "name": "new_name"}
93
        }
94
        response = self.put(url, params=json.dumps(request),
95
                            user=test_net.userid)
96
        self.assertEqual(response.status_code, 200)
97

    
98
    def test_put_network_wrong_data(self):
99
        test_net = mf.NetworkFactory.create()
100
        url = join_urls(NETWORKS_URL, str(test_net.id))
101
        request = {
102
            "network": {
103
                "wrong_field": "new_name"}
104
        }
105
        response = self.put(url, params=json.dumps(request),
106
                            user=test_net.userid)
107
        self.assertEqual(response.status_code, 400)
108

    
109
    def test_put_no_data(self):
110
        test_net = mf.NetworkFactory.create()
111
        url = join_urls(NETWORKS_URL, str(test_net.id))
112
        response = self.put(url, params="", user=test_net.userid)
113
        self.assertEqual(response.status_code, 400)
114

    
115

    
116
class RouterTest(BaseAPITest):
117
    def test_list_empty_routers(self):
118
        response = self.get(ROUTERS_URL)
119
        self.assertSuccess(response)
120
        routers = json.loads(response.content)
121
        self.assertEqual(routers, {"routers": []})
122

    
123
    def test_create_router(self):
124
        pass
125

    
126
    def test_get_router(self):
127
        router = dbmf.VirtualMachineFactory.create(router=True)
128
        net = mf.NetworkFactory.create(public=True)
129
        nic = mf.NetworkInterfaceFactory.create(network=net, machine=router)
130
        response = self.get(ROUTERS_URL, user=router.userid)
131
        self.assertSuccess(response)
132

    
133
    def test_delete_router(self):
134
        router = dbmf.VirtualMachineFactory.create(router=True)
135
        url = join_urls(ROUTERS_URL, str(router.id))
136
        response = self.delete(url, user=router.userid)
137
        self.assertEqual(response.status_code, 204)
138

    
139
    def test_delete_router_with_private_net(self):
140
        router = dbmf.VirtualMachineFactory.create(router=True)
141
        net = mf.NetworkFactory.create(public=False)
142
        nic = mf.NetworkInterfaceFactory.create(network=net, machine=router)
143
        url = join_urls(ROUTERS_URL, str(router.id))
144
        response = self.delete(url, user=router.userid)
145
        self.assertEqual(response.status_code, 409)
146

    
147
    def test_update_router(self):
148
        router = dbmf.VirtualMachineFactory.create(router=True)
149
        net1 = mf.NetworkFactory.create(public=True)
150
        net2 = mf.NetworkFactory.create(public=True)
151
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router)
152
        request = {
153
            "router": {
154
                "name": "new_name",
155
                "external_gateway_info": {"network_id": net2.id}
156
                }
157
            }
158
        url = join_urls(ROUTERS_URL, str(router.id))
159
        response = self.put(url, params=json.dumps(request),
160
                            user=router.userid)
161
        info = json.loads(response.content)
162
        self.assertEqual(info['router']['external_gateway_info']['network_id'],
163
                         net2.id)
164
        self.assertEqual(info['router']['name'], "new_name")
165

    
166
    def test_remove_interface_no_body(self):
167
        url = join_urls(join_urls(ROUTERS_URL, "123"),
168
                        "remove_router_interface")
169

    
170
        response = self.put(url, params="")
171
        self.assertEqual(response.status_code, 400)
172

    
173
    def test_remove_interface_unfound_port(self):
174
        router = dbmf.VirtualMachineFactory.create(router=True)
175
        request = {"port_id": "123"}
176
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
177
                        "remove_router_interface")
178
        response = self.put(url, params=json.dumps(request),
179
                            user=router.userid)
180
        self.assertEqual(response.status_code, 404)
181

    
182
    def test_remove_interface_unfound_subnet(self):
183
        router = dbmf.VirtualMachineFactory.create(router=True)
184
        request = {"subnet_id": "123"}
185
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
186
                        "remove_router_interface")
187
        response = self.put(url, params=json.dumps(request),
188
                            user=router.userid)
189
        self.assertEqual(response.status_code, 404)
190

    
191
    def test_remove_interface_no_info(self):
192
        router = dbmf.VirtualMachineFactory.create(router=True)
193
        request = {"wrong": "123"}
194
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
195
                        "remove_router_interface")
196
        response = self.put(url, params=json.dumps(request),
197
                            user=router.userid)
198
        self.assertEqual(response.status_code, 400)
199

    
200
    def test_remove_interface_both_info_wrong(self):
201
        router = dbmf.VirtualMachineFactory.create(router=True)
202
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
203
        subnet = mf.SubnetFactory.create(network=net1)
204
        subnet2 = mf.SubnetFactory.create(network=net1)
205
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router,
206
                                                subnet=subnet2)
207
        request = {"subnet_id": subnet.id, "port_id": nic.id}
208
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
209
                        "remove_router_interface")
210
        response = self.put(url, params=json.dumps(request),
211
                            user=router.userid)
212
        self.assertEqual(response.status_code, 409)
213

    
214
    def test_remove_interface_subnet(self):
215
        router = dbmf.VirtualMachineFactory.create(router=True)
216
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
217
        subnet = mf.SubnetFactory.create(network=net1)
218
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router,
219
                                                subnet=subnet)
220
        request = {"subnet_id": subnet.id}
221
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
222
                        "remove_router_interface")
223
        response = self.put(url, params=json.dumps(request),
224
                            user=router.userid)
225
        self.assertEqual(response.status_code, 200)
226

    
227
    def test_remove_interface_port(self):
228
        router = dbmf.VirtualMachineFactory.create(router=True)
229
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
230
        subnet = mf.SubnetFactory.create(network=net1)
231
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router,
232
                                                subnet=subnet)
233
        request = {"port_id": nic.id}
234
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
235
                        "remove_router_interface")
236
        response = self.put(url, params=json.dumps(request),
237
                            user=router.userid)
238
        self.assertEqual(response.status_code, 200)
239

    
240
    def test_remove_interface_both_info(self):
241
        router = dbmf.VirtualMachineFactory.create(router=True)
242
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
243
        subnet = mf.SubnetFactory.create(network=net1)
244
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router,
245
                                                subnet=subnet)
246
        request = {"subnet_id": subnet.id, "port_id": nic.id}
247
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
248
                        "remove_router_interface")
249
        response = self.put(url, params=json.dumps(request),
250
                            user=router.userid)
251
        self.assertEqual(response.status_code, 200)
252

    
253
    def test_add_interface_no_info(self):
254
        url = join_urls(join_urls(ROUTERS_URL, "123"), "add_router_interface")
255
        response = self.put(url, params="")
256
        self.assertEqual(response.status_code, 400)
257

    
258
    def test_add_interface_wrong_info(self):
259
        router = dbmf.VirtualMachineFactory.create(router=True)
260
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
261
                        "add_router_interface")
262
        request = {}
263
        response = self.put(url, params=json.dumps(request),
264
                            user=router.userid)
265
        self.assertEqual(response.status_code, 400)
266

    
267
    def test_add_interface_unfound_port(self):
268
        router = dbmf.VirtualMachineFactory.create(router=True)
269
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
270
                        "add_router_interface")
271
        request = {"port_id": "123"}
272
        response = self.put(url, params=json.dumps(request),
273
                            user=router.userid)
274
        self.assertEqual(response.status_code, 404)
275

    
276
    def test_add_interface_unfound_subnet(self):
277
        router = dbmf.VirtualMachineFactory.create(router=True)
278
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
279
                        "add_router_interface")
280
        request = {"subnet_id": "123"}
281
        response = self.put(url, params=json.dumps(request),
282
                            user=router.userid)
283
        self.assertEqual(response.status_code, 404)
284

    
285
    def test_add_interface_both(self):
286
        router = dbmf.VirtualMachineFactory.create(router=True)
287
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
288
        subnet = mf.SubnetFactory.create(network=net1)
289
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
290
                        "add_router_interface")
291
        request = {"subnet_id": subnet.id, "port_id": "123"}
292
        response = self.put(url, params=json.dumps(request),
293
                            user=router.userid)
294
        self.assertEqual(response.status_code, 400)
295

    
296
    def test_add_interface_subnet(self):
297
        router = dbmf.VirtualMachineFactory.create(router=True)
298
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
299
        subnet = mf.SubnetFactory.create(network=net1)
300
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
301
                        "add_router_interface")
302
        request = {"subnet_id": subnet.id}
303
        response = self.put(url, params=json.dumps(request),
304
                            user=router.userid)
305
        self.assertEqual(response.status_code, 200)
306

    
307
    def test_add_interface_port_no_ip(self):
308
        router = dbmf.VirtualMachineFactory.create(router=True)
309
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
310
        subnet = mf.SubnetFactory.create(network=net1)
311
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=router,
312
                                                subnet=subnet)
313
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
314
                        "add_router_interface")
315
        request = {"port_id": nic.id}
316
        response = self.put(url, params=json.dumps(request),
317
                            user=router.userid)
318
        self.assertEqual(response.status_code, 400)
319

    
320
    def test_add_interface_port_connected(self):
321
        router = dbmf.VirtualMachineFactory.create(router=True)
322
        vm = dbmf.VirtualMachineFactory.create()
323
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
324
        subnet = mf.SubnetFactory.create(network=net1)
325
        nic = mf.NetworkInterfaceFactory.create(network=net1, machine=vm,
326
                                                subnet=subnet,
327
                                                ipv4="192.168.0.1")
328
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
329
                        "add_router_interface")
330
        request = {"port_id": nic.id}
331
        response = self.put(url, params=json.dumps(request),
332
                            user=router.userid)
333
        self.assertEqual(response.status_code, 409)
334

    
335
    def test_add_interface_port(self):
336
        router = dbmf.VirtualMachineFactory.create(router=True)
337
        net1 = mf.NetworkFactory.create(public=True, userid=router.userid)
338
        subnet = mf.SubnetFactory.create(network=net1)
339
        nic = mf.NetworkInterfaceFactory.create(network=net1,
340
                                                machine=None,
341
                                                subnet=subnet,
342
                                                ipv4="192.168.0.1")
343
        url = join_urls(join_urls(ROUTERS_URL, str(router.id)),
344
                        "add_router_interface")
345
        request = {"port_id": nic.id}
346
        response = self.put(url, params=json.dumps(request),
347
                            user=router.userid)
348
        self.assertEqual(response.status_code, 200)
349

    
350

    
351
class SubnetTest(BaseAPITest):
352
    def test_list_subnets(self):
353
        '''Test list subnets without data'''
354
        response = self.get(SUBNETS_URL)
355
        self.assertSuccess(response)
356
        subnets = json.loads(response.content)
357
        self.assertEqual(subnets, {'subnets': []})
358

    
359
    def test_list_subnets_data(self):
360
        '''Test list subnets with data'''
361
        test_net = mf.NetworkFactory()
362
        test_subnet_ipv4 = mf.SubnetFactory(network=test_net)
363
        test_subnet_ipv6 = mf.SubnetFactory(network=test_net, ipversion=6,
364
                                            cidr='2620: 0:2d0:200::7/32')
365
        response = self.get(SUBNETS_URL, user=test_net.userid)
366
        self.assertSuccess(response)
367

    
368
    def test_get_subnet(self):
369
        '''Test get info of a single subnet'''
370
        test_net = mf.NetworkFactory()
371
        test_subnet = mf.SubnetFactory(network=test_net)
372
        url = join_urls(SUBNETS_URL, str(test_subnet.id))
373
        response = self.get(url, user=test_net.userid)
374
        self.assertSuccess(response)
375

    
376
    def test_get_subnet_404(self):
377
        '''Test get info of a subnet that doesn't exist'''
378
        url = join_urls(SUBNETS_URL, '42')
379
        response = self.get(url)
380
        self.assertItemNotFound(response)
381

    
382
    def test_subnet_delete(self):
383
        '''Test delete a subnet -- not supported'''
384
        url = join_urls(SUBNETS_URL, '42')
385
        response = self.delete(url)
386
        self.assertBadRequest(response)
387

    
388
   # def test_create_subnet_success(self):
389
   #     '''Test create a subnet successfully'''
390
   #     test_net = mf.NetworkFactory()
391
   #     request = {
392
   #         'subnet': {
393
   #             'network_id': test_net.id,
394
   #             'cidr': '10.0.3.0/24',
395
   #             'ip_version': 4}
396
   #     }
397
   #     response = self.post(SUBNETS_URL, test_net.userid,
398
   #                          json.dumps(request), "json")
399
   #     self.assertSuccess(response)
400

    
401
    def test_create_subnet_with_invalid_network_id(self):
402
        '''Test create a subnet with a network id that doesn't exist'''
403
        test_net = mf.NetworkFactory()
404
        request = {
405
            'subnet': {
406
                'network_id': '42',
407
                'cidr': '10.0.3.0/24',
408
                'ip_version': 4}
409
        }
410
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
411
                             "json")
412
        self.assertItemNotFound(response)
413

    
414
    def test_create_subnet_with_malformed_ipversion(self):
415
        '''Create a subnet with a malformed ip_version type'''
416
        test_net = mf.NetworkFactory()
417
        request = {
418
            'subnet': {
419
                'network_id': test_net.id,
420
                'cidr': '10.0.3.0/24',
421
                'ip_version': 8}
422
        }
423
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
424
                             "json")
425
        self.assertBadRequest(response)
426

    
427
    def test_create_subnet_with_invalid_cidr(self):
428
        '''Create a subnet with an invalid cidr'''
429
        test_net = mf.NetworkFactory()
430
        request = {
431
            'subnet': {
432
                'network_id': test_net.id,
433
                'cidr': '192.168.3.0/8'}
434
        }
435
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
436
                             "json")
437
        self.assertBadRequest(response)
438

    
439
    def test_create_subnet_with_invalid_gateway(self):
440
        '''Create a subnet with a gateway outside of the subnet range'''
441
        test_net = mf.NetworkFactory()
442
        request = {
443
            'subnet': {
444
                'network_id': test_net.id,
445
                'cidr': '192.168.3.0/24',
446
                'gateway_ip': '192.168.0.1'}
447
        }
448
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
449
                             "json")
450
        self.assertBadRequest(response)
451

    
452
    def test_create_subnet_with_invalid_name(self):
453
        '''Create a subnet with an invalid subnet name'''
454
        test_net = mf.NetworkFactory()
455
        request = {
456
            'subnet': {
457
                'network_id': test_net.id,
458
                'cidr': '192.168.3.0/24',
459
                'name': 'a' * 300}
460
        }
461
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
462
                             "json")
463
        self.assertBadRequest(response)
464

    
465
    def test_create_subnet_with_invalid_dhcp(self):
466
        '''Create a subnet with an invalid dhcp value'''
467
        test_net = mf.NetworkFactory()
468
        request = {
469
            'subnet': {
470
                'network_id': test_net.id,
471
                'cidr': '192.168.3.0/24',
472
                'enable_dhcp': 'None'}
473
        }
474
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
475
                             "json")
476
        self.assertBadRequest(response)