Modify structure of test to separe functional tests
We move and rename the functional tests related to networks. And we remove the unit tests of middleware. Change-Id: Ic82e44d656121f9f87828b58e5c547669d70ea37
This commit is contained in:
jorgesece
parent
599fa241ac
commit
d7b13e354d
|
|
@ -17,15 +17,9 @@
|
|||
|
||||
import uuid
|
||||
|
||||
import mock
|
||||
|
||||
from ooi.api import network_link
|
||||
from ooi import exception
|
||||
from ooi.occi.core import collection
|
||||
from ooi.tests import fakes_network as fakes
|
||||
from ooi.tests import fakes
|
||||
from ooi.tests.functional.middleware import test_middleware
|
||||
from ooi import utils
|
||||
from ooi import wsgi
|
||||
|
||||
|
||||
class TestNetInterfaceController(test_middleware.TestMiddleware):
|
||||
|
|
@ -34,14 +28,11 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
def setUp(self):
|
||||
super(TestNetInterfaceController, self).setUp()
|
||||
self.application_url = fakes.application_url
|
||||
self.app = wsgi.OCCIMiddleware(None)
|
||||
self.app = self.get_app()
|
||||
|
||||
@mock.patch.object(network_link.Controller, "index")
|
||||
def test_list_ifaces_empty(self, mock_index):
|
||||
def test_list_ifaces_empty(self):
|
||||
tenant = fakes.tenants["bar"]
|
||||
mock_index.return_value = fakes.fake_network_link_occi(
|
||||
fakes.network_links[tenant['id']]
|
||||
)
|
||||
|
||||
for url in ("/networklink/", "/networklink"):
|
||||
req = self._build_req(url, tenant["id"], method="GET")
|
||||
|
||||
|
|
@ -54,14 +45,8 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
self.assertExpectedResult(expected_result, resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "index")
|
||||
def test_list_ifaces(self, mock_index):
|
||||
tenant = fakes.tenants["foo"]
|
||||
mock_index.return_value = collection.Collection(
|
||||
fakes.fake_network_link_occi(
|
||||
fakes.network_links[tenant['id']]
|
||||
)
|
||||
)
|
||||
def test_list_ifaces(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
|
||||
for url in ("/networklink/", "/networklink"):
|
||||
req = self._build_req(url, tenant["id"], method="GET")
|
||||
|
|
@ -70,62 +55,67 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = []
|
||||
for ip in fakes.network_links[tenant["id"]]:
|
||||
if ip["instance_id"]:
|
||||
# fixme(jorgesece): test in case of instance None
|
||||
link_id = '_'.join([ip["instance_id"],
|
||||
ip["network_id"],
|
||||
ip["ip"]])
|
||||
float_list = {}
|
||||
for floating_ip in fakes.floating_ips[tenant["id"]]:
|
||||
if floating_ip["instance_id"]:
|
||||
float_list.update({floating_ip['fixed_ip']: floating_ip})
|
||||
instance_vm = fakes.linked_vm_id
|
||||
for p in fakes.ports[tenant["id"]]:
|
||||
for ip in p["fixed_ips"]:
|
||||
link_id = '_'.join([instance_vm,
|
||||
p["net_id"],
|
||||
ip["ip_address"]])
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"networklink/%s" % link_id))
|
||||
)
|
||||
float_ip = float_list.get(ip['ip_address'], None)
|
||||
if float_ip:
|
||||
link_id = '_'.join([instance_vm,
|
||||
"PUBLIC",
|
||||
float_ip["ip"]])
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"networklink/%s" % link_id))
|
||||
)
|
||||
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "show")
|
||||
def test_show_iface(self, m_show):
|
||||
tenant = fakes.tenants["foo"]
|
||||
m_show.return_value = fakes.fake_network_link_occi(
|
||||
fakes.network_links[tenant['id']]
|
||||
)
|
||||
|
||||
for ip in fakes.network_links[tenant["id"]]:
|
||||
if ip["instance_id"] is not None:
|
||||
link_id = '_'.join([ip["instance_id"],
|
||||
ip["network_id"],
|
||||
ip["ip"]]
|
||||
def test_show_iface(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
instance_vm = fakes.linked_vm_id
|
||||
for p in fakes.ports[tenant["id"]]:
|
||||
for ip in p["fixed_ips"]:
|
||||
link_id = '_'.join([instance_vm,
|
||||
p["net_id"],
|
||||
ip["ip_address"]]
|
||||
)
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="GET")
|
||||
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
# net_id = uuid.uuid4().hex
|
||||
|
||||
source = utils.join_url(self.application_url + "/",
|
||||
"compute/%s" % ip["instance_id"])
|
||||
"compute/%s" % instance_vm)
|
||||
target = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % ip['network_id'])
|
||||
"network/%s" % p["net_id"])
|
||||
self.assertResultIncludesLinkAttr(link_id, source, target,
|
||||
resp)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "show")
|
||||
def test_show_invalid_id(self, m_show):
|
||||
def test_show_invalid_id(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
link_id = uuid.uuid4().hex
|
||||
m_show.side_effect = exception.LinkNotFound(link_id=link_id)
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(404, resp.status_code)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "create")
|
||||
def test_create_link_invalid(self, m_create):
|
||||
def test_create_link_invalid(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
m_create.side_effect = exception.Invalid
|
||||
net_id = fakes.network_links[tenant['id']][0]['network_id']
|
||||
net_id = fakes.ports[tenant['id']][0]['net_id']
|
||||
occi_net_id = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id)
|
||||
headers = {
|
||||
|
|
@ -133,23 +123,21 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind"'),
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="foo", '
|
||||
'occi.core.target="%s"'
|
||||
) % occi_net_id
|
||||
'X-OCCI-Attribute': ('occi.core.source="foo", '
|
||||
'occi.core.target="%s"'
|
||||
) % occi_net_id
|
||||
}
|
||||
req = self._build_req("/networklink", None, method="POST",
|
||||
headers=headers)
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(400, resp.status_code)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "create")
|
||||
def test_create_link_no_pool(self, m_create):
|
||||
def test_create_link_no_pool(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
m_create.return_value = fakes.fake_network_link_occi(
|
||||
fakes.network_links[tenant['id']]
|
||||
)[0]
|
||||
net_id = fakes.network_links[tenant['id']][0]['network_id']
|
||||
net_id = fakes.ports[tenant['id']][0]['net_id']
|
||||
occi_compute_id = utils.join_url(
|
||||
self.application_url + "/",
|
||||
"compute/%s" % fakes.linked_vm_id)
|
||||
occi_net_id = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id)
|
||||
headers = {
|
||||
|
|
@ -157,50 +145,44 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind"'),
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="foo", '
|
||||
'occi.core.target="%s"'
|
||||
) % occi_net_id
|
||||
'X-OCCI-Attribute': ('occi.core.source="%s", '
|
||||
'occi.core.target="%s"'
|
||||
) % (occi_compute_id, occi_net_id)
|
||||
}
|
||||
req = self._build_req("/networklink", None, method="POST",
|
||||
req = self._build_req("/networklink", tenant["id"], method="POST",
|
||||
headers=headers)
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "create")
|
||||
def test_create_link_with_pool(self, m_create):
|
||||
tenant = fakes.tenants["foo"]
|
||||
m_create.return_value = collection.Collection(
|
||||
[fakes.fake_network_link_occi(
|
||||
fakes.network_links[tenant['id']]
|
||||
)[0]])
|
||||
link_info = fakes.network_links[tenant['id']][0]
|
||||
def test_create_link_with_pool(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
link_info = fakes.ports[tenant['id']][0]
|
||||
|
||||
server_url = utils.join_url(self.application_url + "/",
|
||||
"compute/%s" % link_info['instance_id'])
|
||||
"compute/%s" % link_info['server_id'])
|
||||
net_url = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % link_info['network_id'])
|
||||
"network/%s" % link_info['net_id'])
|
||||
pool_name = 'pool'
|
||||
headers = {
|
||||
'Category': (
|
||||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind",'
|
||||
'%s;'
|
||||
'scheme="http://schemas.openstack.org/network/'
|
||||
'floatingippool#"; class="mixin"') % pool_name,
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="%s", '
|
||||
'occi.core.target="%s"'
|
||||
) % (server_url, net_url)
|
||||
'Category': ('networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/'
|
||||
'infrastructure#";'
|
||||
'class="kind",'
|
||||
'%s;'
|
||||
'scheme="http://schemas.openstack.org/network/'
|
||||
'floatingippool#"; class="mixin"') % pool_name,
|
||||
'X-OCCI-Attribute': ('occi.core.source="%s", '
|
||||
'occi.core.target="%s"'
|
||||
) % (server_url, net_url)
|
||||
}
|
||||
req = self._build_req("/networklink", tenant["id"], method="POST",
|
||||
headers=headers)
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
link_id = '_'.join([link_info['instance_id'],
|
||||
link_info['network_id'],
|
||||
link_info['ip']])
|
||||
link_id = '_'.join([link_info['server_id'],
|
||||
link_info['net_id'],
|
||||
link_info['fixed_ips'][0]
|
||||
["ip_address"]])
|
||||
expected = [("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"networklink/%s" % link_id))]
|
||||
|
|
@ -208,39 +190,35 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
self.assertExpectedResult(expected, resp)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "delete")
|
||||
def test_delete_fixed(self, m_delete):
|
||||
tenant = fakes.tenants["foo"]
|
||||
m_delete.return_value = []
|
||||
def test_delete_fixed(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
|
||||
for n in fakes.network_links[tenant["id"]]:
|
||||
if n["network_id"] != "PUBLIC":
|
||||
if n["instance_id"]:
|
||||
link_id = '_'.join([n["instance_id"],
|
||||
n["network_id"],
|
||||
n["ip"]])
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="DELETE")
|
||||
for n in fakes.ports[tenant["id"]]:
|
||||
if n["net_id"] != "PUBLIC":
|
||||
if n["server_id"]:
|
||||
link_id = '_'.join([n["server_id"],
|
||||
n["net_id"],
|
||||
n["fixed_ips"]
|
||||
[0]["ip_address"]])
|
||||
req = self._build_req(
|
||||
"/networklink/%s" % link_id,
|
||||
tenant["id"], method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
|
||||
@mock.patch.object(network_link.Controller, "delete")
|
||||
def test_delete_public(self, m_delete):
|
||||
tenant = fakes.tenants["public"]
|
||||
m_delete.return_value = []
|
||||
|
||||
for n in fakes.network_links[tenant["id"]]:
|
||||
if n["network_id"] != "PUBLIC":
|
||||
if n["instance_id"]:
|
||||
link_id = '_'.join([n["instance_id"],
|
||||
n["network_id"],
|
||||
n["ip"]])
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
def test_delete_public(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
for n in fakes.floating_ips[tenant["id"]]:
|
||||
if n["instance_id"]:
|
||||
link_id = '_'.join([n["instance_id"],
|
||||
"PUBLIC",
|
||||
n["ip"]])
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
|
||||
def test_show_non_existant_compute(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
|
|
@ -260,10 +238,9 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind"'),
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="foo", '
|
||||
'occi.core.target="%s"'
|
||||
) % net_id
|
||||
'X-OCCI-Attribute': ('occi.core.source="foo", '
|
||||
'occi.core.target="%s"'
|
||||
) % net_id
|
||||
}
|
||||
req = self._build_req("/networklink", None, method="POST",
|
||||
headers=headers)
|
||||
|
|
@ -279,10 +256,9 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
|
|||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind"'),
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="%s", '
|
||||
'occi.core.target="bar"'
|
||||
) % server_id
|
||||
'X-OCCI-Attribute': ('occi.core.source="%s", '
|
||||
'occi.core.target="bar"'
|
||||
) % server_id
|
||||
}
|
||||
req = self._build_req("/networklink", None, method="POST",
|
||||
headers=headers)
|
||||
|
|
|
|||
|
|
@ -17,37 +17,37 @@
|
|||
|
||||
import mock
|
||||
|
||||
from ooi.api import helpers_neutron
|
||||
from ooi.api import network
|
||||
from ooi.occi.core import collection
|
||||
import uuid
|
||||
import warnings
|
||||
|
||||
from oslo_config import cfg
|
||||
import webob
|
||||
|
||||
from ooi.api import helpers
|
||||
from ooi.tests import fakes_network as fakes
|
||||
from ooi.tests.functional.middleware import test_middleware
|
||||
from ooi import utils
|
||||
from ooi import wsgi
|
||||
|
||||
|
||||
def create_occi_results(data):
|
||||
return network.Controller(None)._get_network_resources(data)
|
||||
class TestNetNeutronController(test_middleware.TestMiddleware):
|
||||
"""Test OCCI network controller."""
|
||||
|
||||
|
||||
class TestMiddlewareNeutron(test_middleware.TestMiddleware):
|
||||
"""OCCI middleware test for Neutron middleware.
|
||||
|
||||
According to the OCCI HTTP rendering, no Accept header
|
||||
means text/plain.
|
||||
"""
|
||||
def setUp(self):
|
||||
super(TestMiddlewareNeutron, self).setUp()
|
||||
super(TestNetNeutronController, self).setUp()
|
||||
self.schema = 'http://schemas.ogf.org/occi/infrastructure#network'
|
||||
self.accept = self.content_type = None
|
||||
self.application_url = fakes.application_url
|
||||
self.app = wsgi.OCCIMiddleware(None)
|
||||
neutron_ooi_endpoint = "foo"
|
||||
|
||||
def mock_endpoint(self, bar):
|
||||
if bar == "neutron_ooi_endpoint":
|
||||
return neutron_ooi_endpoint
|
||||
|
||||
class TestNetworkController(TestMiddlewareNeutron):
|
||||
"""Test OCCI compute controller."""
|
||||
|
||||
def setUp(self):
|
||||
super(TestNetworkController, self).setUp()
|
||||
with mock.patch.object(cfg.ConfigOpts, "__getattr__",
|
||||
side_effect=mock_endpoint,
|
||||
autospec=True):
|
||||
self.app = wsgi.OCCIMiddleware(None)
|
||||
|
||||
def assertExpectedResult(self, expected, result):
|
||||
expected = ["%s: %s" % e for e in expected]
|
||||
|
|
@ -55,34 +55,43 @@ class TestNetworkController(TestMiddlewareNeutron):
|
|||
results = str(result.text).splitlines()
|
||||
self.assertItemsEqual(expected, results)
|
||||
|
||||
@mock.patch.object(network.Controller, "index")
|
||||
def test_deprecated_configuration(self):
|
||||
with warnings.catch_warnings(record=True) as w:
|
||||
neutron_endopoint = "/foo"
|
||||
warnings.simplefilter("always", DeprecationWarning)
|
||||
wsgi.OCCIMiddleware(None,
|
||||
neutron_ooi_endpoint=neutron_endopoint)
|
||||
expected_message = (
|
||||
"Configuration of neutron_ooi_endpoint"
|
||||
" in api-paste.ini file is deprecated,"
|
||||
" include it in nova.conf")
|
||||
self.assertEqual(1, len(w))
|
||||
self.assertIs(DeprecationWarning,
|
||||
w[-1].category)
|
||||
self.assertEqual(expected_message, w[-1].message.message)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks_empty(self, m):
|
||||
tenant = fakes.tenants["bar"]
|
||||
headers = {
|
||||
'Category': 'network; scheme="http://schema#";class="kind";',
|
||||
'X_OCCI_Attribute': 'project=%s' % tenant["id"],
|
||||
}
|
||||
url = "/network"
|
||||
req = self._build_req(path=url,
|
||||
tenant_id='X',
|
||||
method="GET",
|
||||
headers=headers, content_type="text/occi")
|
||||
m.return_value = collection.Collection(
|
||||
create_occi_results(fakes.networks[tenant['id']]))
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(204, resp.status_code)
|
||||
expected_result = ""
|
||||
self.assertExpectedResult(expected_result, resp)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
@mock.patch.object(network.Controller, "index")
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
ooi_net = helpers_neutron.OpenStackNeutron._build_networks(
|
||||
fakes.networks[tenant['id']]
|
||||
)
|
||||
m.return_value = collection.Collection(
|
||||
create_occi_results(ooi_net))
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
|
@ -98,38 +107,80 @@ class TestNetworkController(TestMiddlewareNeutron):
|
|||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(network.Controller, "create")
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_create(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": fakes.networks[tenant['id']][0]}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
subnet_out = fakes.create_fake_json_resp(
|
||||
{"subnet": fakes.networks[tenant['id']][0]["subnet_info"]},
|
||||
200)
|
||||
mock_subnet = mock.Mock(webob.Request)
|
||||
mock_subnet.get_response.return_value = subnet_out
|
||||
public_out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks[tenant['id']]},
|
||||
200)
|
||||
|
||||
mock_public = mock.Mock(webob.Request)
|
||||
mock_public.get_response.return_value = public_out
|
||||
router_out = fakes.create_fake_json_resp(
|
||||
{"router": {"id": uuid.uuid4().hex}},
|
||||
200)
|
||||
mock_router = mock.Mock(webob.Request)
|
||||
mock_router.get_response.return_value = router_out
|
||||
mock_iface = mock.Mock(webob.Request)
|
||||
mock_iface.get_response.return_value = fakes.create_fake_json_resp(
|
||||
{"foo": "foo"}, 200)
|
||||
m.side_effect = [mock_net, mock_subnet, mock_public,
|
||||
mock_router, mock_iface
|
||||
]
|
||||
name = fakes.networks[tenant["id"]][0]["name"]
|
||||
net_id = fakes.networks[tenant["id"]][0]["id"]
|
||||
address = fakes.networks[tenant["id"]][0]["subnet_info"]["cidr"]
|
||||
headers = {
|
||||
'Category': 'network; scheme="http://schema#";class="kind",' +
|
||||
'mixinID;'
|
||||
'scheme="http://schemas.openstack.org/template/os#";'
|
||||
' class=mixin',
|
||||
'X_Occi_Attribute': 'project=%s' % tenant["id"],
|
||||
'Category': 'network;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind",'
|
||||
'ipnetwork;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure/'
|
||||
'network#";'
|
||||
'class="mixin",',
|
||||
'X-OCCI-Attribute': '"occi.core.title"="%s",'
|
||||
'"occi.network.address"="%s"' %
|
||||
(name, address)
|
||||
}
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X',
|
||||
method="POST",
|
||||
headers=headers)
|
||||
fake_net = fakes.fake_network_occi(
|
||||
fakes.networks[tenant['id']]
|
||||
)[0]
|
||||
m.return_value = collection.Collection([fake_net])
|
||||
|
||||
m.return_value = fakes.networks[tenant['id']][0]
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = [("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % fake_net.id))]
|
||||
"network/%s" % net_id))]
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(network.Controller, "show")
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_show_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
|
||||
for n in fakes.networks[tenant["id"]]:
|
||||
ooi_net = helpers_neutron.OpenStackNeutron._build_networks([n])[0]
|
||||
m.return_value = create_occi_results([ooi_net])[0]
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": n}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
subnet_out = fakes.create_fake_json_resp(
|
||||
{"subnet": n["subnet_info"]}, 200)
|
||||
mock_subnet = mock.Mock(webob.Request)
|
||||
mock_subnet.get_response.return_value = subnet_out
|
||||
m.side_effect = [mock_net, mock_subnet]
|
||||
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="GET")
|
||||
|
|
@ -139,11 +190,159 @@ class TestNetworkController(TestMiddlewareNeutron):
|
|||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(network.Controller, "delete")
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_delete_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
port_out = fakes.create_fake_json_resp(
|
||||
{"ports": fakes.ports[tenant['id']]}, 200)
|
||||
mock_port = mock.Mock(webob.Request)
|
||||
mock_port.get_response.return_value = port_out
|
||||
empty_out = fakes.create_fake_json_resp([], 204)
|
||||
mock_empty = mock.Mock(webob.Request)
|
||||
mock_empty.get_response.return_value = empty_out
|
||||
m.side_effect = [mock_port, mock_empty, mock_empty,
|
||||
mock_empty, mock_empty]
|
||||
for n in fakes.networks[tenant["id"]]:
|
||||
m.return_value = create_occi_results([])
|
||||
m.return_value = fakes.create_fake_json_resp(
|
||||
{"subnet": n["subnet_info"]}, 200)
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
|
||||
class NetworkNeutronTextPlain(test_middleware.TestMiddlewareTextPlain,
|
||||
TestNetNeutronController):
|
||||
"""Test OCCI network controller with Accept: text/plain."""
|
||||
|
||||
|
||||
class NetworklNeutronTextOcci(test_middleware.TestMiddlewareTextOcci,
|
||||
TestNetNeutronController):
|
||||
"""Test OCCI network controller with Accept: text/occi."""
|
||||
|
||||
|
||||
class TestNetNovaController(test_middleware.TestMiddleware):
|
||||
"""Test OCCI network controller."""
|
||||
|
||||
def setUp(self):
|
||||
super(TestNetNovaController, self).setUp()
|
||||
self.schema = 'http://schemas.ogf.org/occi/infrastructure#network'
|
||||
self.accept = self.content_type = None
|
||||
self.application_url = fakes.application_url
|
||||
self.app = wsgi.OCCIMiddleware(None)
|
||||
|
||||
def assertExpectedResult(self, expected, result):
|
||||
expected = ["%s: %s" % e for e in expected]
|
||||
# NOTE(aloga): the order of the result does not matter
|
||||
results = str(result.text).splitlines()
|
||||
self.assertItemsEqual(expected, results)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks_empty(self, m):
|
||||
tenant = fakes.tenants["bar"]
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks_nova[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(204, resp.status_code)
|
||||
expected_result = ""
|
||||
self.assertExpectedResult(expected_result, resp)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks_nova[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = []
|
||||
for s in fakes.networks_nova[tenant["id"]]:
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % s["id"]))
|
||||
)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_create(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": fakes.networks_nova[tenant['id']][0]}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
m.side_effect = [mock_net]
|
||||
name = fakes.networks_nova[tenant["id"]][0]["label"]
|
||||
net_id = fakes.networks_nova[tenant["id"]][0]["id"]
|
||||
address = fakes.networks_nova[tenant["id"]][0]["cidr"]
|
||||
headers = {
|
||||
'Category': 'network;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind",'
|
||||
'ipnetwork;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network#";'
|
||||
'class="mixin",',
|
||||
'X-OCCI-Attribute': '"occi.core.title"="%s",'
|
||||
'"occi.network.address"="%s"' %
|
||||
(name, address)
|
||||
}
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X',
|
||||
method="POST",
|
||||
headers=headers)
|
||||
|
||||
m.return_value = fakes.networks_nova[tenant['id']][0]
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = [("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id))]
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_show_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
|
||||
for n in fakes.networks_nova[tenant["id"]]:
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": n}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
m.side_effect = [mock_net]
|
||||
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
expected = fakes.build_occi_nova(n)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_delete_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
empty_out = fakes.create_fake_json_resp(
|
||||
[], 204)
|
||||
mock_empty = mock.Mock(webob.Request)
|
||||
mock_empty.get_response.return_value = empty_out
|
||||
for n in fakes.networks_nova[tenant["id"]]:
|
||||
m.side_effect = [mock_empty]
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="DELETE")
|
||||
|
|
@ -184,11 +383,11 @@ class TestNetworkController(TestMiddlewareNeutron):
|
|||
self.assertEqual(400, resp.status_code)
|
||||
|
||||
|
||||
class NetworkControllerTextPlain(test_middleware.TestMiddlewareTextPlain,
|
||||
TestNetworkController):
|
||||
class NetworkNovaControllerTextPlain(test_middleware.TestMiddlewareTextPlain,
|
||||
TestNetNovaController):
|
||||
"""Test OCCI network controller with Accept: text/plain."""
|
||||
|
||||
|
||||
class NetworkControllerTextOcci(test_middleware.TestMiddlewareTextOcci,
|
||||
TestNetworkController):
|
||||
class NetworkNovaControllerTextOcci(test_middleware.TestMiddlewareTextOcci,
|
||||
TestNetNovaController):
|
||||
"""Test OCCI network controller with Accept: text/occi."""
|
||||
|
|
|
|||
|
|
@ -1,233 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2015 Spanish National Research Council
|
||||
# Copyright 2016 LIP - Lisbon
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import uuid
|
||||
|
||||
from ooi.tests import fakes
|
||||
from ooi.tests.functional.middleware import test_middleware
|
||||
from ooi import utils
|
||||
|
||||
|
||||
class TestFunctionalNetInterface(test_middleware.TestMiddleware):
|
||||
"""Test OCCI network interface controller."""
|
||||
|
||||
def setUp(self):
|
||||
super(TestFunctionalNetInterface, self).setUp()
|
||||
self.application_url = fakes.application_url
|
||||
self.app = self.get_app()
|
||||
|
||||
def test_list_ifaces_empty(self):
|
||||
tenant = fakes.tenants["bar"]
|
||||
|
||||
for url in ("/networklink/", "/networklink"):
|
||||
req = self._build_req(url, tenant["id"], method="GET")
|
||||
|
||||
req.environ["HTTP_X_PROJECT_ID"] = tenant["id"]
|
||||
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
expected_result = ""
|
||||
self.assertContentType(resp)
|
||||
self.assertExpectedResult(expected_result, resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
|
||||
def test_list_ifaces(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
|
||||
for url in ("/networklink/", "/networklink"):
|
||||
req = self._build_req(url, tenant["id"], method="GET")
|
||||
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = []
|
||||
float_list = {}
|
||||
for floating_ip in fakes.floating_ips[tenant["id"]]:
|
||||
if floating_ip["instance_id"]:
|
||||
float_list.update({floating_ip['fixed_ip']: floating_ip})
|
||||
instance_vm = fakes.linked_vm_id
|
||||
for p in fakes.ports[tenant["id"]]:
|
||||
for ip in p["fixed_ips"]:
|
||||
link_id = '_'.join([instance_vm,
|
||||
p["net_id"],
|
||||
ip["ip_address"]])
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"networklink/%s" % link_id))
|
||||
)
|
||||
float_ip = float_list.get(ip['ip_address'], None)
|
||||
if float_ip:
|
||||
link_id = '_'.join([instance_vm,
|
||||
"PUBLIC",
|
||||
float_ip["ip"]])
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"networklink/%s" % link_id))
|
||||
)
|
||||
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
def test_show_iface(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
instance_vm = fakes.linked_vm_id
|
||||
for p in fakes.ports[tenant["id"]]:
|
||||
for ip in p["fixed_ips"]:
|
||||
link_id = '_'.join([instance_vm,
|
||||
p["net_id"],
|
||||
ip["ip_address"]]
|
||||
)
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="GET")
|
||||
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
source = utils.join_url(self.application_url + "/",
|
||||
"compute/%s" % instance_vm)
|
||||
target = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % p["net_id"])
|
||||
self.assertResultIncludesLinkAttr(link_id, source, target,
|
||||
resp)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
|
||||
def test_show_invalid_id(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
link_id = uuid.uuid4().hex
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(404, resp.status_code)
|
||||
|
||||
def test_create_link_invalid(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
net_id = fakes.ports[tenant['id']][0]['net_id']
|
||||
occi_net_id = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id)
|
||||
headers = {
|
||||
'Category': (
|
||||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind"'),
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="foo", '
|
||||
'occi.core.target="%s"'
|
||||
) % occi_net_id
|
||||
}
|
||||
req = self._build_req("/networklink", None, method="POST",
|
||||
headers=headers)
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(400, resp.status_code)
|
||||
|
||||
def test_create_link_no_pool(self):
|
||||
tenant = fakes.tenants["foo"]
|
||||
net_id = fakes.ports[tenant['id']][0]['net_id']
|
||||
occi_compute_id = utils.join_url(
|
||||
self.application_url + "/",
|
||||
"compute/%s" % fakes.linked_vm_id)
|
||||
occi_net_id = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id)
|
||||
headers = {
|
||||
'Category': (
|
||||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind"'),
|
||||
'X-OCCI-Attribute': ('occi.core.source="%s", '
|
||||
'occi.core.target="%s"'
|
||||
) % (occi_compute_id, occi_net_id)
|
||||
}
|
||||
req = self._build_req("/networklink", tenant["id"], method="POST",
|
||||
headers=headers)
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
|
||||
def test_create_link_with_pool(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
link_info = fakes.ports[tenant['id']][0]
|
||||
|
||||
server_url = utils.join_url(self.application_url + "/",
|
||||
"compute/%s" % link_info['server_id'])
|
||||
net_url = utils.join_url(self.application_url + "/",
|
||||
"network/%s" % link_info['net_id'])
|
||||
pool_name = 'pool'
|
||||
headers = {
|
||||
'Category': (
|
||||
'networkinterface;'
|
||||
'scheme="http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind",'
|
||||
'%s;'
|
||||
'scheme="http://schemas.openstack.org/network/'
|
||||
'floatingippool#"; class="mixin"') % pool_name,
|
||||
'X-OCCI-Attribute': (
|
||||
'occi.core.source="%s", '
|
||||
'occi.core.target="%s"'
|
||||
) % (server_url, net_url)
|
||||
}
|
||||
req = self._build_req("/networklink", tenant["id"], method="POST",
|
||||
headers=headers)
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
link_id = '_'.join([link_info['server_id'],
|
||||
link_info['net_id'],
|
||||
link_info['fixed_ips'][0]
|
||||
["ip_address"]])
|
||||
expected = [("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"networklink/%s" % link_id))]
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
def test_delete_fixed(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
|
||||
for n in fakes.ports[tenant["id"]]:
|
||||
if n["net_id"] != "PUBLIC":
|
||||
if n["server_id"]:
|
||||
link_id = '_'.join([n["server_id"],
|
||||
n["net_id"],
|
||||
n["fixed_ips"]
|
||||
[0]["ip_address"]])
|
||||
req = self._build_req(
|
||||
"/networklink/%s" % link_id,
|
||||
tenant["id"], method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
|
||||
def test_delete_public(self):
|
||||
tenant = fakes.tenants["baz"]
|
||||
for n in fakes.floating_ips[tenant["id"]]:
|
||||
if n["instance_id"]:
|
||||
link_id = '_'.join([n["instance_id"],
|
||||
"PUBLIC",
|
||||
n["ip"]])
|
||||
req = self._build_req("/networklink/%s" % link_id,
|
||||
tenant["id"], method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertContentType(resp)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
|
||||
|
||||
class NetInterfaceControllerTextPlain(test_middleware.TestMiddlewareTextPlain,
|
||||
TestFunctionalNetInterface):
|
||||
"""Test OCCI network link controller with Accept: text/plain."""
|
||||
|
||||
|
||||
class NetInterfaceControllerTextOcci(test_middleware.TestMiddlewareTextOcci,
|
||||
TestFunctionalNetInterface):
|
||||
"""Test OCCI network link controller with Accept: text/occi."""
|
||||
|
|
@ -1,351 +0,0 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
|
||||
# Copyright 2015 Spanish National Research Council
|
||||
# Copyright 2016 LIP - Lisbon
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
||||
# not use this file except in compliance with the License. You may obtain
|
||||
# a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
||||
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import mock
|
||||
|
||||
import uuid
|
||||
import warnings
|
||||
|
||||
from oslo_config import cfg
|
||||
import webob
|
||||
|
||||
from ooi.api import helpers
|
||||
from ooi.tests import fakes_network as fakes
|
||||
from ooi.tests.functional.middleware import test_middleware
|
||||
from ooi import utils
|
||||
from ooi import wsgi
|
||||
|
||||
|
||||
class TestFunctionalNeutron(test_middleware.TestMiddleware):
|
||||
"""Test OCCI compute controller."""
|
||||
|
||||
def setUp(self):
|
||||
super(TestFunctionalNeutron, self).setUp()
|
||||
self.schema = 'http://schemas.ogf.org/occi/infrastructure#network'
|
||||
self.accept = self.content_type = None
|
||||
self.application_url = fakes.application_url
|
||||
neutron_ooi_endpoint = "foo"
|
||||
|
||||
def mock_endpoint(self, bar):
|
||||
if bar == "neutron_ooi_endpoint":
|
||||
return neutron_ooi_endpoint
|
||||
|
||||
with mock.patch.object(cfg.ConfigOpts, "__getattr__",
|
||||
side_effect=mock_endpoint,
|
||||
autospec=True):
|
||||
self.app = wsgi.OCCIMiddleware(None)
|
||||
|
||||
def assertExpectedResult(self, expected, result):
|
||||
expected = ["%s: %s" % e for e in expected]
|
||||
# NOTE(aloga): the order of the result does not matter
|
||||
results = str(result.text).splitlines()
|
||||
self.assertItemsEqual(expected, results)
|
||||
|
||||
def test_deprecated_configuration(self):
|
||||
with warnings.catch_warnings(record=True) as w:
|
||||
neutron_endopoint = "/foo"
|
||||
warnings.simplefilter("always", DeprecationWarning)
|
||||
wsgi.OCCIMiddleware(None,
|
||||
neutron_ooi_endpoint=neutron_endopoint)
|
||||
expected_message = (
|
||||
"Configuration of neutron_ooi_endpoint"
|
||||
" in api-paste.ini file is deprecated,"
|
||||
" include it in nova.conf")
|
||||
self.assertEqual(1, len(w))
|
||||
self.assertIs(DeprecationWarning,
|
||||
w[-1].category)
|
||||
self.assertEqual(expected_message, w[-1].message.message)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks_empty(self, m):
|
||||
tenant = fakes.tenants["bar"]
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(204, resp.status_code)
|
||||
expected_result = ""
|
||||
self.assertExpectedResult(expected_result, resp)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = []
|
||||
for s in fakes.networks[tenant["id"]]:
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % s["id"]))
|
||||
)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_create(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": fakes.networks[tenant['id']][0]}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
subnet_out = fakes.create_fake_json_resp(
|
||||
{"subnet": fakes.networks[tenant['id']][0]["subnet_info"]},
|
||||
200)
|
||||
mock_subnet = mock.Mock(webob.Request)
|
||||
mock_subnet.get_response.return_value = subnet_out
|
||||
public_out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks[tenant['id']]},
|
||||
200)
|
||||
|
||||
mock_public = mock.Mock(webob.Request)
|
||||
mock_public.get_response.return_value = public_out
|
||||
router_out = fakes.create_fake_json_resp(
|
||||
{"router": {"id": uuid.uuid4().hex}},
|
||||
200)
|
||||
mock_router = mock.Mock(webob.Request)
|
||||
mock_router.get_response.return_value = router_out
|
||||
mock_iface = mock.Mock(webob.Request)
|
||||
mock_iface.get_response.return_value = fakes.create_fake_json_resp(
|
||||
{"foo": "foo"}, 200)
|
||||
m.side_effect = [mock_net, mock_subnet, mock_public,
|
||||
mock_router, mock_iface
|
||||
]
|
||||
name = fakes.networks[tenant["id"]][0]["name"]
|
||||
net_id = fakes.networks[tenant["id"]][0]["id"]
|
||||
address = fakes.networks[tenant["id"]][0]["subnet_info"]["cidr"]
|
||||
headers = {
|
||||
'Category': 'network;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind",'
|
||||
'ipnetwork;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure/'
|
||||
'network#";'
|
||||
'class="mixin",',
|
||||
'X-OCCI-Attribute': '"occi.core.title"="%s",'
|
||||
'"occi.network.address"="%s"' %
|
||||
(name, address)
|
||||
}
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X',
|
||||
method="POST",
|
||||
headers=headers)
|
||||
|
||||
m.return_value = fakes.networks[tenant['id']][0]
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = [("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id))]
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_show_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
|
||||
for n in fakes.networks[tenant["id"]]:
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": n}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
subnet_out = fakes.create_fake_json_resp(
|
||||
{"subnet": n["subnet_info"]}, 200)
|
||||
mock_subnet = mock.Mock(webob.Request)
|
||||
mock_subnet.get_response.return_value = subnet_out
|
||||
m.side_effect = [mock_net, mock_subnet]
|
||||
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
expected = fakes.build_occi_network(n)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_delete_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
port_out = fakes.create_fake_json_resp(
|
||||
{"ports": fakes.ports[tenant['id']]}, 200)
|
||||
mock_port = mock.Mock(webob.Request)
|
||||
mock_port.get_response.return_value = port_out
|
||||
empty_out = fakes.create_fake_json_resp([], 204)
|
||||
mock_empty = mock.Mock(webob.Request)
|
||||
mock_empty.get_response.return_value = empty_out
|
||||
m.side_effect = [mock_port, mock_empty, mock_empty,
|
||||
mock_empty, mock_empty]
|
||||
for n in fakes.networks[tenant["id"]]:
|
||||
m.return_value = fakes.create_fake_json_resp(
|
||||
{"subnet": n["subnet_info"]}, 200)
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
|
||||
class NetworkControllerTextPlain(test_middleware.TestMiddlewareTextPlain,
|
||||
TestFunctionalNeutron):
|
||||
"""Test OCCI network controller with Accept: text/plain."""
|
||||
|
||||
|
||||
class NetworkControllerTextOcci(test_middleware.TestMiddlewareTextOcci,
|
||||
TestFunctionalNeutron):
|
||||
"""Test OCCI network controller with Accept: text/occi."""
|
||||
|
||||
|
||||
class TestFunctionalNova(test_middleware.TestMiddleware):
|
||||
"""Test OCCI compute controller."""
|
||||
|
||||
def setUp(self):
|
||||
super(TestFunctionalNova, self).setUp()
|
||||
self.schema = 'http://schemas.ogf.org/occi/infrastructure#network'
|
||||
self.accept = self.content_type = None
|
||||
self.application_url = fakes.application_url
|
||||
self.app = wsgi.OCCIMiddleware(None)
|
||||
|
||||
def assertExpectedResult(self, expected, result):
|
||||
expected = ["%s: %s" % e for e in expected]
|
||||
# NOTE(aloga): the order of the result does not matter
|
||||
results = str(result.text).splitlines()
|
||||
self.assertItemsEqual(expected, results)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks_empty(self, m):
|
||||
tenant = fakes.tenants["bar"]
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks_nova[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(204, resp.status_code)
|
||||
expected_result = ""
|
||||
self.assertExpectedResult(expected_result, resp)
|
||||
self.assertDefaults(resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_list_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
out = fakes.create_fake_json_resp(
|
||||
{"networks": fakes.networks_nova[tenant['id']]}, 200)
|
||||
m.return_value.get_response.return_value = out
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X', method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = []
|
||||
for s in fakes.networks_nova[tenant["id"]]:
|
||||
expected.append(
|
||||
("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % s["id"]))
|
||||
)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_create(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": fakes.networks_nova[tenant['id']][0]}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
m.side_effect = [mock_net]
|
||||
name = fakes.networks_nova[tenant["id"]][0]["label"]
|
||||
net_id = fakes.networks_nova[tenant["id"]][0]["id"]
|
||||
address = fakes.networks_nova[tenant["id"]][0]["cidr"]
|
||||
headers = {
|
||||
'Category': 'network;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
'class="kind",'
|
||||
'ipnetwork;'
|
||||
' scheme='
|
||||
'"http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network#";'
|
||||
'class="mixin",',
|
||||
'X-OCCI-Attribute': '"occi.core.title"="%s",'
|
||||
'"occi.network.address"="%s"' %
|
||||
(name, address)
|
||||
}
|
||||
req = self._build_req(path="/network",
|
||||
tenant_id='X',
|
||||
method="POST",
|
||||
headers=headers)
|
||||
|
||||
m.return_value = fakes.networks_nova[tenant['id']][0]
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
expected = [("X-OCCI-Location",
|
||||
utils.join_url(self.application_url + "/",
|
||||
"network/%s" % net_id))]
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_show_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
|
||||
for n in fakes.networks_nova[tenant["id"]]:
|
||||
net_out = fakes.create_fake_json_resp(
|
||||
{"network": n}, 200)
|
||||
mock_net = mock.Mock(webob.Request)
|
||||
mock_net.get_response.return_value = net_out
|
||||
m.side_effect = [mock_net]
|
||||
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
expected = fakes.build_occi_nova(n)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
||||
@mock.patch.object(helpers.BaseHelper, "_get_req")
|
||||
def test_delete_networks(self, m):
|
||||
tenant = fakes.tenants["foo"]
|
||||
empty_out = fakes.create_fake_json_resp(
|
||||
[], 204)
|
||||
mock_empty = mock.Mock(webob.Request)
|
||||
mock_empty.get_response.return_value = empty_out
|
||||
for n in fakes.networks_nova[tenant["id"]]:
|
||||
m.side_effect = [mock_empty]
|
||||
req = self._build_req(path="/network/%s" % n["id"],
|
||||
tenant_id='X',
|
||||
method="DELETE")
|
||||
resp = req.get_response(self.app)
|
||||
self.assertEqual(204, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
Loading…
Reference in New Issue