Statistics
| Branch: | Tag: | Revision:

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

History | View | Annotate | Download (14.8 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 IPAddress
38
from synnefo.db.models_factory import (IPAddressFactory, 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
IPAddressPoolFactory = partial(NetworkFactory, public=True, deleted=False,
56
                                floating_ip_pool=True)
57

    
58

    
59
class IPAddressAPITest(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 = IPAddressFactory(userid="user1")
67
        IPAddressFactory(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 = IPAddressFactory(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 = IPAddressFactory(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 = IPAddressFactory(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 = IPAddressPoolFactory(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 = IPAddress.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": str(ip.id),
115
                          "pool": str(net.id)})
116

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

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

    
150
    def test_reserve_with_address(self):
151
        net = IPAddressPoolFactory(userid="test_user",
152
                                    subnet="192.168.2.0/24")
153
        request = {'pool': net.id, "address": "192.168.2.10"}
154
        with mocked_quotaholder():
155
            response = self.post(URL, "test_user", json.dumps(request), "json")
156
        self.assertSuccess(response)
157
        ip = IPAddress.objects.get()
158
        self.assertEqual(json.loads(response.content)["floating_ip"],
159
                         {"instance_id": None, "ip": "192.168.2.10",
160
                          "fixed_ip": None, "id": str(ip.id), "pool":
161
                          str(net.id)})
162

    
163
        # Already reserved
164
        IPAddressFactory(network=net, ipv4="192.168.2.3")
165
        request = {'pool': net.id, "address": "192.168.2.3"}
166
        with mocked_quotaholder():
167
            response = self.post(URL, "test_user", json.dumps(request), "json")
168
        self.assertFault(response, 409, "conflict")
169

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

    
190
        # Address out of pool
191
        request = {'pool': net.id, "address": "192.168.3.5"}
192
        with mocked_quotaholder():
193
            response = self.post(URL, "test_user", json.dumps(request), "json")
194
        self.assertBadRequest(response)
195

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

    
220
    def test_release(self):
221
        ip = IPAddressFactory(machine=None)
222
        with mocked_quotaholder():
223
            response = self.delete(URL + "/%s" % ip.id, ip.userid)
224
        self.assertSuccess(response)
225
        ips_after = IPAddress.objects.filter(id=ip.id)
226
        self.assertEqual(len(ips_after), 0)
227

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

    
245

    
246
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools")
247

    
248

    
249
class IPAddressPoolsAPITest(BaseAPITest):
250
    def test_no_pool(self):
251
        response = self.get(POOLS_URL)
252
        self.assertSuccess(response)
253
        self.assertEqual(json.loads(response.content)["floating_ip_pools"], [])
254

    
255
    def test_list_pools(self):
256
        net = IPAddressPoolFactory(subnet="192.168.0.0/30",
257
                                    gateway="192.168.0.1")
258
        NetworkFactory(public=True, deleted=True)
259
        NetworkFactory(public=False, deleted=False)
260
        NetworkFactory(public=True, deleted=False)
261
        response = self.get(POOLS_URL)
262
        self.assertSuccess(response)
263
        self.assertEqual(json.loads(response.content)["floating_ip_pools"],
264
                         [{"name": str(net.id), "size": 4, "free": 1}])
265

    
266

    
267
class IPAddressActionsTest(BaseAPITest):
268
    def setUp(self):
269
        vm = VirtualMachineFactory()
270
        vm.operstate = "ACTIVE"
271
        vm.save()
272
        self.vm = vm
273

    
274
    def test_bad_request(self):
275
        url = SERVERS_URL + "/%s/action" % self.vm.id
276
        response = self.post(url, self.vm.userid, json.dumps({}), "json")
277
        self.assertBadRequest(response)
278
        response = self.post(url, self.vm.userid,
279
                             json.dumps({"addFloatingIp": {}}),
280
                             "json")
281
        self.assertBadRequest(response)
282

    
283
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
284
    def test_add_floating_ip(self, mock):
285
        # Not exists
286
        url = SERVERS_URL + "/%s/action" % self.vm.id
287
        request = {"addFloatingIp": {"address": "10.0.0.1"}}
288
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
289
        self.assertItemNotFound(response)
290
        # In use
291
        vm1 = VirtualMachineFactory()
292
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=vm1)
293
        BackendNetworkFactory(network=ip1.network, backend=vm1.backend,
294
                              operstate='ACTIVE')
295
        request = {"addFloatingIp": {"address": ip1.ipv4}}
296
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
297
        self.assertFault(response, 409, "conflict")
298
        # Success
299
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=None)
300
        BackendNetworkFactory(network=ip1.network, backend=self.vm.backend,
301
                              operstate='ACTIVE')
302
        request = {"addFloatingIp": {"address": ip1.ipv4}}
303
        mock().ModifyInstance.return_value = 1
304
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
305
        self.assertEqual(response.status_code, 202)
306
        ip1_after = IPAddress.objects.get(id=ip1.id)
307
        self.assertEqual(ip1_after.machine, self.vm)
308
        self.assertTrue(ip1_after.in_use())
309
        nic = self.vm.nics.get(ipv4=ip1_after.ipv4)
310
        nic.state = "ACTIVE"
311
        nic.save()
312
        response = self.get(SERVERS_URL + "/%s" % self.vm.id,
313
                            self.vm.userid)
314
        self.assertSuccess(response)
315
        nic = json.loads(response.content)["server"]["attachments"][0]
316
        self.assertEqual(nic["OS-EXT-IPS:type"], "floating")
317

    
318
    @patch('synnefo.logic.rapi_pool.GanetiRapiClient')
319
    def test_remove_floating_ip(self, mock):
320
        # Not exists
321
        url = SERVERS_URL + "/%s/action" % self.vm.id
322
        request = {"removeFloatingIp": {"address": "10.0.0.1"}}
323
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
324
        self.assertItemNotFound(response)
325
        # Not In Use
326
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=None)
327
        request = {"removeFloatingIp": {"address": ip1.ipv4}}
328
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
329
        self.assertItemNotFound(response)
330
        # Success
331
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=self.vm)
332
        NetworkInterfaceFactory(machine=self.vm, ipv4=ip1.ipv4)
333
        request = {"removeFloatingIp": {"address": ip1.ipv4}}
334
        mock().ModifyInstance.return_value = 2
335
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
336
        self.assertEqual(response.status_code, 202)
337
        # Yet used. Wait for the callbacks
338
        ip1_after = IPAddress.objects.get(id=ip1.id)
339
        self.assertEqual(ip1_after.machine, self.vm)
340
        self.assertTrue(ip1_after.in_use())