Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / api / tests / floating_ips.py @ ddffd973

History | View | Annotate | Download (19.6 kB)

1
# Copyright 2012 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 django.utils import simplejson as json
35
from snf_django.utils.testing import BaseAPITest, mocked_quotaholder
36
from synnefo.db.models import IPAddress, Network, Subnet, IPPoolTable
37
from synnefo.db import models_factory as mf
38

    
39
from mock import patch, Mock
40

    
41
from synnefo.cyclades_settings import cyclades_services
42
from synnefo.lib.services import get_service_path
43
from synnefo.lib import join_urls
44

    
45

    
46
network_path = get_service_path(cyclades_services, "network", version="v2.0")
47
URL = join_urls(network_path, "floatingips")
48
NETWORKS_URL = join_urls(network_path, "networks")
49
SERVERS_URL = join_urls(network_path, "servers")
50

    
51

    
52
floating_ips = IPAddress.objects.filter(floating_ip=True)
53

    
54

    
55
class FloatingIPAPITest(BaseAPITest):
56
    def setUp(self):
57
        self.pool = mf.NetworkWithSubnetFactory(floating_ip_pool=True,
58
                                                public=True,
59
                                                subnet__cidr="192.168.2.0/24",
60
                                                subnet__gateway="192.168.2.1")
61

    
62
    def test_no_floating_ip(self):
63
        response = self.get(URL)
64
        self.assertSuccess(response)
65
        self.assertEqual(json.loads(response.content)["floatingips"], [])
66

    
67
    def test_list_ips(self):
68
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
69
        with mocked_quotaholder():
70
            response = self.get(URL, "user1")
71
        self.assertSuccess(response)
72
        api_ip = json.loads(response.content)["floatingips"][0]
73
        self.assertEqual(api_ip,
74
                         {"instance_id": str(ip.nic.machine_id),
75
                          "floating_ip_address": ip.address,
76
                          "fixed_ip_address": None,
77
                          "id": str(ip.id),
78
                          "port_id": str(ip.nic.id),
79
                          "floating_network_id": str(ip.network_id)})
80

    
81
    def test_get_ip(self):
82
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
83
        with mocked_quotaholder():
84
            response = self.get(URL + "/%s" % ip.id, "user1")
85
        self.assertSuccess(response)
86
        api_ip = json.loads(response.content)["floatingip"]
87
        self.assertEqual(api_ip,
88
                         {"instance_id": str(ip.nic.machine_id),
89
                          "floating_ip_address": ip.address,
90
                          "fixed_ip_address": None,
91
                          "id": str(ip.id),
92
                          "port_id": str(ip.nic.id),
93
                          "floating_network_id": str(ip.network_id)})
94

    
95
    def test_wrong_user(self):
96
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
97
        response = self.delete(URL + "/%s" % ip.id, "user2")
98
        self.assertItemNotFound(response)
99

    
100
    def test_deleted_ip(self):
101
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True,
102
                                   deleted=True)
103
        response = self.delete(URL + "/%s" % ip.id, "user1")
104
        self.assertItemNotFound(response)
105

    
106
    def test_reserve(self):
107
        request = {"floatingip": {
108
            "floating_network_id": self.pool.id}
109
            }
110
        with mocked_quotaholder():
111
            response = self.post(URL, "test_user", json.dumps(request), "json")
112
        self.assertSuccess(response)
113
        api_ip = json.loads(response.content, encoding="utf-8")["floatingip"]
114
        ip = floating_ips.get()
115
        self.assertEqual(ip.address, "192.168.2.2")
116
        self.assertEqual(ip.nic, None)
117
        self.assertEqual(ip.network, self.pool)
118
        self.assertEqual(api_ip,
119
                         {"instance_id": None,
120
                          "floating_ip_address": "192.168.2.2",
121
                          "fixed_ip_address": None,
122
                          "id": str(ip.id),
123
                          "port_id": None,
124
                          "floating_network_id": str(self.pool.id)})
125

    
126
    def test_reserve_empty_body(self):
127
        """Test reserve FIP without specifying network."""
128
        request = {"floatingip": {}}
129
        # delete all pools..
130
        IPPoolTable.objects.all().delete()
131
        Subnet.objects.all().delete()
132
        Network.objects.all().delete()
133
        # CASE: no floating IP pool
134
        with mocked_quotaholder():
135
            response = self.post(URL, "test_user", json.dumps(request), "json")
136
        self.assertConflict(response)
137
        # CASE: Full floating IP pool
