Statistics
| Branch: | Tag: | Revision:

root / snf-cyclades-app / synnefo / api / test / floating_ips.py @ 2cd55da6

History | View | Annotate | Download (14.2 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
import json
35

    
36
from snf_django.utils.testing import BaseAPITest, mocked_quotaholder
37
from synnefo.db.models import FloatingIP
38
from synnefo.db.models_factory import (FloatingIPFactory, NetworkFactory,
39
                                       VirtualMachineFactory,
40
                                       NetworkInterfaceFactory,
41
                                       BackendNetworkFactory)
42
from mock import patch, Mock
43
from functools import partial
44

    
45
from synnefo.cyclades_settings import cyclades_services
46
from synnefo.lib.services import get_service_path
47
from synnefo.lib import join_urls
48

    
49

    
50
compute_path = get_service_path(cyclades_services, "compute", version="v2.0")
51
URL = join_urls(compute_path, "os-floating-ips")
52
NETWORKS_URL = join_urls(compute_path, "networks")
53
SERVERS_URL = join_urls(compute_path, "servers")
54

    
55
FloatingIPPoolFactory = partial(NetworkFactory, public=True, deleted=False,
56
                                floating_ip_pool=True)
57

    
58

    
59
class FloatingIPAPITest(BaseAPITest):
60
    def test_no_floating_ip(self):
61
        response = self.get(URL)
62
        self.assertSuccess(response)
63
        self.assertEqual(json.loads(response.content)["floating_ips"], [])
64

    
65
    def test_list_ips(self):
66
        ip = FloatingIPFactory(userid="user1")
67
        FloatingIPFactory(userid="user1", deleted=True)
68
        with mocked_quotaholder():
69
            response = self.get(URL, "user1")
70
        self.assertSuccess(response)
71
        api_ip = json.loads(response.content)["floating_ips"][0]
72
        self.assertEqual(api_ip,
73
                         {"instance_id": str(ip.machine.id), "ip": ip.ipv4,
74
                          "fixed_ip": None, "id": str(ip.id),  "pool":
75
                          str(ip.network.id)})
76

    
77
    def test_get_ip(self):
78
        ip = FloatingIPFactory(userid="user1")
79
        with mocked_quotaholder():
80
            response = self.get(URL + "/%s" % ip.id, "user1")
81
        self.assertSuccess(response)
82
        api_ip = json.loads(response.content)["floating_ip"]
83
        self.assertEqual(api_ip,
84
                         {"instance_id": str(ip.machine.id), "ip": ip.ipv4,
85
                          "fixed_ip": None, "id": str(ip.id),  "pool":
86
                          str(ip.network.id)})
87

    
88
    def test_wrong_user(self):
89
        ip = FloatingIPFactory(userid="user1")
90
        with mocked_quotaholder():
91
            response = self.delete(URL + "/%s" % ip.id, "user2")
92
        self.assertItemNotFound(response)
93

    
94
    def test_deleted_ip(self):
95
        ip = FloatingIPFactory(userid="user1", deleted=True)
96
        with mocked_quotaholder():
97
            response = self.delete(URL + "/%s" % ip.id, "user1")
98
        self.assertItemNotFound(response)
99

    
100
    def test_reserve(self):
101
        net = FloatingIPPoolFactory(userid="test_user",
102
                                    subnet="192.168.2.0/24",
103
                                    gateway=None)
104
        request = {'pool': net.id}
105
        with mocked_quotaholder():
106
            response = self.post(URL, "test_user", json.dumps(request), "json")
107
        self.assertSuccess(response)
108
        ip = FloatingIP.objects.get()
109
        self.assertEqual(ip.ipv4, "192.168.2.1")
110
        self.assertEqual(ip.machine, None)
111
        self.assertEqual(ip.network, net)
112
        self.assertEqual(json.loads(response.content)["floating_ip"],
113
                         {"instance_id": None, "ip": "192.168.2.1",
114
                          "fixed_ip": None, "id": "1", "pool": "1"})
115

    
116
    def test_reserve_no_pool(self):
117
        # No networks
118
        with mocked_quotaholder():
119
            response = self.post(URL, "test_user", json.dumps({}), "json")
120
        self.assertFault(response, 413, "overLimit")
121
        # Full network
122
        FloatingIPPoolFactory(userid="test_user",
123
                              subnet="192.168.2.0/32",
124
                              gateway=None)
125
        with mocked_quotaholder():
126
            response = self.post(URL, "test_user", json.dumps({}), "json")
127
        self.assertFault(response, 413, "overLimit")
128
        # Success
129
        net2 = FloatingIPPoolFactory(userid="test_user",
130
                                     subnet="192.168.2.0/24",
131
                                     gateway=None)
132
        with mocked_quotaholder():
133
            response = self.post(URL, "test_user", json.dumps({}), "json")
134
        self.assertSuccess(response)
135
        self.assertEqual(json.loads(response.content)["floating_ip"],
136
                         {"instance_id": None, "ip": "192.168.2.1",
137
                          "fixed_ip": None, "id": "1", "pool": str(net2.id)})
138

    
139
    def test_reserve_full(self):
140
        net = FloatingIPPoolFactory(userid="test_user",
141
                                    subnet="192.168.2.0/32")
142
        request = {'pool': net.id}
143
        with mocked_quotaholder():
144
            response = self.post(URL, "test_user", json.dumps(request), "json")
145
        self.assertEqual(response.status_code, 413)
146

    
147
    def test_reserve_with_address(self):
148
        net = FloatingIPPoolFactory(userid="test_user",
149
                                    subnet="192.168.2.0/24")
150
        request = {'pool': net.id, "address": "192.168.2.10"}
151
        with mocked_quotaholder():
152
            response = self.post(URL, "test_user", json.dumps(request), "json")
153
        self.assertSuccess(response)
154
        self.assertEqual(json.loads(response.content)["floating_ip"],
155
                         {"instance_id": None, "ip": "192.168.2.10",
156
                          "fixed_ip": None, "id": "1", "pool": "1"})
157

    
158
        # Already reserved
159
        FloatingIPFactory(network=net, ipv4="192.168.2.3")
160
        request = {'pool': net.id, "address": "192.168.2.3"}
161
        with mocked_quotaholder():
162
            response = self.post(URL, "test_user", json.dumps(request), "json")
163
        self.assertFault(response, 409, "conflict")
164

    
165
        # Already used
166
        pool = net.get_pool()
167
        pool.reserve("192.168.2.5")
168
        pool.save()
169
        # ..by another_user
170
        nic = NetworkInterfaceFactory(network=net, ipv4="192.168.2.5",
171
                                      machine__userid="test2")
172
        request = {'pool': net.id, "address": "192.168.2.5"}
173
        with mocked_quotaholder():
174
            response = self.post(URL, "test_user", json.dumps(request), "json")
175
        self.assertFault(response, 409, "conflict")
176
        # ..and by him
177
        nic.delete()
178
        NetworkInterfaceFactory(network=net, ipv4="192.168.2.5",
179
                                machine__userid="test_user")
180
        request = {'pool': net.id, "address": "192.168.2.5"}
181
        with mocked_quotaholder():
182
            response = self.post(URL, "test_user", json.dumps(request), "json")
183
        self.assertSuccess(response)
184

    
185
        # Address out of pool
186
        request = {'pool': net.id, "address": "192.168.3.5"}
187
        with mocked_quotaholder():
188
            response = self.post(URL, "test_user", json.dumps(request), "json")
189
        self.assertBadRequest(response)
190

    
191
    def test_release_in_use(self):
192
        ip = FloatingIPFactory()
193
        vm = ip.machine
194
        vm.operstate = "ACTIVE"
195
        vm.userid = ip.userid
196
        vm.save()
197
        vm.nics.create(index=0, ipv4=ip.ipv4, network=ip.network,
198
                       state="ACTIVE")
199
        with mocked_quotaholder():
200
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
201
        self.assertFault(response, 409, "conflict")
202
        # Also send a notification to remove the NIC and assert that FIP is in
203
        # use until notification from ganeti arrives
204
        request = {"removeFloatingIp": {"address": ip.ipv4}}
205
        url = SERVERS_URL + "/%s/action" % vm.id
206
        with patch('synnefo.logic.rapi_pool.GanetiRapiClient') as c:
207
            c().ModifyInstance.return_value = 10
208
            response = self.post(url, vm.userid, json.dumps(request),
209
                                 "json")
210
        self.assertEqual(response.status_code, 202)
211
        with mocked_quotaholder():
212
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
213
        self.assertFault(response, 409, "conflict")
214

    
215
    def test_release(self):
216
        ip = FloatingIPFactory(machine=None)
217
        with mocked_quotaholder():
218
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
219
        self.assertSuccess(response)
220
        ips_after = FloatingIP.objects.filter(id=ip.id)
221
        self.assertEqual(len(ips_after), 0)
222

    
223
    @patch("synnefo.logic.backend", Mock())
224
    def test_delete_network_with_floating_ips(self):
225
        ip = FloatingIPFactory(machine=None, network__flavor="IP_LESS_ROUTED")
226
        net = ip.network
227
        # Can not remove network with floating IPs
228
        with mocked_quotaholder():
229
            response = self.delete(NETWORKS_URL + "/%s" % net.id,
230
                                   net.userid)
231
        self.assertFault(response, 421, "networkInUse")
232
        # But we can with only deleted Floating Ips
233
        ip.deleted = True
234
        ip.save()
235
        with mocked_quotaholder():
236
            response = self.delete(NETWORKS_URL + "/%s" % net.id,
237
                                   net.userid)
238
        self.assertSuccess(response)
239

    
240

    
241
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools")
242

    
243

    
244
class FloatingIPPoolsAPITest(BaseAPITest):
245
    def test_no_pool(self):
246
        response = self.get(POOLS_URL)
247
        self.assertSuccess(response)
248
        self.assertEqual(json.loads(response.content)["floating_ip_pools"], [])
249

    
250
    def test_list_pools(self):
251
        net = FloatingIPPoolFactory()
252
        NetworkFactory(public=True, deleted=True)
253
        NetworkFactory(public=False, deleted=False)
254
        NetworkFactory(public=True, deleted=False)
255
        response = self.get(POOLS_URL)
256
        self.assertSuccess(response)
257
        self.assertEqual(json.loads(response.content)["floating_ip_pools"],
258
                        [{"name": str(net.id)}])
259

    
260

    
261
class FloatingIPActionsTest(BaseAPITest):
262
    def setUp(self):
263
        vm = VirtualMachineFactory()
264
        vm.operstate = "ACTIVE"
265
        vm.save()
266
        self.vm = vm
267

    
268
    def test_bad_request(self):
269
        url = SERVERS_URL + "/%s/action" % self.vm.id
270
        response = self.post(url, self.vm.userid, json.dumps({}), "json")
271
        self.assertBadRequest(response)
272
        response = self.post(url, self.vm.userid,
273
                             json.dumps({"addFloatingIp": {}}),
274
                             "json")
275
        self.assertBadRequest(response)
276

    
277
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
278
    def test_add_floating_ip(self, mock):
279
        # Not exists
280
        url = SERVERS_URL + "/%s/action" % self.vm.id
281
        request = {"addFloatingIp": {"address": "10.0.0.1"}}
282
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
283
        self.assertItemNotFound(response)
284
        # In use
285
        vm1 = VirtualMachineFactory()
286
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=vm1)
287
        BackendNetworkFactory(network=ip1.network, backend=vm1.backend,
288
                              operstate='ACTIVE')
