Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (20.3 kB)

1
# Copyright 2012-2014 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", project="user1",
69
                                   floating_ip=True)
70
        with mocked_quotaholder():
71
            response = self.get(URL, "user1")
72
        self.assertSuccess(response)
73
        api_ip = json.loads(response.content)["floatingips"][0]
74
        self.assertEqual(api_ip,
75
                         {"instance_id": str(ip.nic.machine_id),
76
                          "floating_ip_address": ip.address,
77
                          "fixed_ip_address": None,
78
                          "id": str(ip.id),
79
                          "port_id": str(ip.nic.id),
80
                          "deleted": False,
81
                          "user_id": "user1",
82
                          "tenant_id": "user1",
83
                          "floating_network_id": str(ip.network_id)})
84

    
85
    def test_get_ip(self):
86
        ip = mf.IPv4AddressFactory(userid="user1", project="user1",
87
                                   floating_ip=True)
88
        with mocked_quotaholder():
89
            response = self.get(URL + "/%s" % ip.id, "user1")
90
        self.assertSuccess(response)
91
        api_ip = json.loads(response.content)["floatingip"]
92
        self.assertEqual(api_ip,
93
                         {"instance_id": str(ip.nic.machine_id),
94
                          "floating_ip_address": ip.address,
95
                          "fixed_ip_address": None,
96
                          "id": str(ip.id),
97
                          "port_id": str(ip.nic.id),
98
                          "deleted": False,
99
                          "user_id": "user1",
100
                          "tenant_id": "user1",
101
                          "floating_network_id": str(ip.network_id)})
102

    
103
    def test_wrong_user(self):
104
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
105
        response = self.delete(URL + "/%s" % ip.id, "user2")
106
        self.assertItemNotFound(response)
107

    
108
    def test_deleted_ip(self):
109
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True,
110
                                   deleted=True)
111
        response = self.delete(URL + "/%s" % ip.id, "user1")
112
        self.assertItemNotFound(response)
113

    
114
    def test_reserve(self):
115
        request = {"floatingip": {
116
            "floating_network_id": self.pool.id}
117
            }
118
        with mocked_quotaholder():
119
            response = self.post(URL, "test_user", json.dumps(request), "json")
120
        self.assertSuccess(response)
121
        api_ip = json.loads(response.content, encoding="utf-8")["floatingip"]
122
        ip = floating_ips.get()
123
        self.assertEqual(ip.address, "192.168.2.2")
124
        self.assertEqual(ip.nic, None)
125
        self.assertEqual(ip.network, self.pool)
126
        self.assertEqual(api_ip,
127
                         {"instance_id": None,
128
                          "floating_ip_address": "192.168.2.2",
129
                          "fixed_ip_address": None,
130
                          "id": str(ip.id),
131
                          "port_id": None,
132
                          "deleted": False,
133
                          "user_id": "test_user",
134
                          "tenant_id": "test_user",
135
                          "floating_network_id": str(self.pool.id)})
136

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

    
169
    def test_reserve_no_pool(self):
170
        # Network is not a floating IP pool
171
        pool2 = mf.NetworkWithSubnetFactory(floating_ip_pool=False,
172
                                            public=True,
173
                                            subnet__cidr="192.168.2.0/24",
174
                                            subnet__gateway="192.168.2.1")
175
        request = {"floatingip": {
176
            'floating_network_id': pool2.id}
177
            }
178
        response = self.post(URL, "test_user", json.dumps(request), "json")
179
        self.assertConflict(response)
180

    
181
        # Full network
182
        net = mf.NetworkWithSubnetFactory(floating_ip_pool=True,
183
                                          public=True,
184
                                          subnet__cidr="192.168.2.0/31",
185
                                          subnet__gateway="192.168.2.1",
186
                                          subnet__pool__size=0)
187
        request = {"floatingip": {
188
            'floating_network_id': net.id}
189
            }
190
        response = self.post(URL, "test_user", json.dumps(request), "json")
191
        self.assertConflict(response)
192

    
193
    def test_reserve_with_address(self):
194
        request = {"floatingip": {
195
            "floating_network_id": self.pool.id,
196
            "floating_ip_address": "192.168.2.10"}
197
            }
198
        with mocked_quotaholder():
199
            response = self.post(URL, "test_user", json.dumps(request), "json")
200
        self.assertSuccess(response)
201
        ip = floating_ips.get()
202
        self.assertEqual(json.loads(response.content)["floatingip"],
203
                         {"instance_id": None,
204
                          "floating_ip_address": "192.168.2.10",
205
                          "fixed_ip_address": None,
206
                          "id": str(ip.id),
207
                          "port_id": None,
208
                          "deleted": False,
209
                          "user_id": "test_user",
210
                          "tenant_id": "test_user",
211
                          "floating_network_id": str(self.pool.id)})
212

    
213
        # Already reserved
214
        with mocked_quotaholder():
215
            response = self.post(URL, "test_user", json.dumps(request), "json")
216
        self.assertFault(response, 409, "conflict")
217

    
218
        # Used by instance
219
        self.pool.reserve_address("192.168.2.20")