138
        mf.NetworkWithSubnetFactory(floating_ip_pool=True, public=True,
139
                                    subnet__pool__size=0)
140
        with mocked_quotaholder():
141
            response = self.post(URL, "test_user", json.dumps(request), "json")
142
        self.assertConflict(response)
143
        # CASE: Available floating IP pool
144
        p1 = mf.NetworkWithSubnetFactory(floating_ip_pool=True, public=True,
145
                                         subnet__cidr="192.168.2.0/30",
146
                                         subnet__pool__size=1)
147
        with mocked_quotaholder():
148
            response = self.post(URL, "test_user", json.dumps(request), "json")
149
        self.assertSuccess(response)
150
        floating_ip = json.loads(response.content)["floatingip"]
151
        db_fip = IPAddress.objects.get(id=floating_ip["id"])
152
        self.assertEqual(db_fip.address, floating_ip["floating_ip_address"])
153
        self.assertTrue(db_fip.floating_ip)
154
        # Test that address is reserved
155
        ip_pool = p1.get_ip_pools()[0]
156
        self.assertFalse(ip_pool.is_available(db_fip.address))
157

    
158
    def test_reserve_no_pool(self):
159
        # Network is not a floating IP pool
160
        pool2 = mf.NetworkWithSubnetFactory(floating_ip_pool=False,
161
                                            public=True,
162
                                            subnet__cidr="192.168.2.0/24",
163
                                            subnet__gateway="192.168.2.1")
164
        request = {"floatingip": {
165
            'floating_network_id': pool2.id}
166
            }
167
        response = self.post(URL, "test_user", json.dumps(request), "json")
168
        self.assertConflict(response)
169

    
170
        # Full network
171
        net = mf.NetworkWithSubnetFactory(floating_ip_pool=True,
172
                                          public=True,
173
                                          subnet__cidr="192.168.2.0/31",
174
                                          subnet__gateway="192.168.2.1",
175
                                          subnet__pool__size=0)
176
        request = {"floatingip": {
177
            'floating_network_id': net.id}
178
            }
179
        response = self.post(URL, "test_user", json.dumps(request), "json")
180
        self.assertConflict(response)
181

    
182
    def test_reserve_with_address(self):
183
        request = {"floatingip": {
184
            "floating_network_id": self.pool.id,
185
            "floating_ip_address": "192.168.2.10"}
186
            }
187
        with mocked_quotaholder():
188
            response = self.post(URL, "test_user", json.dumps(request), "json")
189
        self.assertSuccess(response)
190
        ip = floating_ips.get()
191
        self.assertEqual(json.loads(response.content)["floatingip"],
192
                         {"instance_id": None,
193
                          "floating_ip_address": "192.168.2.10",
194
                          "fixed_ip_address": None,
195
                          "id": str(ip.id),
196
                          "port_id": None,
197
                          "floating_network_id": str(self.pool.id)})
198

    
199
        # Already reserved
200
        with mocked_quotaholder():
201
            response = self.post(URL, "test_user", json.dumps(request), "json")
202
        self.assertFault(response, 409, "conflict")
203

    
204
        # Used by instance
205
        self.pool.reserve_address("192.168.2.20")
206
        request = {"floatingip": {
207
            "floating_network_id": self.pool.id,
208
            "floating_ip_address": "192.168.2.20"}
209
            }
210
        with mocked_quotaholder():
211
            response = self.post(URL, "test_user", json.dumps(request), "json")
212
        self.assertFault(response, 409, "conflict")
213

    
214
        # Address out of pool
215
        request = {"floatingip": {
216
            "floating_network_id": self.pool.id,
217
            "floating_ip_address": "192.168.3.5"}
218
            }
219
        with mocked_quotaholder():
220
            response = self.post(URL, "test_user", json.dumps(request), "json")
221
        self.assertBadRequest(response)
