Revision b9364b15
| b/snf-cyclades-app/synnefo/api/tests/floating_ips.py | ||
|---|---|---|
| 38 | 38 |
from synnefo.db.models_factory import (NetworkFactory, |
| 39 | 39 |
VirtualMachineFactory) |
| 40 | 40 |
|
| 41 |
from mock import patch |
|
| 41 |
from mock import patch, Mock
|
|
| 42 | 42 |
from functools import partial |
| 43 | 43 |
|
| 44 | 44 |
from synnefo.cyclades_settings import cyclades_services |
| ... | ... | |
| 257 | 257 |
|
| 258 | 258 |
def test_release_in_use(self): |
| 259 | 259 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
| 260 |
vm = ip.nic.machine |
|
| 260 |
vm = mf.VirtualMachineFactory(userid="user1") |
|
| 261 |
request = {"floatingip": {
|
|
| 262 |
"device_id": vm.id} |
|
| 263 |
} |
|
| 261 | 264 |
with mocked_quotaholder(): |
| 262 |
response = self.delete(URL + "/%s" % ip.id, ip.userid) |
|
| 263 |
self.assertFault(response, 409, "conflict") |
|
| 264 |
# Also send a notification to remove the NIC and assert that FIP is in |
|
| 265 |
# use until notification from ganeti arrives |
|
| 266 |
request = {"removeFloatingIp": {"address": ip.address}}
|
|
| 267 |
url = SERVERS_URL + "/%s/action" % vm.id |
|
| 268 |
with patch('synnefo.logic.rapi_pool.GanetiRapiClient') as c:
|
|
| 269 |
c().ModifyInstance.return_value = 10 |
|
| 270 |
response = self.post(url, vm.userid, json.dumps(request), |
|
| 271 |
"json") |
|
| 265 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 266 |
json.dumps(request), "json") |
|
| 267 |
self.assertEqual(response.status_code, 409) |
|
| 268 |
|
|
| 269 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 270 |
def test_update_dettach(self, mrapi): |
|
| 271 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
| 272 |
request = {"floatingip": {
|
|
| 273 |
"device_id": None} |
|
| 274 |
} |
|
| 275 |
mrapi().ModifyInstance.return_value = 42 |
|
| 276 |
with mocked_quotaholder(): |
|
| 277 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 278 |
json.dumps(request), "json") |
|
| 279 |
print response.content |
|
| 272 | 280 |
self.assertEqual(response.status_code, 202) |
| 281 |
|
|
| 282 |
def test_update_dettach_unassociated(self): |
|
| 283 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None) |
|
| 284 |
request = {"floatingip": {}}
|
|
| 285 |
with mocked_quotaholder(): |
|
| 286 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 287 |
json.dumps(request), "json") |
|
| 288 |
self.assertEqual(response.status_code, 400) |
|
| 289 |
|
|
| 290 |
def test_release_in_use(self): |
|
| 291 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
| 292 |
vm = ip.nic.machine |
|
| 273 | 293 |
with mocked_quotaholder(): |
| 274 | 294 |
response = self.delete(URL + "/%s" % ip.id, ip.userid) |
| 275 | 295 |
self.assertFault(response, 409, "conflict") |
| ... | ... | |
| 281 | 301 |
self.assertSuccess(response) |
| 282 | 302 |
ips_after = floating_ips.filter(id=ip.id) |
| 283 | 303 |
self.assertEqual(len(ips_after), 0) |
| 284 |
''' |
|
| 304 |
|
|
| 285 | 305 |
@patch("synnefo.logic.backend", Mock())
|
| 286 | 306 |
def test_delete_network_with_floating_ips(self): |
| 287 | 307 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, |
| ... | ... | |
| 303 | 323 |
self.pool.userid) |
| 304 | 324 |
self.assertSuccess(response) |
| 305 | 325 |
|
| 306 |
|
|
| 326 |
''' |
|
| 307 | 327 |
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools") |
| 308 | 328 |
|
| 309 | 329 |
|
Also available in: Unified diff