220
        request = {"floatingip": {
221
            "floating_network_id": self.pool.id,
222
            "floating_ip_address": "192.168.2.20"}
223
            }
224
        with mocked_quotaholder():
225
            response = self.post(URL, "test_user", json.dumps(request), "json")
226
        self.assertFault(response, 409, "conflict")
227

    
228
        # Address out of pool
229
        request = {"floatingip": {
230
            "floating_network_id": self.pool.id,
231
            "floating_ip_address": "192.168.3.5"}
232
            }
233
        with mocked_quotaholder():
234
            response = self.post(URL, "test_user", json.dumps(request), "json")
235
        self.assertBadRequest(response)
236

    
237
    '''
238
    @patch("synnefo.db.models.get_rapi_client")
239
    def test_reserve_and_connect(self, mrapi):
240
        vm = mf.VirtualMachineFactory(userid="test_user")
241
        request = {"floatingip": {
242
            "floating_network_id": self.pool.id,
243
            "floating_ip_address": "192.168.2.12",
244
            "device_id": vm.id}
245
            }
246
        response = self.post(URL, "test_user", json.dumps(request), "json")
247
        ip = floating_ips.get()
248
        api_ip = json.loads(response.content, "utf-8")["floatingip"]
249
        self.assertEqual(api_ip,
250
                         {"instance_id": str(vm.id),
251
                          "floating_ip_address": "192.168.2.12",
252
                          "fixed_ip_address": None,
253
                          "id": str(ip.id),
254
                          "port_id": str(vm.nics.all()[0].id),
255
                          "floating_network_id": str(self.pool.id)})
256

257
    @patch("synnefo.db.models.get_rapi_client")
258
    def test_update_attach(self, mrapi):
259
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
260
        vm = mf.VirtualMachineFactory(userid="user1")
261
        request = {"floatingip": {
262
            "device_id": vm.id}
263
            }
264
        with mocked_quotaholder():
265
            response = self.put(URL + "/%s" % ip.id, "user1",
266
                                json.dumps(request), "json")
267
        self.assertEqual(response.status_code, 202)
268

269
    def test_update_attach_conflict(self):
270
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
271
        vm = mf.VirtualMachineFactory(userid="user1")
272
        request = {"floatingip": {
273
            "device_id": vm.id}
274
            }
275
        with mocked_quotaholder():
276
            response = self.put(URL + "/%s" % ip.id, "user1",
277
                                json.dumps(request), "json")
278
        self.assertEqual(response.status_code, 409)
279

280
    @patch("synnefo.db.models.get_rapi_client")
281
    def test_update_dettach(self, mrapi):
282
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
283
        request = {"floatingip": {
284
            "device_id": None}
285
            }
286
        mrapi().ModifyInstance.return_value = 42
287
        with mocked_quotaholder():
288
            response = self.put(URL + "/%s" % ip.id, "user1",
289
                                json.dumps(request), "json")
290
        self.assertEqual(response.status_code, 202)
291

292
    def test_update_dettach_unassociated(self):
293
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
294
        request = {"floatingip": {}}
295
        with mocked_quotaholder():
296
            response = self.put(URL + "/%s" % ip.id, "user1",
297
                                json.dumps(request), "json")
298
        self.assertEqual(response.status_code, 400)
299

300
    def test_release_in_use(self):
301
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
302
        vm = mf.VirtualMachineFactory(userid="user1")
303
        request = {"floatingip": {
304
            "device_id": vm.id}
305
            }
306
        with mocked_quotaholder():
307
            response = self.put(URL + "/%s" % ip.id, "user1",
308
                                json.dumps(request), "json")
309
        self.assertEqual(response.status_code, 409)
310

311
    @patch("synnefo.db.models.get_rapi_client")
312
    def test_update_dettach(self, mrapi):
313
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
314
        request = {"floatingip": {
315
            "device_id": None}
316
            }
317
        mrapi().ModifyInstance.return_value = 42
318
        with mocked_quotaholder():
319
            response = self.put(URL + "/%s" % ip.id, "user1",
320
                                json.dumps(request), "json")
321
        self.assertEqual(response.status_code, 202)
322

323
    def test_update_dettach_unassociated(self):
324
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
325
        request = {"floatingip": {}}
326
        with mocked_quotaholder():
327
            response = self.put(URL + "/%s" % ip.id, "user1",
328
                                json.dumps(request), "json")
329
        self.assertEqual(response.status_code, 400)
330

331
    def test_release_in_use(self):
332
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
333
        vm = ip.nic.machine
334
        with mocked_quotaholder():
335
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
336
        self.assertFault(response, 409, "conflict")
337
    '''
338

    
339
    def test_update(self):
340
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
341
        with mocked_quotaholder():
342
            response = self.put(URL + "/%s" % ip.id, ip.userid)
343
        self.assertEqual(response.status_code, 501)
344

    
345
    def test_release(self):
346
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
347
        with mocked_quotaholder():
348
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
349
        self.assertSuccess(response)
350
        ips_after = floating_ips.filter(id=ip.id)
351
        self.assertEqual(len(ips_after), 0)