222

    
223
    '''
224
    @patch("synnefo.db.models.get_rapi_client")
225
    def test_reserve_and_connect(self, mrapi):
226
        vm = mf.VirtualMachineFactory(userid="test_user")
227
        request = {"floatingip": {
228
            "floating_network_id": self.pool.id,
229
            "floating_ip_address": "192.168.2.12",
230
            "device_id": vm.id}
231
            }
232
        response = self.post(URL, "test_user", json.dumps(request), "json")
233
        ip = floating_ips.get()
234
        api_ip = json.loads(response.content, "utf-8")["floatingip"]
235
        self.assertEqual(api_ip,
236
                         {"instance_id": str(vm.id),
237
                          "floating_ip_address": "192.168.2.12",
238
                          "fixed_ip_address": None,
239
                          "id": str(ip.id),
240
                          "port_id": str(vm.nics.all()[0].id),
241
                          "floating_network_id": str(self.pool.id)})
242

243
    @patch("synnefo.db.models.get_rapi_client")
244
    def test_update_attach(self, mrapi):
245
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
246
        vm = mf.VirtualMachineFactory(userid="user1")
247
        request = {"floatingip": {
248
            "device_id": vm.id}
249
            }
250
        with mocked_quotaholder():
251
            response = self.put(URL + "/%s" % ip.id, "user1",
252
                                json.dumps(request), "json")
253
        self.assertEqual(response.status_code, 202)
254

255
    def test_update_attach_conflict(self):
256
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
257
        vm = mf.VirtualMachineFactory(userid="user1")
258
        request = {"floatingip": {
259
            "device_id": vm.id}
260
            }
261
        with mocked_quotaholder():
262
            response = self.put(URL + "/%s" % ip.id, "user1",
263
                                json.dumps(request), "json")
264
        self.assertEqual(response.status_code, 409)
265

266
    @patch("synnefo.db.models.get_rapi_client")
267
    def test_update_dettach(self, mrapi):
268
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
269
        request = {"floatingip": {
270
            "device_id": None}
271
            }
272
        mrapi().ModifyInstance.return_value = 42
273
        with mocked_quotaholder():
274
            response = self.put(URL + "/%s" % ip.id, "user1",
275
                                json.dumps(request), "json")
276
        self.assertEqual(response.status_code, 202)
277

278
    def test_update_dettach_unassociated(self):
279
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
280
        request = {"floatingip": {}}
281
        with mocked_quotaholder():
282
            response = self.put(URL + "/%s" % ip.id, "user1",
283
                                json.dumps(request), "json")
284
        self.assertEqual(response.status_code, 400)
285

286
    def test_release_in_use(self):
287
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
288
        vm = mf.VirtualMachineFactory(userid="user1")
289
        request = {"floatingip": {
290
            "device_id": vm.id}
291
            }
292
        with mocked_quotaholder():
293
            response = self.put(URL + "/%s" % ip.id, "user1",
294
                                json.dumps(request), "json")
295
        self.assertEqual(response.status_code, 409)
296

297
    @patch("synnefo.db.models.get_rapi_client")
298
    def test_update_dettach(self, mrapi):
299
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
300
        request = {"floatingip": {
301
            "device_id": None}
302
            }
303
        mrapi().ModifyInstance.return_value = 42
304
        with mocked_quotaholder():
305
            response = self.put(URL + "/%s" % ip.id, "user1",
306
                                json.dumps(request), "json")
307
        self.assertEqual(response.status_code, 202)
308

309
    def test_update_dettach_unassociated(self):
310
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
311
        request = {"floatingip": {}}
312
        with mocked_quotaholder():
313
            response = self.put(URL + "/%s" % ip.id, "user1",
314
                                json.dumps(request), "json")
315
        self.assertEqual(response.status_code, 400)
316

317
    def test_release_in_use(self):
318
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
319
        vm = ip.nic.machine
320
        with mocked_quotaholder():
321
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
322
        self.assertFault(response, 409, "conflict")
323
    '''
324

    
325
    def test_update(self):
326
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
327
        with mocked_quotaholder():
328
            response = self.put(URL + "/%s" % ip.id, ip.userid)
329
        self.assertEqual(response.status_code, 501)
330

    
331
    def test_release(self):
332
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
333
        with mocked_quotaholder():
334
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
335
        self.assertSuccess(response)
336
        ips_after = floating_ips.filter(id=ip.id)
337
        self.assertEqual(len(ips_after), 0)
338

    
339
    @patch("synnefo.logic.backend", Mock())
340
    def test_delete_network_with_floating_ips(self):
341
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True,
342
                                   network=self.pool, nic=None)
343
        # Mark the network as non-pubic to not get 403
344
        network = ip.network
345
        network.public = False
346
        network.save()
347
        # Cannot remove network with floating IPs
348
        with mocked_quotaholder():
349
            response = self.delete(NETWORKS_URL + "/%s" % self.pool.id,
350
                                   self.pool.userid)
351
        self.assertConflict(response)
352
        # But we can with only deleted Floating Ips
353
        ip.deleted = True
354
        ip.save()