289
        request = {"addFloatingIp": {"address": ip1.ipv4}}
290
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
291
        self.assertFault(response, 409, "conflict")
292
        # Success
293
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None)
294
        BackendNetworkFactory(network=ip1.network, backend=self.vm.backend,
295
                              operstate='ACTIVE')
296
        request = {"addFloatingIp": {"address": ip1.ipv4}}
297
        mock().ModifyInstance.return_value = 1
298
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
299
        self.assertEqual(response.status_code, 202)
300
        ip1_after = FloatingIP.objects.get(id=ip1.id)
301
        self.assertEqual(ip1_after.machine, self.vm)
302
        self.assertTrue(ip1_after.in_use())
303

    
304
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
305
    def test_remove_floating_ip(self, mock):
306
        # Not exists
307
        url = SERVERS_URL + "/%s/action" % self.vm.id
308
        request = {"removeFloatingIp": {"address": "10.0.0.1"}}
309
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
310
        self.assertItemNotFound(response)
311
        # Not In Use
312
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None)
313
        request = {"removeFloatingIp": {"address": ip1.ipv4}}
314
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
315
        self.assertItemNotFound(response)
316
        # Success
317
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=self.vm)
318
        NetworkInterfaceFactory(machine=self.vm, ipv4=ip1.ipv4)
319
        request = {"removeFloatingIp": {"address": ip1.ipv4}}
320
        mock().ModifyInstance.return_value = 2
321
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
322
        self.assertEqual(response.status_code, 202)
323
        # Yet used. Wait for the callbacks
324
        ip1_after = FloatingIP.objects.get(id=ip1.id)
325
        self.assertEqual(ip1_after.machine, self.vm)
326
        self.assertTrue(ip1_after.in_use())