Revision 5b8a9240 snf-cyclades-app/synnefo/api/tests/networks.py

b/snf-cyclades-app/synnefo/api/tests/networks.py
1
from snf_django.utils.testing import BaseAPITest
1
# Copyright 2012-2013 GRNET S.A. All rights reserved.
2
#
3
# Redistribution and use in source and binary forms, with or
4
# without modification, are permitted provided that the following
5
# conditions are met:
6
#
7
#   1. Redistributions of source code must retain the above
8
#      copyright notice, this list of conditions and the following
9
#      disclaimer.
10
#
11
#   2. Redistributions in binary form must reproduce the above
12
#      copyright notice, this list of conditions and the following
13
#      disclaimer in the documentation and/or other materials
14
#      provided with the distribution.
15
#
16
# THIS SOFTWARE IS PROVIDED BY GRNET S.A. ``AS IS'' AND ANY EXPRESS
17
# OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
18
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
19
# PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL GRNET S.A OR
20
# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
21
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
22
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
23
# USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED
24
# AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
25
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
26
# ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
27
# POSSIBILITY OF SUCH DAMAGE.
28
#
29
# The views and conclusions contained in the software and
30
# documentation are those of the authors and should not be
31
# interpreted as representing official policies, either expressed
32
# or implied, of GRNET S.A.
33

  
34
from snf_django.utils.testing import (BaseAPITest, override_settings)
2 35
from django.utils import simplejson as json
3 36
from synnefo.cyclades_settings import cyclades_services
4 37
from synnefo.lib.services import get_service_path
5 38
from synnefo.lib import join_urls
6
import json
7 39
import synnefo.db.models_factory as dbmf
8

  
40
from synnefo.db.models import Network
41
from django.conf import settings
9 42

  
10 43
COMPUTE_URL = get_service_path(cyclades_services, 'compute',
11
                                             version='v2.0')
44
                               version='v2.0')
12 45
NETWORKS_URL = join_urls(COMPUTE_URL, "networks/")
13 46

  
14 47

  
15 48
class NetworkTest(BaseAPITest):
16

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

  
23
    def test_create_network(self):
55
    def test_invalid_create(self):
56
        """Test invalid flavor"""
57
        request = {'network': {}}
58
        response = self.post(NETWORKS_URL, "user1", params=json.dumps(request))
59
        self.assertBadRequest(response)
60
        request = {'network': {"type": "foo"}}
61
        response = self.post(NETWORKS_URL, "user1", params=json.dumps(request))
62
        self.assertBadRequest(response)
63
        request = {'network': {"type": "MAC_FILTERED"}}
64
        with override_settings(settings,
65
                               API_ENABLED_NETWORK_FLAVORS=["CUSTOM"]):
66
            response = self.post(NETWORKS_URL, "user1",
67
                                 params=json.dumps(request))
68
        self.assertEqual(response.status_code, 403)
69
        self.assertEqual(len(Network.objects.filter(userid='user1')), 0)
70

  
71
    def test_create(self):
24 72
        request = {
25 73
            "network": {
26 74
                "name": "sample_network",
27 75
                "type": "MAC_FILTERED"
28
                }
76
            }
29 77
        }
30 78
        response = self.post(NETWORKS_URL, params=json.dumps(request))
31 79
        code = response.status_code
80
        self.assertEqual(code, 503)  # No MAC prefix pool
81
        dbmf.MacPrefixPoolTableFactory()
82
        response = self.post(NETWORKS_URL, params=json.dumps(request))
83
        code = response.status_code
84
        self.assertEqual(code, 201)
85
        res = json.loads(response.content)
86
        self.assertEqual(res["network"]["name"], "sample_network")
87

  
88
        # TEST QUOTAS!!!
89
        name, args, kwargs =\
90
            self.mocked_quotaholder.issue_one_commission.mock_calls[0]
91
        commission_resources = args[3]
92
        self.assertEqual(commission_resources, {"cyclades.network.private": 1})
93
        name, args, kwargs =\