355
        with mocked_quotaholder():
356
            response = self.delete(NETWORKS_URL + "/%s" % self.pool.id,
357
                                   self.pool.userid)
358
        self.assertSuccess(response)
359

    
360
'''
361
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools")
362

363

364
class FloatingIPPoolsAPITest(BaseAPITest):
365
    def test_no_pool(self):
366
        response = self.get(POOLS_URL)
367
        self.assertSuccess(response)
368
        self.assertEqual(json.loads(response.content)["floating_ip_pools"], [])
369

370
    def test_list_pools(self):
371
        net = mf.NetworkWithSubnetFactory(floating_ip_pool=True,
372
                                          public=True,
373
                                          subnet__cidr="192.168.2.0/30",
374
                                          subnet__gateway="192.168.2.1",
375
                                          subnet__pool__size=1,
376
                                          subnet__pool__offset=1)
377
        mf.NetworkWithSubnetFactory(public=True, deleted=True)
378
        mf.NetworkWithSubnetFactory(public=False, deleted=False)
379
        mf.NetworkWithSubnetFactory(public=True, floating_ip_pool=False)
380
        response = self.get(POOLS_URL)
381
        self.assertSuccess(response)
382
        self.assertEqual(json.loads(response.content)["floating_ip_pools"],
383
                         [{"name": str(net.id), "size": 1, "free": 1}])
384

385

386
class FloatingIPActionsTest(BaseAPITest):
387
    def setUp(self):
388
        self.vm = VirtualMachineFactory()
389
        self.vm.operstate = "ACTIVE"
390
        self.vm.save()
391

392
    def test_bad_request(self):
393
        url = SERVERS_URL + "/%s/action" % self.vm.id
394
        response = self.post(url, self.vm.userid, json.dumps({}), "json")
395
        self.assertBadRequest(response)
396
        response = self.post(url, self.vm.userid,
397
                             json.dumps({"addFloatingIp": {}}),
398
                             "json")
399
        self.assertBadRequest(response)
400

401
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
402
    def test_add_floating_ip(self, mock):
403
        # Not exists
404
        url = SERVERS_URL + "/%s/action" % self.vm.id
405
        request = {"addFloatingIp": {"address": "10.0.0.1"}}
406
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
407
        self.assertItemNotFound(response)
408
        # In use
409
        ip = mf.IPv4AddressFactory(floating_ip=True, userid=self.vm.userid)
410
        request = {"addFloatingIp": {"address": ip.address}}
411
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
412
        self.assertConflict(response)
413
        # Success
414
        ip = mf.IPv4AddressFactory(floating_ip=True, nic=None,
415
                                   userid=self.vm.userid)
416
        request = {"addFloatingIp": {"address": ip.address}}
417
        mock().ModifyInstance.return_value = 1
418
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
419
        self.assertEqual(response.status_code, 202)
420
        ip_after = floating_ips.get(id=ip.id)
421
        self.assertEqual(ip_after.nic.machine, self.vm)
422
        nic = self.vm.nics.get()
423
        nic.state = "ACTIVE"
424
        nic.save()
425
        response = self.get(SERVERS_URL + "/%s" % self.vm.id,
426
                            self.vm.userid)
427
        self.assertSuccess(response)
428
        nic = json.loads(response.content)["server"]["attachments"][0]
429
        self.assertEqual(nic["OS-EXT-IPS:type"], "floating")
430

431
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
432
    def test_remove_floating_ip(self, mock):
433
        # Not exists
434
        url = SERVERS_URL + "/%s/action" % self.vm.id
435
        request = {"removeFloatingIp": {"address": "10.0.0.1"}}
436
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
437
        self.assertBadRequest(response)
438
        # Not In Use
439
        ip = mf.IPv4AddressFactory(floating_ip=True, nic=None,
440
                                   userid=self.vm.userid)
441
        request = {"removeFloatingIp": {"address": ip.address}}
442
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
443
        self.assertBadRequest(response)
444
        # Success
445
        ip = mf.IPv4AddressFactory(floating_ip=True,
446
                                   userid=self.vm.userid, nic__machine=self.vm)
447
        request = {"removeFloatingIp": {"address": ip.address}}
448
        mock().ModifyInstance.return_value = 2
449
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
450
        self.assertEqual(response.status_code, 202)
451
        # Yet used. Wait for the callbacks
452
        ip_after = floating_ips.get(id=ip.id)
453
        self.assertEqual(ip_after.nic.machine, self.vm)
454
'''