352

    
353
    @patch("synnefo.logic.backend", Mock())
354
    def test_delete_network_with_floating_ips(self):
355
        ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True,
356
                                   network=self.pool, nic=None)
357
        # Mark the network as non-pubic to not get 403
358
        network = ip.network
359
        network.public = False
360
        network.save()
361
        # Cannot remove network with floating IPs
362
        with mocked_quotaholder():
363
            response = self.delete(NETWORKS_URL + "/%s" % self.pool.id,
364
                                   self.pool.userid)
365
        self.assertConflict(response)
366
        # But we can with only deleted Floating Ips
367
        ip.deleted = True
368
        ip.save()
369
        with mocked_quotaholder():
370
            response = self.delete(NETWORKS_URL + "/%s" % self.pool.id,
371
                                   self.pool.userid)
372
        self.assertSuccess(response)
373

    
374
'''
375
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools")
376

377

378
class FloatingIPPoolsAPITest(BaseAPITest):
379
    def test_no_pool(self):
380
        response = self.get(POOLS_URL)
381
        self.assertSuccess(response)
382
        self.assertEqual(json.loads(response.content)["floating_ip_pools"], [])
383

384
    def test_list_pools(self):
385
        net = mf.NetworkWithSubnetFactory(floating_ip_pool=True,
386
                                          public=True,
387
                                          subnet__cidr="192.168.2.0/30",
388
                                          subnet__gateway="192.168.2.1",
389
                                          subnet__pool__size=1,
390
                                          subnet__pool__offset=1)
391
        mf.NetworkWithSubnetFactory(public=True, deleted=True)
392
        mf.NetworkWithSubnetFactory(public=False, deleted=False)
393
        mf.NetworkWithSubnetFactory(public=True, floating_ip_pool=False)
394
        response = self.get(POOLS_URL)
395
        self.assertSuccess(response)
396
        self.assertEqual(json.loads(response.content)["floating_ip_pools"],
397
                         [{"name": str(net.id), "size": 1, "free": 1}])
398

399

400
class FloatingIPActionsTest(BaseAPITest):
401
    def setUp(self):
402
        self.vm = VirtualMachineFactory()
403
        self.vm.operstate = "ACTIVE"
404
        self.vm.save()
405

406
    def test_bad_request(self):
407
        url = SERVERS_URL + "/%s/action" % self.vm.id
408
        response = self.post(url, self.vm.userid, json.dumps({}), "json")
409
        self.assertBadRequest(response)
410
        response = self.post(url, self.vm.userid,
411
                             json.dumps({"addFloatingIp": {}}),
412
                             "json")
413
        self.assertBadRequest(response)
414

415
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
416
    def test_add_floating_ip(self, mock):
417
        # Not exists
418
        url = SERVERS_URL + "/%s/action" % self.vm.id
419
        request = {"addFloatingIp": {"address": "10.0.0.1"}}
420
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
421
        self.assertItemNotFound(response)
422
        # In use
423
        ip = mf.IPv4AddressFactory(floating_ip=True, userid=self.vm.userid)
424
        request = {"addFloatingIp": {"address": ip.address}}
425
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
426
        self.assertConflict(response)
427
        # Success
428
        ip = mf.IPv4AddressFactory(floating_ip=True, nic=None,
429
                                   userid=self.vm.userid)
430
        request = {"addFloatingIp": {"address": ip.address}}
431
        mock().ModifyInstance.return_value = 1
432
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
433
        self.assertEqual(response.status_code, 202)
434
        ip_after = floating_ips.get(id=ip.id)
435
        self.assertEqual(ip_after.nic.machine, self.vm)
436
        nic = self.vm.nics.get()
437
        nic.state = "ACTIVE"
438
        nic.save()
439
        response = self.get(SERVERS_URL + "/%s" % self.vm.id,
440
                            self.vm.userid)
441
        self.assertSuccess(response)
442
        nic = json.loads(response.content)["server"]["attachments"][0]
443
        self.assertEqual(nic["OS-EXT-IPS:type"], "floating")
444

445
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
446
    def test_remove_floating_ip(self, mock):
447
        # Not exists
448
        url = SERVERS_URL + "/%s/action" % self.vm.id
449
        request = {"removeFloatingIp": {"address": "10.0.0.1"}}
450
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
451
        self.assertBadRequest(response)
452
        # Not In Use
453
        ip = mf.IPv4AddressFactory(floating_ip=True, nic=None,
454
                                   userid=self.vm.userid)
455
        request = {"removeFloatingIp": {"address": ip.address}}
456
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
457
        self.assertBadRequest(response)
458
        # Success
459
        ip = mf.IPv4AddressFactory(floating_ip=True,
460
                                   userid=self.vm.userid, nic__machine=self.vm)
461
        request = {"removeFloatingIp": {"address": ip.address}}
462
        mock().ModifyInstance.return_value = 2
463
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
464
        self.assertEqual(response.status_code, 202)
465
        # Yet used. Wait for the callbacks
466
        ip_after = floating_ips.get(id=ip.id)
467
        self.assertEqual(ip_after.nic.machine, self.vm)
468
'''