94
            self.mocked_quotaholder.resolve_commissions.mock_calls[0]
95
        serial = Network.objects.get().serial.serial
96
        accepted_serials = args[1]
97
        rejected_serials = args[2]
98
        self.assertEqual(accepted_serials, [serial])
99
        self.assertEqual(rejected_serials, [])
100

  
101
        # test no name
102
        request["network"].pop("name")
103
        response = self.post(NETWORKS_URL, params=json.dumps(request))
104
        code = response.status_code
32 105
        self.assertEqual(code, 201)
106
        res = json.loads(response.content)
107
        self.assertEqual(res["network"]["name"], "")
33 108

  
34 109
    def test_get_unfound_network(self):
35 110
        url = join_urls(NETWORKS_URL, "123")
......
37 112
        self.assertItemNotFound(response)
38 113

  
39 114
    def test_get_network(self):
40
        test_net = dbmf.NetworkFactory.create()
115
        test_net = dbmf.NetworkFactory()
41 116
        url = join_urls(NETWORKS_URL, str(test_net.id))
42 117
        response = self.get(url, user=test_net.userid)
43 118
        # validate response
......
50 125
        self.assertEqual(response.status_code, 200)
51 126

  
52 127
    def test_get_deleted_network(self):
53
        test_net = dbmf.NetworkFactory.create()
128
        test_net = dbmf.NetworkFactory(flavor="CUSTOM")
54 129
        url = join_urls(NETWORKS_URL, str(test_net.id))
55 130
        self.delete(url, user=test_net.userid)
56 131
        response = self.get(url, user=test_net.userid)
57
        self.assertEqual(response.status_code, 400)
132
        self.assertEqual(response.status_code, 200)
58 133

  
59 134
    def test_delete_unfound_network(self):
60 135
        url = join_urls(NETWORKS_URL, "123")
......
62 137
        self.assertItemNotFound(response)
63 138

  
64 139
    def test_delete_network(self):
65
        test_net = dbmf.NetworkFactory.create()
66
        subnet = dbmf.IPv4SubnetFactory.create(network=test_net)
140
        test_net = dbmf.NetworkFactory()
141
        dbmf.IPv4SubnetFactory(network=test_net)
67 142
        url = join_urls(NETWORKS_URL, str(test_net.id))
68 143
        response = self.delete(url, user=test_net.userid)
69 144
        self.assertEqual(response.status_code, 204)
145
        # But not the public network!!
146
        test_net.public = True
147
        test_net.save()
148
        response = self.delete(url, user=test_net.userid)
149
        self.assertFault(response, 403, 'forbidden')
70 150

  
71 151
    def test_delete_network_in_use(self):
72
        test_net = dbmf.NetworkFactory.create()
73
        test_iface = dbmf.NetworkInterfaceFactory.create(network=test_net)
152
        test_net = dbmf.NetworkFactory()
153
        dbmf.NetworkInterfaceFactory(network=test_net)
154
        url = join_urls(NETWORKS_URL, str(test_net.id))
155
        response = self.delete(url, user=test_net.userid)
156
        self.assertEqual(response.status_code, 409)
157

  
158
    def test_delete_network_with_floating_ips(self):
159
        test_net = dbmf.NetworkFactory()
160
        dbmf.IPv4AddressFactory(network=test_net, floating_ip=True, nic=None)
74 161
        url = join_urls(NETWORKS_URL, str(test_net.id))
75 162
        response = self.delete(url, user=test_net.userid)
76 163
        self.assertEqual(response.status_code, 409)
......
81 168
        self.assertItemNotFound(response)
82 169

  
83 170
    def test_put_network(self):
84
        test_net = dbmf.NetworkFactory.create()
171
        test_net = dbmf.NetworkFactory()
85 172
        url = join_urls(NETWORKS_URL, str(test_net.id))
