Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / neutron / tests / api.py @ 2e0916f2

History | View | Annotate | Download (11.5 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
class NetworkTest(BaseAPITest):
16

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

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

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

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

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

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

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

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

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

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

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

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

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

    
114

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

    
122
    def test_create_router(self):
123
        pass
124

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

    
132
    def test_delete_router(self):
133
        router = dbmf.VirtualMachineFactory.create(router=True)
134
        url = join_urls(ROUTERS_URL, str(router.id))
135
        response = self.delete(url, user=router.userid)
136
        print response.content
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
        print response.content
146
        self.assertEqual(response.status_code, 409)
147

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

    
166
    def test_remove_interface(self):
167
        url = join_urls(join_urls(ROUTERS_URL,"123"), "remove_router_interface")
168
        response = self.put(url, params="")
169
        print response.content
170

    
171
    def test_add_interface(self):
172
        url = join_urls(join_urls(ROUTERS_URL,"123"), "add_router_interface")
173
        request =  {"port_id": "a2f1f29d-571b-4533-907f-5803ab96ead1","subnet_id": "a2f1f29d-571b-4533-907f-5803ab96ead1"}
174
        print json.dumps(request)
175
        response = self.put(url, params=json.dumps(request))
176
        print response.content
177

    
178
class SubnetTest(BaseAPITest):
179
    def test_list_subnets(self):
180
        '''Test list subnets without data'''
181
        response = self.get(SUBNETS_URL)
182
        self.assertSuccess(response)
183
        subnets = json.loads(response.content)
184
        self.assertEqual(subnets, {'subnets': []})
185

    
186
    def test_list_subnets_data(self):
187
        '''Test list subnets with data'''
188
        test_net = mf.NetworkFactory()
189
        test_subnet_ipv4 = mf.SubnetFactory(network=test_net)
190
        test_subnet_ipv6 = mf.SubnetFactory(network=test_net, ipversion=6,
191
                                            cidr='2620:0:2d0:200::7/32')
192
        response = self.get(SUBNETS_URL, user=test_net.userid)
193
        self.assertSuccess(response)
194

    
195
    def test_get_subnet(self):
196
        '''Test get info of a single subnet'''
197
        test_net = mf.NetworkFactory()
198
        test_subnet = mf.SubnetFactory(network=test_net)
199
        url = join_urls(SUBNETS_URL, str(test_subnet.id))
200
        response = self.get(url, user=test_net.userid)
201
        self.assertSuccess(response)
202

    
203
    def test_get_subnet_404(self):
204
        '''Test get info of a subnet that doesn't exist'''
205
        url = join_urls(SUBNETS_URL, '42')
206
        response = self.get(url)
207
        self.assertItemNotFound(response)
208

    
209
    def test_subnet_delete(self):
210
        '''Test delete a subnet -- not supported'''
211
        url = join_urls(SUBNETS_URL, '42')
212
        response = self.delete(url)
213
        self.assertBadRequest(response)
214

    
215
   # def test_create_subnet_success(self):
216
   #     '''Test create a subnet successfully'''
217
   #     test_net = mf.NetworkFactory()
218
   #     request = {
219
   #         'subnet': {
220
   #             'network_id': test_net.id,
221
   #             'cidr': '10.0.3.0/24',
222
   #             'ip_version': 4}
223
   #     }
224
   #     response = self.post(SUBNETS_URL, test_net.userid,
225
   #                          json.dumps(request), "json")
226
   #     self.assertSuccess(response)
227

    
228
    def test_create_subnet_with_invalid_network_id(self):
229
        '''Test create a subnet with a network id that doesn't exist'''
230
        test_net = mf.NetworkFactory()
231
        request = {
232
            'subnet': {
233
                'network_id': '42',
234
                'cidr': '10.0.3.0/24',
235
                'ip_version': 4}
236
        }
237
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
238
                             "json")
239
        self.assertItemNotFound(response)
240

    
241
    def test_create_subnet_with_malformed_ipversion(self):
242
        '''Create a subnet with a malformed ip_version type'''
243
        test_net = mf.NetworkFactory()
244
        request = {
245
            'subnet': {
246
                'network_id': test_net.id,
247
                'cidr': '10.0.3.0/24',
248
                'ip_version': 8}
249
        }
250
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
251
                             "json")
252
        self.assertBadRequest(response)
253

    
254
    def test_create_subnet_with_invalid_cidr(self):
255
        '''Create a subnet with an invalid cidr'''
256
        test_net = mf.NetworkFactory()
257
        request = {
258
            'subnet': {
259
                'network_id': test_net.id,
260
                'cidr': '192.168.3.0/8'}
261
        }
262
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
263
                             "json")
264
        self.assertBadRequest(response)
265

    
266
    def test_create_subnet_with_invalid_gateway(self):
267
        '''Create a subnet with a gateway outside of the subnet range'''
268
        test_net = mf.NetworkFactory()
269
        request = {
270
            'subnet': {
271
                'network_id': test_net.id,
272
                'cidr': '192.168.3.0/24',
273
                'gateway_ip': '192.168.0.1'}
274
        }
275
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
276
                             "json")
277
        self.assertBadRequest(response)
278

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

    
292
    def test_create_subnet_with_invalid_dhcp(self):
293
        '''Create a subnet with an invalid dhcp value'''
294
        test_net = mf.NetworkFactory()
295
        request = {
296
            'subnet': {
297
                'network_id': test_net.id,
298
                'cidr': '192.168.3.0/24',
299
                'enable_dhcp': 'None'}
300
        }
301
        response = self.post(SUBNETS_URL, test_net.userid, json.dumps(request),
302
                             "json")
303
        self.assertBadRequest(response)