Revision 2c2d4f0a
| b/snf-cyclades-app/synnefo/api/tests/ports.py | ||
|---|---|---|
| 31 | 31 |
# interpreted as representing official policies, either expressed |
| 32 | 32 |
# or implied, of GRNET S.A.i |
| 33 | 33 |
|
| 34 |
from snf_django.utils.testing import (BaseAPITest, override_settings) |
|
| 35 | 34 |
from snf_django.utils.testing import BaseAPITest |
| 36 | 35 |
from django.utils import simplejson as json |
| 37 | 36 |
from synnefo.cyclades_settings import cyclades_services |
| 38 | 37 |
from synnefo.lib.services import get_service_path |
| 39 | 38 |
from synnefo.lib import join_urls |
| 39 |
from mock import patch |
|
| 40 | 40 |
import synnefo.db.models_factory as dbmf |
| 41 | 41 |
|
| 42 | 42 |
COMPUTE_URL = get_service_path(cyclades_services, 'compute', |
| ... | ... | |
| 63 | 63 |
response = self.get(url, user=nic.network.userid) |
| 64 | 64 |
self.assertEqual(response.status_code, 200) |
| 65 | 65 |
|
| 66 |
def test_delete_port(self): |
|
| 66 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 67 |
def test_delete_port(self, mrapi): |
|
| 67 | 68 |
nic = dbmf.NetworkInterfaceFactory.create(device_owner='vm') |
| 68 | 69 |
url = join_urls(PORTS_URL, str(nic.id)) |
| 69 | 70 |
response = self.delete(url, user=nic.network.userid) |
| ... | ... | |
| 127 | 128 |
response = self.post(PORTS_URL, params=json.dumps(request)) |
| 128 | 129 |
self.assertEqual(response.status_code, 404) |
| 129 | 130 |
|
| 130 |
def test_create_port_private_net(self): |
|
| 131 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 132 |
def test_create_port_private_net(self, mrapi): |
|
| 131 | 133 |
net = dbmf.NetworkFactory.create(public=False) |
| 132 |
subnet1 = dbmf.IPv4SubnetFactory.create(network=net)
|
|
| 133 |
subnet2 = dbmf.IPv6SubnetFactory.create(network=net)
|
|
| 134 |
dbmf.IPv4SubnetFactory.create(network=net) |
|
| 135 |
dbmf.IPv6SubnetFactory.create(network=net) |
|
| 134 | 136 |
sg1 = dbmf.SecurityGroupFactory.create() |
| 135 | 137 |
sg2 = dbmf.SecurityGroupFactory.create() |
| 136 | 138 |
vm = dbmf.VirtualMachineFactory.create(userid=net.userid) |
| ... | ... | |
| 168 | 170 |
"name": "port1", |
| 169 | 171 |
"network_id": str(net.id), |
| 170 | 172 |
"device_id": str(vm.id), |
| 171 |
"fixed_ips" : [{"ip_address":"8.8.8.8"}]
|
|
| 173 |
"fixed_ips": [{"ip_address": "8.8.8.8"}]
|
|
| 172 | 174 |
} |
| 173 | 175 |
} |
| 174 | 176 |
response = self.post(PORTS_URL, params=json.dumps(request), |
| ... | ... | |
| 184 | 186 |
"name": "port1", |
| 185 | 187 |
"network_id": str(net.id), |
| 186 | 188 |
"device_id": str(vm.id), |
| 187 |
"fixed_ips" : [{"ip_address":fip.address}]
|
|
| 189 |
"fixed_ips": [{"ip_address": fip.address}]
|
|
| 188 | 190 |
} |
| 189 | 191 |
} |
| 190 | 192 |
response = self.post(PORTS_URL, params=json.dumps(request), |
| ... | ... | |
| 200 | 202 |
"name": "port1", |
| 201 | 203 |
"network_id": str(net.id), |
| 202 | 204 |
"device_id": str(vm.id), |
| 203 |
"fixed_ips" : [{"ip_address":fip.address}]
|
|
| 205 |
"fixed_ips": [{"ip_address": fip.address}]
|
|
| 204 | 206 |
} |
| 205 | 207 |
} |
| 206 | 208 |
response = self.post(PORTS_URL, params=json.dumps(request), |
| 207 | 209 |
user=net.userid) |
| 208 | 210 |
self.assertEqual(response.status_code, 400) |
| 209 | 211 |
|
| 210 |
def test_create_port_public_net(self): |
|
| 212 |
@patch("synnefo.db.models.get_rapi_client")
|
|
| 213 |
def test_create_port_public_net(self, mrapi): |
|
| 211 | 214 |
net = dbmf.NetworkFactory.create(public=True) |
| 212 | 215 |
fip = dbmf.FloatingIPFactory(network=net, nic=None, userid=net.userid) |
| 213 | 216 |
vm = dbmf.VirtualMachineFactory.create(userid=net.userid) |
| ... | ... | |
| 216 | 219 |
"name": "port1", |
| 217 | 220 |
"network_id": str(net.id), |
| 218 | 221 |
"device_id": str(vm.id), |
| 219 |
"fixed_ips" : [{"ip_address":fip.address}]
|
|
| 222 |
"fixed_ips": [{"ip_address": fip.address}]
|
|
| 220 | 223 |
} |
| 221 | 224 |
} |
| 222 | 225 |
response = self.post(PORTS_URL, params=json.dumps(request), |
| 223 | 226 |
user=net.userid) |
| 224 |
print response.content |
|
| 225 |
self.assertEqual(response.status_code, 200) |
|
| 227 |
self.assertEqual(response.status_code, 201) |
|
| 226 | 228 |
|
| 227 | 229 |
def test_add_nic_to_deleted_network(self): |
| 228 | 230 |
user = 'userr' |
| 229 | 231 |
vm = dbmf.VirtualMachineFactory(name='yo', userid=user, |
| 230 |
operstate="ACTIVE")
|
|
| 232 |
operstate="ACTIVE") |
|
| 231 | 233 |
net = dbmf.NetworkFactory(state='ACTIVE', userid=user, |
| 232 |
deleted=True)
|
|
| 234 |
deleted=True) |
|
| 233 | 235 |
request = {
|
| 234 | 236 |
"port": {
|
| 235 | 237 |
"device_id": str(vm.id), |
| ... | ... | |
| 254 | 256 |
} |
| 255 | 257 |
response = self.post(PORTS_URL, params=json.dumps(request), |
| 256 | 258 |
user=net.userid) |
| 257 |
self.assertFault(response, 403, 'forbidden') |
|
| 259 |
self.assertBadRequest(response) |
|
| 260 |
#self.assertFault(response, 403, 'forbidden') |
|
| 258 | 261 |
|
| 259 | 262 |
def test_add_nic_malformed_1(self): |
| 260 | 263 |
user = 'userr' |
| 261 |
vm = dbmf.VirtualMachineFactory(name='yo', userid=user)
|
|
| 264 |
dbmf.VirtualMachineFactory(name='yo', userid=user) |
|
| 262 | 265 |
net = dbmf.NetworkFactory(state='ACTIVE', userid=user) |
| 263 | 266 |
request = {
|
| 264 | 267 |
"port": {
|
Also available in: Unified diff