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