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

b/snf-cyclades-app/synnefo/api/tests/floating_ips.py
34 34
import json
35 35

  
36 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,
37
from synnefo.db.models import IPAddress
38
from synnefo.db.models_factory import (IPAddressFactory, NetworkFactory,
39 39
                                       VirtualMachineFactory,
40 40
                                       NetworkInterfaceFactory,
41 41
                                       BackendNetworkFactory)
......
52 52
NETWORKS_URL = join_urls(compute_path, "networks")
53 53
SERVERS_URL = join_urls(compute_path, "servers")
54 54

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

  
58 58

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

  
65 65
    def test_list_ips(self):
66
        ip = FloatingIPFactory(userid="user1")
67
        FloatingIPFactory(userid="user1", deleted=True)
66
        ip = IPAddressFactory(userid="user1")
67
        IPAddressFactory(userid="user1", deleted=True)
68 68
        with mocked_quotaholder():
69 69
            response = self.get(URL, "user1")
70 70
        self.assertSuccess(response)
......
75 75
                          str(ip.network.id)})
76 76

  
77 77
    def test_get_ip(self):
78
        ip = FloatingIPFactory(userid="user1")
78
        ip = IPAddressFactory(userid="user1")
79 79
        with mocked_quotaholder():
80 80
            response = self.get(URL + "/%s" % ip.id, "user1")
81 81
        self.assertSuccess(response)
......
86 86
                          str(ip.network.id)})
87 87

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

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

  
100 100
    def test_reserve(self):
101
        net = FloatingIPPoolFactory(userid="test_user",
101
        net = IPAddressPoolFactory(userid="test_user",
102 102
                                    subnet="192.168.2.0/24",
103 103
                                    gateway=None)
104 104
        request = {'pool': net.id}
105 105
        with mocked_quotaholder():
106 106
            response = self.post(URL, "test_user", json.dumps(request), "json")
107 107
        self.assertSuccess(response)
108
        ip = FloatingIP.objects.get()
108
        ip = IPAddress.objects.get()
109 109
        self.assertEqual(ip.ipv4, "192.168.2.1")
110 110
        self.assertEqual(ip.machine, None)
111 111
        self.assertEqual(ip.network, net)
......
120 120
            response = self.post(URL, "test_user", json.dumps({}), "json")
121 121
        self.assertFault(response, 413, "overLimit")
122 122
        # Full network
123
        FloatingIPPoolFactory(userid="test_user",
123
        IPAddressPoolFactory(userid="test_user",
124 124
                              subnet="192.168.2.0/32",
125 125
                              gateway=None)
126 126
        with mocked_quotaholder():
127 127
            response = self.post(URL, "test_user", json.dumps({}), "json")
128 128
        self.assertFault(response, 413, "overLimit")
129 129
        # Success
130
        net2 = FloatingIPPoolFactory(userid="test_user",
130
        net2 = IPAddressPoolFactory(userid="test_user",
131 131
                                     subnet="192.168.2.0/24",
132 132
                                     gateway=None)
133 133
        with mocked_quotaholder():
134 134
            response = self.post(URL, "test_user", json.dumps({}), "json")
135 135
        self.assertSuccess(response)
136
        ip = FloatingIP.objects.get()
136
        ip = IPAddress.objects.get()
137 137
        self.assertEqual(json.loads(response.content)["floating_ip"],
138 138
                         {"instance_id": None, "ip": "192.168.2.1",
139 139
                          "fixed_ip": None, "id": str(ip.id),
140 140
                          "pool": str(net2.id)})
141 141

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

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

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

  
196 196
    def test_release_in_use(self):
197
        ip = FloatingIPFactory()
197
        ip = IPAddressFactory()
198 198
        vm = ip.machine
199 199
        vm.operstate = "ACTIVE"
200 200
        vm.userid = ip.userid
......
218 218
        self.assertFault(response, 409, "conflict")
219 219

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

  
228 228
    @patch("synnefo.logic.backend", Mock())
229 229
    def test_delete_network_with_floating_ips(self):
230
        ip = FloatingIPFactory(machine=None, network__flavor="IP_LESS_ROUTED")
230
        ip = IPAddressFactory(machine=None, network__flavor="IP_LESS_ROUTED")
231 231
        net = ip.network
232 232
        # Can not remove network with floating IPs
233 233
        with mocked_quotaholder():
......
246 246
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools")
247 247

  
248 248

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

  
255 255
    def test_list_pools(self):
256
        net = FloatingIPPoolFactory(subnet="192.168.0.0/30",
256
        net = IPAddressPoolFactory(subnet="192.168.0.0/30",
257 257
                                    gateway="192.168.0.1")
258 258
        NetworkFactory(public=True, deleted=True)
259 259
        NetworkFactory(public=False, deleted=False)
......
264 264
                         [{"name": str(net.id), "size": 4, "free": 1}])
265 265

  
266 266

  
267
class FloatingIPActionsTest(BaseAPITest):
267
class IPAddressActionsTest(BaseAPITest):
268 268
    def setUp(self):
269 269
        vm = VirtualMachineFactory()
270 270
        vm.operstate = "ACTIVE"
......
289 289
        self.assertItemNotFound(response)
290 290
        # In use
291 291
        vm1 = VirtualMachineFactory()
292
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=vm1)
292
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=vm1)
293 293
        BackendNetworkFactory(network=ip1.network, backend=vm1.backend,
294 294
                              operstate='ACTIVE')
295 295
        request = {"addFloatingIp": {"address": ip1.ipv4}}
296 296
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
297 297
        self.assertFault(response, 409, "conflict")
298 298
        # Success
299
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None)
299
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=None)
300 300
        BackendNetworkFactory(network=ip1.network, backend=self.vm.backend,
301 301
                              operstate='ACTIVE')
302 302
        request = {"addFloatingIp": {"address": ip1.ipv4}}
303 303
        mock().ModifyInstance.return_value = 1
304 304
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
305 305
        self.assertEqual(response.status_code, 202)
306
        ip1_after = FloatingIP.objects.get(id=ip1.id)
306
        ip1_after = IPAddress.objects.get(id=ip1.id)
307 307
        self.assertEqual(ip1_after.machine, self.vm)
308 308
        self.assertTrue(ip1_after.in_use())
309 309
        nic = self.vm.nics.get(ipv4=ip1_after.ipv4)
......
323 323
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
324 324
        self.assertItemNotFound(response)
325 325
        # Not In Use
326
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None)
326
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=None)
327 327
        request = {"removeFloatingIp": {"address": ip1.ipv4}}
328 328
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
329 329
        self.assertItemNotFound(response)
330 330
        # Success
331
        ip1 = FloatingIPFactory(userid=self.vm.userid, machine=self.vm)
331
        ip1 = IPAddressFactory(userid=self.vm.userid, machine=self.vm)
332 332
        NetworkInterfaceFactory(machine=self.vm, ipv4=ip1.ipv4)
333 333
        request = {"removeFloatingIp": {"address": ip1.ipv4}}
334 334
        mock().ModifyInstance.return_value = 2
335 335
        response = self.post(url, self.vm.userid, json.dumps(request), "json")
336 336
        self.assertEqual(response.status_code, 202)
337 337
        # Yet used. Wait for the callbacks
338
        ip1_after = FloatingIP.objects.get(id=ip1.id)
338
        ip1_after = IPAddress.objects.get(id=ip1.id)
339 339
        self.assertEqual(ip1_after.machine, self.vm)
340 340
        self.assertTrue(ip1_after.in_use())

Also available in: Unified diff