Revision 92d2d1ce 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 (IPAddressFactory, NetworkFactory,
|
|
| 38 |
from synnefo.db.models_factory import (FloatingIPFactory, 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 |
IPAddressPoolFactory = partial(NetworkFactory, public=True, deleted=False, |
|
| 55 |
|
|
| 56 |
floating_ips = IPAddress.objects.filter(floating_ip=True) |
|
| 57 |
FloatingIPPoolFactory = partial(NetworkFactory, public=True, deleted=False, |
|
| 56 | 58 |
floating_ip_pool=True) |
| 57 | 59 |
|
| 58 | 60 |
|
| 59 |
class IPAddressAPITest(BaseAPITest):
|
|
| 61 |
class FloatingIPAPITest(BaseAPITest):
|
|
| 60 | 62 |
def test_no_floating_ip(self): |
| 61 | 63 |
response = self.get(URL) |
| 62 | 64 |
self.assertSuccess(response) |
| 63 | 65 |
self.assertEqual(json.loads(response.content)["floating_ips"], []) |
| 64 | 66 |
|
| 65 | 67 |
def test_list_ips(self): |
| 66 |
ip = IPAddressFactory(userid="user1")
|
|
| 67 |
IPAddressFactory(userid="user1", deleted=True)
|
|
| 68 |
ip = FloatingIPFactory(userid="user1")
|
|
| 69 |
FloatingIPFactory(userid="user1", deleted=True)
|
|
| 68 | 70 |
with mocked_quotaholder(): |
| 69 | 71 |
response = self.get(URL, "user1") |
| 70 | 72 |
self.assertSuccess(response) |
| ... | ... | |
| 75 | 77 |
str(ip.network.id)}) |
| 76 | 78 |
|
| 77 | 79 |
def test_get_ip(self): |
| 78 |
ip = IPAddressFactory(userid="user1")
|
|
| 80 |
ip = FloatingIPFactory(userid="user1")
|
|
| 79 | 81 |
with mocked_quotaholder(): |
| 80 | 82 |
response = self.get(URL + "/%s" % ip.id, "user1") |
| 81 | 83 |
self.assertSuccess(response) |
| ... | ... | |
| 86 | 88 |
str(ip.network.id)}) |
| 87 | 89 |
|
| 88 | 90 |
def test_wrong_user(self): |
| 89 |
ip = IPAddressFactory(userid="user1")
|
|
| 91 |
ip = FloatingIPFactory(userid="user1")
|
|
| 90 | 92 |
with mocked_quotaholder(): |
| 91 | 93 |
response = self.delete(URL + "/%s" % ip.id, "user2") |
| 92 | 94 |
self.assertItemNotFound(response) |
| 93 | 95 |
|
| 94 | 96 |
def test_deleted_ip(self): |
| 95 |
ip = IPAddressFactory(userid="user1", deleted=True)
|
|
| 97 |
ip = FloatingIPFactory(userid="user1", deleted=True)
|
|
| 96 | 98 |
with mocked_quotaholder(): |
| 97 | 99 |
response = self.delete(URL + "/%s" % ip.id, "user1") |
| 98 | 100 |
self.assertItemNotFound(response) |
| 99 | 101 |
|
| 100 | 102 |
def test_reserve(self): |
| 101 |
net = IPAddressPoolFactory(userid="test_user",
|
|
| 103 |
net = FloatingIPPoolFactory(userid="test_user",
|
|
| 102 | 104 |
subnet="192.168.2.0/24", |
| 103 | 105 |
gateway=None) |
| 104 | 106 |
request = {'pool': net.id}
|
| 105 | 107 |
with mocked_quotaholder(): |
| 106 | 108 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 107 | 109 |
self.assertSuccess(response) |
| 108 |
ip = IPAddress.objects.get()
|
|
| 110 |
ip = floating_ips.get()
|
|
| 109 | 111 |
self.assertEqual(ip.ipv4, "192.168.2.1") |
| 110 | 112 |
self.assertEqual(ip.machine, None) |
| 111 | 113 |
self.assertEqual(ip.network, net) |
| ... | ... | |
| 120 | 122 |
response = self.post(URL, "test_user", json.dumps({}), "json")
|
| 121 | 123 |
self.assertFault(response, 413, "overLimit") |
| 122 | 124 |
# Full network |
| 123 |
IPAddressPoolFactory(userid="test_user",
|
|
| 125 |
FloatingIPPoolFactory(userid="test_user",
|
|
| 124 | 126 |
subnet="192.168.2.0/32", |
| 125 | 127 |
gateway=None) |
| 126 | 128 |
with mocked_quotaholder(): |
| 127 | 129 |
response = self.post(URL, "test_user", json.dumps({}), "json")
|
| 128 | 130 |
self.assertFault(response, 413, "overLimit") |
| 129 | 131 |
# Success |
| 130 |
net2 = IPAddressPoolFactory(userid="test_user",
|
|
| 132 |
net2 = FloatingIPPoolFactory(userid="test_user",
|
|
| 131 | 133 |
subnet="192.168.2.0/24", |
| 132 | 134 |
gateway=None) |
| 133 | 135 |
with mocked_quotaholder(): |
| 134 | 136 |
response = self.post(URL, "test_user", json.dumps({}), "json")
|
| 135 | 137 |
self.assertSuccess(response) |
| 136 |
ip = IPAddress.objects.get()
|
|
| 138 |
ip = floating_ips.get()
|
|
| 137 | 139 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
| 138 | 140 |
{"instance_id": None, "ip": "192.168.2.1",
|
| 139 | 141 |
"fixed_ip": None, "id": str(ip.id), |
| 140 | 142 |
"pool": str(net2.id)}) |
| 141 | 143 |
|
| 142 | 144 |
def test_reserve_full(self): |
| 143 |
net = IPAddressPoolFactory(userid="test_user",
|
|
| 145 |
net = FloatingIPPoolFactory(userid="test_user",
|
|
| 144 | 146 |
subnet="192.168.2.0/32") |
| 145 | 147 |
request = {'pool': net.id}
|
| 146 | 148 |
with mocked_quotaholder(): |
| ... | ... | |
| 148 | 150 |
self.assertEqual(response.status_code, 413) |
| 149 | 151 |
|
| 150 | 152 |
def test_reserve_with_address(self): |
| 151 |
net = IPAddressPoolFactory(userid="test_user",
|
|
| 153 |
net = FloatingIPPoolFactory(userid="test_user",
|
|
| 152 | 154 |
subnet="192.168.2.0/24") |
| 153 | 155 |
request = {'pool': net.id, "address": "192.168.2.10"}
|
| 154 | 156 |
with mocked_quotaholder(): |
| 155 | 157 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 156 | 158 |
self.assertSuccess(response) |
| 157 |
ip = IPAddress.objects.get()
|
|
| 159 |
ip = floating_ips.get()
|
|
| 158 | 160 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
| 159 | 161 |
{"instance_id": None, "ip": "192.168.2.10",
|
| 160 | 162 |
"fixed_ip": None, "id": str(ip.id), "pool": |
| 161 | 163 |
str(net.id)}) |
| 162 | 164 |
|
| 163 | 165 |
# Already reserved |
| 164 |
IPAddressFactory(network=net, ipv4="192.168.2.3")
|
|
| 166 |
FloatingIPFactory(network=net, ipv4="192.168.2.3")
|
|
| 165 | 167 |
request = {'pool': net.id, "address": "192.168.2.3"}
|
| 166 | 168 |
with mocked_quotaholder(): |
| 167 | 169 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| ... | ... | |
| 194 | 196 |
self.assertBadRequest(response) |
| 195 | 197 |
|
| 196 | 198 |
def test_release_in_use(self): |
| 197 |
ip = IPAddressFactory()
|
|
| 199 |
ip = FloatingIPFactory()
|
|
| 198 | 200 |
vm = ip.machine |
| 199 | 201 |
vm.operstate = "ACTIVE" |
| 200 | 202 |
vm.userid = ip.userid |
| ... | ... | |
| 218 | 220 |
self.assertFault(response, 409, "conflict") |
| 219 | 221 |
|
| 220 | 222 |
def test_release(self): |
| 221 |
ip = IPAddressFactory(machine=None)
|
|
| 223 |
ip = FloatingIPFactory(machine=None)
|
|
| 222 | 224 |
with mocked_quotaholder(): |
| 223 | 225 |
response = self.delete(URL + "/%s" % ip.id, ip.userid) |
| 224 | 226 |
self.assertSuccess(response) |
| 225 |
ips_after = IPAddress.objects.filter(id=ip.id)
|
|
| 227 |
ips_after = floating_ips.filter(id=ip.id)
|
|
| 226 | 228 |
self.assertEqual(len(ips_after), 0) |
| 227 | 229 |
|
| 228 | 230 |
@patch("synnefo.logic.backend", Mock())
|
| 229 | 231 |
def test_delete_network_with_floating_ips(self): |
| 230 |
ip = IPAddressFactory(machine=None, network__flavor="IP_LESS_ROUTED")
|
|
| 232 |
ip = FloatingIPFactory(machine=None, network__flavor="IP_LESS_ROUTED")
|
|
| 231 | 233 |
net = ip.network |
| 232 | 234 |
# Can not remove network with floating IPs |
| 233 | 235 |
with mocked_quotaholder(): |
| ... | ... | |
| 246 | 248 |
POOLS_URL = join_urls(compute_path, "os-floating-ip-pools") |
| 247 | 249 |
|
| 248 | 250 |
|
| 249 |
class IPAddressPoolsAPITest(BaseAPITest):
|
|
| 251 |
class FloatingIPPoolsAPITest(BaseAPITest):
|
|
| 250 | 252 |
def test_no_pool(self): |
| 251 | 253 |
response = self.get(POOLS_URL) |
| 252 | 254 |
self.assertSuccess(response) |
| 253 | 255 |
self.assertEqual(json.loads(response.content)["floating_ip_pools"], []) |
| 254 | 256 |
|
| 255 | 257 |
def test_list_pools(self): |
| 256 |
net = IPAddressPoolFactory(subnet="192.168.0.0/30",
|
|
| 258 |
net = FloatingIPPoolFactory(subnet="192.168.0.0/30",
|
|
| 257 | 259 |
gateway="192.168.0.1") |
| 258 | 260 |
NetworkFactory(public=True, deleted=True) |
| 259 | 261 |
NetworkFactory(public=False, deleted=False) |
| ... | ... | |
| 264 | 266 |
[{"name": str(net.id), "size": 4, "free": 1}])
|
| 265 | 267 |
|
| 266 | 268 |
|
| 267 |
class IPAddressActionsTest(BaseAPITest):
|
|
| 269 |
class FloatingIPActionsTest(BaseAPITest):
|
|
| 268 | 270 |
def setUp(self): |
| 269 | 271 |
vm = VirtualMachineFactory() |
| 270 | 272 |
vm.operstate = "ACTIVE" |
| ... | ... | |
| 289 | 291 |
self.assertItemNotFound(response) |
| 290 | 292 |
# In use |
| 291 | 293 |
vm1 = VirtualMachineFactory() |
| 292 |
ip1 = IPAddressFactory(userid=self.vm.userid, machine=vm1)
|
|
| 294 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=vm1)
|
|
| 293 | 295 |
BackendNetworkFactory(network=ip1.network, backend=vm1.backend, |
| 294 | 296 |
operstate='ACTIVE') |
| 295 | 297 |
request = {"addFloatingIp": {"address": ip1.ipv4}}
|
| 296 | 298 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
| 297 | 299 |
self.assertFault(response, 409, "conflict") |
| 298 | 300 |
# Success |
| 299 |
ip1 = IPAddressFactory(userid=self.vm.userid, machine=None)
|
|
| 301 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None)
|
|
| 300 | 302 |
BackendNetworkFactory(network=ip1.network, backend=self.vm.backend, |
| 301 | 303 |
operstate='ACTIVE') |
| 302 | 304 |
request = {"addFloatingIp": {"address": ip1.ipv4}}
|
| 303 | 305 |
mock().ModifyInstance.return_value = 1 |
| 304 | 306 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
| 305 | 307 |
self.assertEqual(response.status_code, 202) |
| 306 |
ip1_after = IPAddress.objects.get(id=ip1.id)
|
|
| 308 |
ip1_after = floating_ips.get(id=ip1.id)
|
|
| 307 | 309 |
self.assertEqual(ip1_after.machine, self.vm) |
| 308 | 310 |
self.assertTrue(ip1_after.in_use()) |
| 309 | 311 |
nic = self.vm.nics.get(ipv4=ip1_after.ipv4) |
| ... | ... | |
| 323 | 325 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
| 324 | 326 |
self.assertItemNotFound(response) |
| 325 | 327 |
# Not In Use |
| 326 |
ip1 = IPAddressFactory(userid=self.vm.userid, machine=None)
|
|
| 328 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=None)
|
|
| 327 | 329 |
request = {"removeFloatingIp": {"address": ip1.ipv4}}
|
| 328 | 330 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
| 329 | 331 |
self.assertItemNotFound(response) |
| 330 | 332 |
# Success |
| 331 |
ip1 = IPAddressFactory(userid=self.vm.userid, machine=self.vm)
|
|
| 333 |
ip1 = FloatingIPFactory(userid=self.vm.userid, machine=self.vm)
|
|
| 332 | 334 |
NetworkInterfaceFactory(machine=self.vm, ipv4=ip1.ipv4) |
| 333 | 335 |
request = {"removeFloatingIp": {"address": ip1.ipv4}}
|
| 334 | 336 |
mock().ModifyInstance.return_value = 2 |
| 335 | 337 |
response = self.post(url, self.vm.userid, json.dumps(request), "json") |
| 336 | 338 |
self.assertEqual(response.status_code, 202) |
| 337 | 339 |
# Yet used. Wait for the callbacks |
| 338 |
ip1_after = IPAddress.objects.get(id=ip1.id)
|
|
| 340 |
ip1_after = floating_ips.get(id=ip1.id)
|
|
| 339 | 341 |
self.assertEqual(ip1_after.machine, self.vm) |
| 340 | 342 |
self.assertTrue(ip1_after.in_use()) |
Also available in: Unified diff