Revision fde2c1f7 snf-cyclades-app/synnefo/api/tests/floating_ips.py
b/snf-cyclades-app/synnefo/api/tests/floating_ips.py | ||
---|---|---|
35 | 35 |
|
36 | 36 |
from snf_django.utils.testing import BaseAPITest, mocked_quotaholder |
37 | 37 |
from synnefo.db.models import IPAddress |
38 |
from synnefo.db.models_factory import (FloatingIPFactory, NetworkFactory, |
|
39 |
VirtualMachineFactory, |
|
40 |
NetworkInterfaceFactory, |
|
41 |
BackendNetworkFactory) |
|
38 |
from synnefo.db import models_factory as mf |
|
39 |
from synnefo.db.models_factory import (NetworkFactory, |
|
40 |
VirtualMachineFactory) |
|
42 | 41 |
from mock import patch, Mock |
43 | 42 |
from functools import partial |
44 | 43 |
|
... | ... | |
59 | 58 |
|
60 | 59 |
|
61 | 60 |
class FloatingIPAPITest(BaseAPITest): |
61 |
def setUp(self): |
|
62 |
self.pool = mf.NetworkWithSubnetFactory(floating_ip_pool=True, |
|
63 |
public=True, |
|
64 |
subnet__cidr="192.168.2.0/24", |
|
65 |
subnet__gateway="192.168.2.1") |
|
66 |
|
|
62 | 67 |
def test_no_floating_ip(self): |
63 | 68 |
response = self.get(URL) |
64 | 69 |
self.assertSuccess(response) |
65 | 70 |
self.assertEqual(json.loads(response.content)["floating_ips"], []) |
66 | 71 |
|
67 | 72 |
def test_list_ips(self): |
68 |
ip = FloatingIPFactory(userid="user1") |
|
69 |
FloatingIPFactory(userid="user1", deleted=True) |
|
73 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
70 | 74 |
with mocked_quotaholder(): |
71 | 75 |
response = self.get(URL, "user1") |
72 | 76 |
self.assertSuccess(response) |
73 | 77 |
api_ip = json.loads(response.content)["floating_ips"][0] |
74 | 78 |
self.assertEqual(api_ip, |
75 |
{"instance_id": str(ip.machine.id), "ip": ip.ipv4, |
|
76 |
"fixed_ip": None, "id": str(ip.id), "pool": |
|
77 |
str(ip.network.id)}) |
|
79 |
{"instance_id": str(ip.nic.machine_id), |
|
80 |
"ip": ip.address, |
|
81 |
"fixed_ip": None, |
|
82 |
"id": str(ip.id), |
|
83 |
"pool": str(ip.network_id)}) |
|
78 | 84 |
|
79 | 85 |
def test_get_ip(self): |
80 |
ip = FloatingIPFactory(userid="user1")
|
|
86 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True)
|
|
81 | 87 |
with mocked_quotaholder(): |
82 | 88 |
response = self.get(URL + "/%s" % ip.id, "user1") |
83 | 89 |
self.assertSuccess(response) |
84 | 90 |
api_ip = json.loads(response.content)["floating_ip"] |
85 | 91 |
self.assertEqual(api_ip, |
86 |
{"instance_id": str(ip.machine.id), "ip": ip.ipv4, |
|
87 |
"fixed_ip": None, "id": str(ip.id), "pool": |
|
88 |
str(ip.network.id)}) |
|
92 |
{"instance_id": str(ip.nic.machine_id), |
|
93 |
"ip": ip.address, |
|
94 |
"fixed_ip": None, |
|
95 |
"id": str(ip.id), |
|
96 |
"pool": str(ip.network_id)}) |
|
89 | 97 |
|
90 | 98 |
def test_wrong_user(self): |
91 |
ip = FloatingIPFactory(userid="user1") |
|
92 |
with mocked_quotaholder(): |
|
93 |
response = self.delete(URL + "/%s" % ip.id, "user2") |
|
99 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
100 |
response = self.delete(URL + "/%s" % ip.id, "user2") |
|
94 | 101 |
self.assertItemNotFound(response) |
95 | 102 |
|
96 | 103 |
def test_deleted_ip(self): |
97 |
ip = FloatingIPFactory(userid="user1", deleted=True)
|
|
98 |
with mocked_quotaholder():
|
|
99 |
response = self.delete(URL + "/%s" % ip.id, "user1")
|
|
104 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True,
|
|
105 |
deleted=True)
|
|
106 |
response = self.delete(URL + "/%s" % ip.id, "user1") |
|
100 | 107 |
self.assertItemNotFound(response) |
101 | 108 |
|
102 | 109 |
def test_reserve(self): |
103 |
net = FloatingIPPoolFactory(userid="test_user", |
|
104 |
subnet="192.168.2.0/24", |
|
105 |
gateway=None) |
|
106 |
request = {'pool': net.id} |
|
110 |
request = {'pool': self.pool.id} |
|
107 | 111 |
with mocked_quotaholder(): |
108 | 112 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
109 | 113 |
self.assertSuccess(response) |
110 | 114 |
ip = floating_ips.get() |
111 |
self.assertEqual(ip.ipv4, "192.168.2.1")
|
|
112 |
self.assertEqual(ip.machine, None)
|
|
113 |
self.assertEqual(ip.network, net)
|
|
115 |
self.assertEqual(ip.address, "192.168.2.2")
|
|
116 |
self.assertEqual(ip.nic, None)
|
|
117 |
self.assertEqual(ip.network, self.pool)
|
|
114 | 118 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
115 |
{"instance_id": None, "ip": "192.168.2.1",
|
|
119 |
{"instance_id": None, "ip": "192.168.2.2",
|
|
116 | 120 |
"fixed_ip": None, "id": str(ip.id), |
117 |
"pool": str(net.id)})
|
|
121 |
"pool": str(self.pool.id)})
|
|
118 | 122 |
|
119 | 123 |
def test_reserve_no_pool(self): |
120 |
# No networks |
|
121 |
with mocked_quotaholder(): |
|
122 |
response = self.post(URL, "test_user", json.dumps({}), "json") |
|
123 |
self.assertFault(response, 413, "overLimit") |
|
124 |
# No floating IP pools |
|
125 |
self.pool.delete() |
|
126 |
response = self.post(URL, "test_user", json.dumps({}), "json") |
|
127 |
self.assertFault(response, 503, 'serviceUnavailable') |
|
128 |
|
|
124 | 129 |
# Full network |
125 |
FloatingIPPoolFactory(userid="test_user", |
|
126 |
subnet="192.168.2.0/32", |
|
127 |
gateway=None) |
|
128 |
with mocked_quotaholder(): |
|
129 |
response = self.post(URL, "test_user", json.dumps({}), "json") |
|
130 |
self.assertFault(response, 413, "overLimit") |
|
131 |
# Success |
|
132 |
net2 = FloatingIPPoolFactory(userid="test_user", |
|
133 |
subnet="192.168.2.0/24", |
|
134 |
gateway=None) |
|
135 |
with mocked_quotaholder(): |
|
136 |
response = self.post(URL, "test_user", json.dumps({}), "json") |
|
137 |
self.assertSuccess(response) |
|
138 |
ip = floating_ips.get() |
|
139 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
|
140 |
{"instance_id": None, "ip": "192.168.2.1", |
|
141 |
"fixed_ip": None, "id": str(ip.id), |
|
142 |
"pool": str(net2.id)}) |
|
130 |
pool = mf.NetworkWithSubnetFactory(floating_ip_pool=True, |
|
131 |
public=True, |
|
132 |
subnet__cidr="192.168.2.0/31", |
|
133 |
subnet__gateway="192.168.2.1") |
|
134 |
response = self.post(URL, "test_user", json.dumps({}), "json") |
|
135 |
self.assertFault(response, 503, 'serviceUnavailable') |
|
143 | 136 |
|
144 |
def test_reserve_full(self): |
|
145 |
net = FloatingIPPoolFactory(userid="test_user", |
|
146 |
subnet="192.168.2.0/32") |
|
147 |
request = {'pool': net.id} |
|
148 |
with mocked_quotaholder(): |
|
149 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
|
150 |
self.assertEqual(response.status_code, 413) |
|
137 |
request = {'pool': pool.id} |
|
138 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
|
139 |
self.assertConflict(response) |
|
151 | 140 |
|
152 | 141 |
def test_reserve_with_address(self): |
153 |
net = FloatingIPPoolFactory(userid="test_user", |
|
154 |
subnet="192.168.2.0/24") |
|
155 |
request = {'pool': net.id, "address": "192.168.2.10"} |
|
142 |
request = {'pool': self.pool.id, "address": "192.168.2.10"} |
|
156 | 143 |
with mocked_quotaholder(): |
157 | 144 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
158 | 145 |
self.assertSuccess(response) |
159 | 146 |
ip = floating_ips.get() |
160 | 147 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
161 | 148 |
{"instance_id": None, "ip": "192.168.2.10", |
162 |
"fixed_ip": None, "id": str(ip.id), "pool":
|
|
163 |
str(net.id)})
|
|
149 |
"fixed_ip": None, "id": str(ip.id), |
|
150 |
"pool": str(self.pool.id)})
|
|
164 | 151 |
|
165 | 152 |
# Already reserved |
166 |
FloatingIPFactory(network=net, ipv4="192.168.2.3") |
|
167 |
request = {'pool': net.id, "address": "192.168.2.3"} |
|
168 | 153 |
with mocked_quotaholder(): |
169 | 154 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
170 | 155 |
self.assertFault(response, 409, "conflict") |
171 | 156 |
|
172 |
# Already used |
|
173 |
pool = net.get_pool() |
|
174 |
pool.reserve("192.168.2.5") |
|
175 |
pool.save() |
|
176 |
# ..by another_user |
|
177 |
nic = NetworkInterfaceFactory(network=net, ipv4="192.168.2.5", |
|
178 |
machine__userid="test2") |
|
179 |
request = {'pool': net.id, "address": "192.168.2.5"} |
|
157 |
# Used by instance |
|
158 |
self.pool.reserve_address("192.168.2.20") |
|
159 |
request = {'pool': self.pool.id, "address": "192.168.2.20"} |
|
180 | 160 |
with mocked_quotaholder(): |
181 | 161 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
182 | 162 |
self.assertFault(response, 409, "conflict") |
183 |
# ..and by him |
|
184 |
nic.delete() |
|
185 |
NetworkInterfaceFactory(network=net, ipv4="192.168.2.5", |
|
186 |
machine__userid="test_user") |
|
187 |
request = {'pool': net.id, "address": "192.168.2.5"} |
|
188 |
with mocked_quotaholder(): |
|
189 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
|
190 |
self.assertSuccess(response) |
|
191 | 163 |
|
192 | 164 |
# Address out of pool |
193 |
request = {'pool': net.id, "address": "192.168.3.5"}
|
|
165 |
request = {'pool': self.pool.id, "address": "192.168.3.5"}
|
|
194 | 166 |
with mocked_quotaholder(): |
195 | 167 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
196 | 168 |
self.assertBadRequest(response) |
197 | 169 |
|
198 | 170 |
def test_release_in_use(self): |
199 |
ip = FloatingIPFactory() |
|
200 |
vm = ip.machine |
|
201 |
vm.operstate = "ACTIVE" |
|
202 |
vm.userid = ip.userid |
|
203 |
vm.save() |
|
204 |
vm.nics.create(index=0, ipv4=ip.ipv4, network=ip.network, |
|
205 |
state="ACTIVE") |
|
171 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
172 |
vm = ip.nic.machine |
|
206 | 173 |
with mocked_quotaholder(): |
207 | 174 |
response = self.delete(URL + "/%s" % ip.id, ip.userid) |
208 | 175 |
self.assertFault(response, 409, "conflict") |
209 | 176 |
# Also send a notification to remove the NIC and assert that FIP is in |
210 | 177 |
# use until notification from ganeti arrives |
211 |
request = {"removeFloatingIp": {"address": ip.ipv4}}
|
|
178 |
request = {"removeFloatingIp": {"address": ip.address}}
|
|
212 | 179 |
url = SERVERS_URL + "/%s/action" % vm.id |
213 | 180 |
with patch('synnefo.logic.rapi_pool.GanetiRapiClient') as c: |
214 | 181 |
c().ModifyInstance.return_value = 10 |
... | ... | |
220 | 187 |
self.assertFault(response, 409, "conflict") |
221 | 188 |
|
222 | 189 |
def test_release(self): |
223 |
ip = FloatingIPFactory(machine=None)
|
|
190 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None)
|
|
224 | 191 |
with mocked_quotaholder(): |
225 | 192 |
response = self.delete(URL + "/%s" % ip.id, ip.userid) |
226 | 193 |
self.assertSuccess(response) |
... | ... | |
229 | 196 |
|
230 | 197 |
@patch("synnefo.logic.backend", Mock()) |
231 | 198 |
def test_delete_network_with_floating_ips(self): |
232 |
ip = FloatingIPFactory(machine=None, network__flavor="IP_LESS_ROUTED") |
|
233 |
net = ip.network |
|
199 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, |
|
200 |
network=self.pool, nic=None) |
|
201 |
# Mark the network as non-pubic to not get 403 |
|
202 |
network = ip.network |
|
203 |
network.public = False |
|
204 |
network.save() |
|
234 | 205 |
# Can not remove network with floating IPs |
235 | 206 |
with mocked_quotaholder(): |
236 |
response = self.delete(NETWORKS_URL + "/%s" % net.id,
|
|
237 |
net.userid)
|
|
238 |
self.assertFault(response, 421, "networkInUse")
|
|
207 |
response = self.delete(NETWORKS_URL + "/%s" % self.pool.id,
|
|
208 |
self.pool.userid)
|
|
209 |
self.assertConflict(response)
|
|
239 | 210 |
# But we can with only deleted Floating Ips |
240 | 211 |
ip.deleted = True |
241 | 212 |
ip.save() |
242 | 213 |
with mocked_quotaholder(): |
243 |
response = self.delete(NETWORKS_URL + "/%s" % net.id,
|
|
244 |
net.userid)
|
|
214 |
response = self.delete(NETWORKS_URL + "/%s" % self.pool.id,
|
|
215 |
self.pool.userid)
|
|
245 | 216 |
self.assertSuccess(response) |
246 | 217 |
|
247 | 218 |
|
... | ... | |
255 | 226 |
self.assertEqual(json.loads(response.content)["floating_ip_pools"], []) |
256 | 227 |
|
257 | 228 |
def test_list_pools(self): |
258 |
net = FloatingIPPoolFactory(subnet="192.168.0.0/30", |
|
259 |
gateway="192.168.0.1") |
|
260 |
NetworkFactory(public=True, deleted=True) |
|
261 |
NetworkFactory(public=False, deleted=False) |
|
262 |
NetworkFactory(public=True, deleted=False) |
|
229 |
net = mf.NetworkWithSubnetFactory(floating_ip_pool=True, |
|
230 |
public=True, |
|
231 |
subnet__cidr="192.168.2.0/30", |
|
232 |
subnet__gateway="192.168.2.1") |
|
233 |
mf.NetworkWithSubnetFactory(public=True, deleted=True) |
|
234 |
mf.NetworkWithSubnetFactory(public=False, deleted=False) |
|
235 |
mf.NetworkWithSubnetFactory(public=True, floating_ip_pool=False) |
|
263 | 236 |
response = self.get(POOLS_URL) |
264 | 237 |
self.assertSuccess(response) |
265 | 238 |
self.assertEqual(json.loads(response.content)["floating_ip_pools"], |
... | ... | |
268 | 241 |
|
269 | 242 |
class FloatingIPActionsTest(BaseAPITest): |
270 | 243 |
def setUp(self): |
271 |
vm = VirtualMachineFactory() |
|
272 |
vm.operstate = "ACTIVE" |
|
273 |
vm.save() |
|
274 |
self.vm = vm |
|
244 |
self.vm = VirtualMachineFactory() |
|
245 |
self.vm.operstate = "ACTIVE" |
|
246 |
self.vm.save() |
|
275 | 247 |
|
276 | 248 |
def test_bad_request(self): |
277 | 249 |
url = SERVERS_URL + "/%s/action" % self.vm.id |
... | ... | |
290 | 262 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
291 | 263 |
self.assertItemNotFound(response) |
292 | 264 |
# In use |
293 |
vm1 = VirtualMachineFactory() |
|
294 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=vm1) |
|
295 |
BackendNetworkFactory(network=ip1.network, backend=vm1.backend, |
|
296 |
operstate='ACTIVE') |
|
297 |
request = {"addFloatingIp": {"address": ip1.ipv4}} |
|
265 |
ip = mf.IPv4AddressFactory(floating_ip=True, userid=self.vm.userid) |
|
266 |
request = {"addFloatingIp": {"address": ip.address}} |
|
298 | 267 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
299 |
self.assertFault(response, 409, "conflict")
|
|
268 |
self.assertConflict(response)
|
|
300 | 269 |
# Success |
301 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None) |
|
302 |
BackendNetworkFactory(network=ip1.network, backend=self.vm.backend, |
|
303 |
operstate='ACTIVE') |
|
304 |
request = {"addFloatingIp": {"address": ip1.ipv4}} |
|
270 |
ip = mf.IPv4AddressFactory(floating_ip=True, nic=None, |
|
271 |
userid=self.vm.userid) |
|
272 |
request = {"addFloatingIp": {"address": ip.address}} |
|
305 | 273 |
mock().ModifyInstance.return_value = 1 |
306 | 274 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
307 | 275 |
self.assertEqual(response.status_code, 202) |
308 |
ip1_after = floating_ips.get(id=ip1.id) |
|
309 |
self.assertEqual(ip1_after.machine, self.vm) |
|
310 |
self.assertTrue(ip1_after.in_use()) |
|
311 |
nic = self.vm.nics.get(ipv4=ip1_after.ipv4) |
|
276 |
ip_after = floating_ips.get(id=ip.id) |
|
277 |
self.assertEqual(ip_after.nic.machine, self.vm) |
|
278 |
nic = self.vm.nics.get() |
|
312 | 279 |
nic.state = "ACTIVE" |
313 | 280 |
nic.save() |
314 | 281 |
response = self.get(SERVERS_URL + "/%s" % self.vm.id, |
... | ... | |
323 | 290 |
url = SERVERS_URL + "/%s/action" % self.vm.id |
324 | 291 |
request = {"removeFloatingIp": {"address": "10.0.0.1"}} |
325 | 292 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
326 |
self.assertItemNotFound(response)
|
|
293 |
self.assertBadRequest(response)
|
|
327 | 294 |
# Not In Use |
328 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None) |
|
329 |
request = {"removeFloatingIp": {"address": ip1.ipv4}} |
|
295 |
ip = mf.IPv4AddressFactory(floating_ip=True, nic=None, |
|
296 |
userid=self.vm.userid) |
|
297 |
request = {"removeFloatingIp": {"address": ip.address}} |
|
330 | 298 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
331 |
self.assertItemNotFound(response)
|
|
299 |
self.assertBadRequest(response)
|
|
332 | 300 |
# Success |
333 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=self.vm)
|
|
334 |
NetworkInterfaceFactory(machine=self.vm, ipv4=ip1.ipv4)
|
|
335 |
request = {"removeFloatingIp": {"address": ip1.ipv4}}
|
|
301 |
ip = mf.IPv4AddressFactory(floating_ip=True,
|
|
302 |
userid=self.vm.userid, nic__machine=self.vm)
|
|
303 |
request = {"removeFloatingIp": {"address": ip.address}}
|
|
336 | 304 |
mock().ModifyInstance.return_value = 2 |
337 | 305 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
338 | 306 |
self.assertEqual(response.status_code, 202) |
339 | 307 |
# Yet used. Wait for the callbacks |
340 |
ip1_after = floating_ips.get(id=ip1.id) |
|
341 |
self.assertEqual(ip1_after.machine, self.vm) |
|
342 |
self.assertTrue(ip1_after.in_use()) |
|
308 |
ip_after = floating_ips.get(id=ip.id) |
|
309 |
self.assertEqual(ip_after.nic.machine, self.vm) |
Also available in: Unified diff