Revision e4f484da snf-cyclades-app/synnefo/api/tests/floating_ips.py
| b/snf-cyclades-app/synnefo/api/tests/floating_ips.py | ||
|---|---|---|
| 31 | 31 |
# interpreted as representing official policies, either expressed |
| 32 | 32 |
# or implied, of GRNET S.A. |
| 33 | 33 |
|
| 34 |
import json |
|
| 35 |
|
|
| 34 |
from django.utils import simplejson as json |
|
| 36 | 35 |
from snf_django.utils.testing import BaseAPITest, mocked_quotaholder |
| 37 | 36 |
from synnefo.db.models import IPAddress |
| 38 | 37 |
from synnefo.db import models_factory as mf |
| 39 | 38 |
from synnefo.db.models_factory import (NetworkFactory, |
| 40 | 39 |
VirtualMachineFactory) |
| 41 |
from mock import patch, Mock |
|
| 40 |
|
|
| 41 |
from mock import patch |
|
| 42 | 42 |
from functools import partial |
| 43 | 43 |
|
| 44 | 44 |
from synnefo.cyclades_settings import cyclades_services |
| ... | ... | |
| 47 | 47 |
|
| 48 | 48 |
|
| 49 | 49 |
compute_path = get_service_path(cyclades_services, "compute", version="v2.0") |
| 50 |
URL = join_urls(compute_path, "os-floating-ips")
|
|
| 50 |
URL = join_urls(compute_path, "floatingips")
|
|
| 51 | 51 |
NETWORKS_URL = join_urls(compute_path, "networks") |
| 52 | 52 |
SERVERS_URL = join_urls(compute_path, "servers") |
| 53 | 53 |
|
| ... | ... | |
| 77 | 77 |
api_ip = json.loads(response.content)["floating_ips"][0] |
| 78 | 78 |
self.assertEqual(api_ip, |
| 79 | 79 |
{"instance_id": str(ip.nic.machine_id),
|
| 80 |
"ip": ip.address,
|
|
| 81 |
"fixed_ip": None, |
|
| 80 |
"floating_ip_address": ip.address,
|
|
| 81 |
"fixed_ip_address": None,
|
|
| 82 | 82 |
"id": str(ip.id), |
| 83 |
"pool": str(ip.network_id)}) |
|
| 83 |
"port_id": str(ip.nic.id), |
|
| 84 |
"floating_network_id": str(ip.network_id)}) |
|
| 84 | 85 |
|
| 85 | 86 |
def test_get_ip(self): |
| 86 | 87 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
| ... | ... | |
| 90 | 91 |
api_ip = json.loads(response.content)["floating_ip"] |
| 91 | 92 |
self.assertEqual(api_ip, |
| 92 | 93 |
{"instance_id": str(ip.nic.machine_id),
|
| 93 |
"ip": ip.address,
|
|
| 94 |
"fixed_ip": None, |
|
| 94 |
"floating_ip_address": ip.address,
|
|
| 95 |
"fixed_ip_address": None,
|
|
| 95 | 96 |
"id": str(ip.id), |
| 96 |
"pool": str(ip.network_id)}) |
|
| 97 |
"port_id": str(ip.nic.id), |
|
| 98 |
"floating_network_id": str(ip.network_id)}) |
|
| 97 | 99 |
|
| 98 | 100 |
def test_wrong_user(self): |
| 99 | 101 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
| ... | ... | |
| 107 | 109 |
self.assertItemNotFound(response) |
| 108 | 110 |
|
| 109 | 111 |
def test_reserve(self): |
| 110 |
request = {'pool': self.pool.id}
|
|
| 112 |
request = {"floatingip": {
|
|
| 113 |
"floating_network_id": self.pool.id} |
|
| 114 |
} |
|
| 111 | 115 |
with mocked_quotaholder(): |
| 112 | 116 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 113 | 117 |
self.assertSuccess(response) |
| 118 |
api_ip = json.loads(response.content, encoding="utf-8")["floating_ip"] |
|
| 114 | 119 |
ip = floating_ips.get() |
| 115 | 120 |
self.assertEqual(ip.address, "192.168.2.2") |
| 116 | 121 |
self.assertEqual(ip.nic, None) |
| 117 | 122 |
self.assertEqual(ip.network, self.pool) |
| 118 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
|
| 119 |
{"instance_id": None, "ip": "192.168.2.2",
|
|
| 120 |
"fixed_ip": None, "id": str(ip.id), |
|
| 121 |
"pool": str(self.pool.id)}) |
|
| 123 |
self.assertEqual(api_ip, |
|
| 124 |
{"instance_id": None,
|
|
| 125 |
"floating_ip_address": "192.168.2.2", |
|
| 126 |
"fixed_ip_address": None, |
|
| 127 |
"id": str(ip.id), |
|
| 128 |
"port_id": None, |
|
| 129 |
"floating_network_id": str(self.pool.id)}) |
|
| 122 | 130 |
|
| 123 | 131 |
def test_reserve_no_pool(self): |
| 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') |
|
| 132 |
# Network is not a floating IP pool |
|
| 133 |
pool2 = mf.NetworkWithSubnetFactory(floating_ip_pool=False, |
|
| 134 |
public=True, |
|
| 135 |
subnet__cidr="192.168.2.0/24", |
|
| 136 |
subnet__gateway="192.168.2.1") |
|
| 137 |
request = {"floatingip": {
|
|
| 138 |
'floating_network_id': pool2.id} |
|
| 139 |
} |
|
| 140 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
|
| 141 |
self.assertEqual(response.status_code, 404) |
|
| 128 | 142 |
|
| 129 | 143 |
# Full network |
| 130 | 144 |
net = mf.NetworkWithSubnetFactory(floating_ip_pool=True, |
| ... | ... | |
| 132 | 146 |
subnet__cidr="192.168.2.0/31", |
| 133 | 147 |
subnet__gateway="192.168.2.1", |
| 134 | 148 |
subnet__pool__size=0) |
| 135 |
response = self.post(URL, "test_user", json.dumps({}), "json")
|
|
| 136 |
self.assertFault(response, 503, 'serviceUnavailable') |
|
| 137 |
|
|
| 138 |
request = {'pool': net.id}
|
|
| 149 |
request = {"floatingip": {
|
|
| 150 |
'floating_network_id': net.id} |
|
| 151 |
} |
|
| 139 | 152 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 140 | 153 |
self.assertConflict(response) |
| 141 | 154 |
|
| 142 | 155 |
def test_reserve_with_address(self): |
| 143 |
request = {'pool': self.pool.id, "address": "192.168.2.10"}
|
|
| 156 |
request = {"floatingip": {
|
|
| 157 |
"floating_network_id": self.pool.id, |
|
| 158 |
"floating_ip_address": "192.168.2.10"} |
|
| 159 |
} |
|
| 144 | 160 |
with mocked_quotaholder(): |
| 145 | 161 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 146 | 162 |
self.assertSuccess(response) |
| 147 | 163 |
ip = floating_ips.get() |
| 148 | 164 |
self.assertEqual(json.loads(response.content)["floating_ip"], |
| 149 |
{"instance_id": None, "ip": "192.168.2.10",
|
|
| 150 |
"fixed_ip": None, "id": str(ip.id), |
|
| 151 |
"pool": str(self.pool.id)}) |
|
| 165 |
{"instance_id": None,
|
|
| 166 |
"floating_ip_address": "192.168.2.10", |
|
| 167 |
"fixed_ip_address": None, |
|
| 168 |
"id": str(ip.id), |
|
| 169 |
"port_id": None, |
|
| 170 |
"floating_network_id": str(self.pool.id)}) |
|
| 152 | 171 |
|
| 153 | 172 |
# Already reserved |
| 154 | 173 |
with mocked_quotaholder(): |
| ... | ... | |
| 157 | 176 |
|
| 158 | 177 |
# Used by instance |
| 159 | 178 |
self.pool.reserve_address("192.168.2.20")
|
| 160 |
request = {'pool': self.pool.id, "address": "192.168.2.20"}
|
|
| 179 |
request = {"floatingip": {
|
|
| 180 |
"floating_network_id": self.pool.id, |
|
| 181 |
"floating_ip_address": "192.168.2.20"} |
|
| 182 |
} |
|
| 161 | 183 |
with mocked_quotaholder(): |
| 162 | 184 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 163 | 185 |
self.assertFault(response, 409, "conflict") |
| 164 | 186 |
|
| 165 | 187 |
# Address out of pool |
| 166 |
request = {'pool': self.pool.id, "address": "192.168.3.5"}
|
|
| 188 |
request = {"floatingip": {
|
|
| 189 |
"floating_network_id": self.pool.id, |
|
| 190 |
"floating_ip_address": "192.168.3.5"} |
|
| 191 |
} |
|
| 167 | 192 |
with mocked_quotaholder(): |
| 168 | 193 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
| 169 | 194 |
self.assertBadRequest(response) |
| 170 | 195 |
|
| 196 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 197 |
def test_reserve_and_connect(self, mrapi): |
|
| 198 |
vm = mf.VirtualMachineFactory(userid="test_user") |
|
| 199 |
request = {"floatingip": {
|
|
| 200 |
"floating_network_id": self.pool.id, |
|
| 201 |
"floating_ip_address": "192.168.2.12", |
|
| 202 |
"device_id": vm.id} |
|
| 203 |
} |
|
| 204 |
response = self.post(URL, "test_user", json.dumps(request), "json") |
|
| 205 |
ip = floating_ips.get() |
|
| 206 |
api_ip = json.loads(response.content, "utf-8")["floating_ip"] |
|
| 207 |
self.assertEqual(api_ip, |
|
| 208 |
{"instance_id": str(vm.id),
|
|
| 209 |
"floating_ip_address": "192.168.2.12", |
|
| 210 |
"fixed_ip_address": None, |
|
| 211 |
"id": str(ip.id), |
|
| 212 |
"port_id": str(vm.nics.all()[0].id), |
|
| 213 |
"floating_network_id": str(self.pool.id)}) |
|
| 214 |
|
|
| 215 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 216 |
def test_update_attach(self, mrapi): |
|
| 217 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None) |
|
| 218 |
vm = mf.VirtualMachineFactory(userid="user1") |
|
| 219 |
request = {"floatingip": {
|
|
| 220 |
"device_id": vm.id} |
|
| 221 |
} |
|
| 222 |
with mocked_quotaholder(): |
|
| 223 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 224 |
json.dumps(request), "json") |
|
| 225 |
self.assertEqual(response.status_code, 202) |
|
| 226 |
|
|
| 227 |
def test_update_attach_conflict(self): |
|
| 228 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
| 229 |
vm = mf.VirtualMachineFactory(userid="user1") |
|
| 230 |
request = {"floatingip": {
|
|
| 231 |
"device_id": vm.id} |
|
| 232 |
} |
|
| 233 |
with mocked_quotaholder(): |
|
| 234 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 235 |
json.dumps(request), "json") |
|
| 236 |
self.assertEqual(response.status_code, 409) |
|
| 237 |
|
|
| 238 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 239 |
def test_update_dettach(self, mrapi): |
|
| 240 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
|
| 241 |
request = {"floatingip": {
|
|
| 242 |
"device_id": None} |
|
| 243 |
} |
|
| 244 |
mrapi().ModifyInstance.return_value = 42 |
|
| 245 |
with mocked_quotaholder(): |
|
| 246 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 247 |
json.dumps(request), "json") |
|
| 248 |
self.assertEqual(response.status_code, 202) |
|
| 249 |
|
|
| 250 |
def test_update_dettach_unassociated(self): |
|
| 251 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, nic=None) |
|
| 252 |
request = {"floatingip": {}}
|
|
| 253 |
with mocked_quotaholder(): |
|
| 254 |
response = self.put(URL + "/%s" % ip.id, "user1", |
|
| 255 |
json.dumps(request), "json") |
|
| 256 |
self.assertEqual(response.status_code, 400) |
|
| 257 |
|
|
| 171 | 258 |
def test_release_in_use(self): |
| 172 | 259 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True) |
| 173 | 260 |
vm = ip.nic.machine |
| ... | ... | |
| 194 | 281 |
self.assertSuccess(response) |
| 195 | 282 |
ips_after = floating_ips.filter(id=ip.id) |
| 196 | 283 |
self.assertEqual(len(ips_after), 0) |
| 197 |
|
|
| 284 |
''' |
|
| 198 | 285 |
@patch("synnefo.logic.backend", Mock())
|
| 199 | 286 |
def test_delete_network_with_floating_ips(self): |
| 200 | 287 |
ip = mf.IPv4AddressFactory(userid="user1", floating_ip=True, |
| ... | ... | |
| 310 | 397 |
# Yet used. Wait for the callbacks |
| 311 | 398 |
ip_after = floating_ips.get(id=ip.id) |
| 312 | 399 |
self.assertEqual(ip_after.nic.machine, self.vm) |
| 400 |
''' |
|
Also available in: Unified diff