86 173
        request = {
87 174
            "network": {
......
92 179
        self.assertEqual(response.status_code, 200)
93 180

  
94 181
    def test_put_network_wrong_data(self):
95
        test_net = dbmf.NetworkFactory.create()
182
        test_net = dbmf.NetworkFactory()
96 183
        url = join_urls(NETWORKS_URL, str(test_net.id))
97 184
        request = {
98 185
            "network": {
......
103 190
        self.assertEqual(response.status_code, 400)
104 191

  
105 192
    def test_put_no_data(self):
106
        test_net = dbmf.NetworkFactory.create()
193
        test_net = dbmf.NetworkFactory()
107 194
        url = join_urls(NETWORKS_URL, str(test_net.id))
108 195
        response = self.put(url, params="", user=test_net.userid)
109 196
        self.assertEqual(response.status_code, 400)
197

  
198
    def test_rename_network(self):
199
        test_net = dbmf.NetworkFactory(name="foo")
200
        url = join_urls(NETWORKS_URL, str(test_net.id))
201
        request = {'network': {'name': "new_name"}}
202
        response = self.put(url, test_net.userid, json.dumps(request), 'json')
203
        self.assertEqual(response.status_code, 200)
204
        self.assertEqual(Network.objects.get(id=test_net.id).name, "new_name")
205
        # test if server deleted
206
        test_net.deleted = True
207
        test_net.save()
208
        response = self.put(url, test_net.userid, json.dumps(request), 'json')
209
        self.assertBadRequest(response)
210
        test_net.deleted = False
211
        test_net.public = True
212
        test_net.save()
213
        response = self.put(url, test_net.userid, json.dumps(request), 'json')
214
        self.assertFault(response, 403, 'forbidden')
215

  
216
    def test_method_not_allowed(self, *args):
217
        # /networks/ allows only POST, GET
218
        response = self.put(NETWORKS_URL, '', '')
219
        self.assertMethodNotAllowed(response)
220
        response = self.delete(NETWORKS_URL)
221
        self.assertMethodNotAllowed(response)
222

  
223
        # /networks/<srvid>/ allows only GET, PUT, DELETE
224
        url = join_urls(NETWORKS_URL, "42")
225
        response = self.post(url)
226
        self.assertMethodNotAllowed(response)
227

  
228

  
229
#class NetworkNICsAPITest(BaseAPITest):
230
#    def test_get_network_building_nics(self, mrapi):
231
#        net = dbmf.NetworkFactory()
232
#        machine = dbmf.VirtualMachineFactory(userid=net.userid)
233
#        dbmf.NetworkInterfaceFactory(network=net, machine=machine,
234
#                                     state="BUILDING")
235
#        response = self.myget('networks/%d' % net.id, net.userid)
236
#        self.assertSuccess(response)
237
#        api_net = json.loads(response.content)["network"]
238
#        self.assertEqual(len(api_net["attachments"]), 0)
239
#
240
#
241
#    def test_add_nic(self, mrapi):
242
#        user = 'userr'
243
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
244
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
245
#        mrapi().ModifyInstance.return_value = 1
246
#        request = {'add': {'serverRef': vm.id}}
247
#        response = self.mypost('networks/%d/action' % net.id,
248
#                               net.userid, json.dumps(request), 'json')
249
#        self.assertEqual(response.status_code, 202)
250
#
251
#    def test_add_nic_to_deleted_network(self, mrapi):
252
#        user = 'userr'
253
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user,
254
#                                            operstate="ACTIVE")
255
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user,
256
#                                      deleted=True)
257
#        request = {'add': {'serverRef': vm.id}}
258
#        response = self.mypost('networks/%d/action' % net.id,
259
#                               net.userid, json.dumps(request), 'json')
260
#        self.assertBadRequest(response)
261
#        self.assertFalse(mrapi.called)
262
#
263
#    def test_add_nic_to_public_network(self, mrapi):
264
#        user = 'userr'
265
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
266
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user, public=True)
267
#        request = {'add': {'serverRef': vm.id}}
268
#        response = self.mypost('networks/%d/action' % net.id,
269
#                               net.userid, json.dumps(request), 'json')
270
#        self.assertFault(response, 403, 'forbidden')
271
#        self.assertFalse(mrapi.called)
272
#
273
#    def test_add_nic_malformed_1(self, mrapi):
274
#        user = 'userr'
275
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
276
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
277
#        request = {'add': {'serveRef': vm.id}}
278
#        response = self.mypost('networks/%d/action' % net.id,
279
#                               net.userid, json.dumps(request), 'json')
280
#        self.assertBadRequest(response)
281
#        self.assertFalse(mrapi.called)
282
#
283
#    def test_add_nic_malformed_2(self, mrapi):
284
#        user = 'userr'
285
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
286
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
287
#        request = {'add': {'serveRef': [vm.id, 22]}}
288
#        response = self.mypost('networks/%d/action' % net.id,
289
#                               net.userid, json.dumps(request), 'json')
290
#        self.assertBadRequest(response)
291
#        self.assertFalse(mrapi.called)
292
#
293
#    def test_add_nic_not_active(self, mrapi):
294
#        """Test connecting VM to non-active network"""
295
#        user = 'dummy'
296
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
297
#        net = dbmf.NetworkFactory(state='PENDING', subnet='10.0.0.0/31',
298
#                                      userid=user)
299
#        request = {'add': {'serverRef': vm.id}}
300
#        response = self.mypost('networks/%d/action' % net.id,
301
#                               net.userid, json.dumps(request), 'json')
302
#        # Test that returns BuildInProgress
303
#        self.assertEqual(response.status_code, 409)
304
#        self.assertFalse(mrapi.called)
305
#
306
#    def test_add_nic_full_network(self, mrapi):
307
#        """Test connecting VM to a full network"""
308
#        user = 'userr'
309
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user,
310
#                                            operstate="STARTED")
311
#        net = dbmf.NetworkFactory(state='ACTIVE', subnet='10.0.0.0/30',
312
#                                      userid=user, dhcp=True)
313
#        pool = net.get_pool()
314
#        while not pool.empty():
315
#            pool.get()
316
#        pool.save()
317
#        pool = net.get_pool()
318
#        self.assertTrue(pool.empty())
319
#        request = {'add': {'serverRef': vm.id}}
320
#        response = self.mypost('networks/%d/action' % net.id,
321
#                               net.userid, json.dumps(request), 'json')
322
#        # Test that returns OverLimit
323
#        self.assertEqual(response.status_code, 413)
324
#        self.assertFalse(mrapi.called)
325
#
326
#    def test_remove_nic(self, mrapi):
327
#        user = 'userr'
328
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user,
329
#                                            operstate="ACTIVE")
330
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
331
#        nic = dbmf.NetworkInterfaceFactory(machine=vm, network=net)
332
#        mrapi().ModifyInstance.return_value = 1
333
#        request = {'remove': {'attachment': "%s" % nic.id}}
334
#        response = self.mypost('networks/%d/action' % net.id,
335
#                               net.userid, json.dumps(request), 'json')
336
#        self.assertEqual(response.status_code, 202)
337
#        vm.task = None
338
#        vm.task_job_id = None
339
#        vm.save()
340
#
341
#    def test_remove_nic_malformed(self, mrapi):
342
#        user = 'userr'
343
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
344
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
345
#        nic = dbmf.NetworkInterfaceFactory(machine=vm, network=net)
346
#        request = {'remove': {'att234achment': '%s' % nic.id}}
347
#        response = self.mypost('networks/%d/action' % net.id,
348
#                               net.userid, json.dumps(request), 'json')
349
#        self.assertBadRequest(response)
350
#
351
#    def test_remove_nic_malformed_2(self, mrapi):
352
#        user = 'userr'
353
#        vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
354
#        net = dbmf.NetworkFactory(state='ACTIVE', userid=user)
355
#        request = {'remove': {'attachment': 'nic-%s' % vm.id}}
356
#        response = self.mypost('networks/%d/action' % net.id,
357
#                               net.userid, json.dumps(request), 'json')
358
#        self.assertBadRequest(response)

Also available in: Unified diff