x/ooi
Code Issues Proposed changes

Modify structure of test to separe functional tests

Browse Source 
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
2016-07-01 11:58:40 +01:00
parent 599fa241ac
commit d7b13e354d
4 changed files with 357 additions and 766 deletions
Download Patch File Download Diff File Expand all files Collapse all files

226
ooi/tests/functional/middleware/test_netinterface_controller.py
View File

@@ -17,15 +17,9 @@
import uuid import uuid
import mock from ooi.tests import fakes
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.functional.middleware import test_middleware from ooi.tests.functional.middleware import test_middleware
from ooi import utils from ooi import utils
from ooi import wsgi
class TestNetInterfaceController(test_middleware.TestMiddleware): class TestNetInterfaceController(test_middleware.TestMiddleware):
@@ -34,14 +28,11 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
def setUp(self): def setUp(self):
super(TestNetInterfaceController, self).setUp() super(TestNetInterfaceController, self).setUp()
self.application_url = fakes.application_url 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):
def test_list_ifaces_empty(self, mock_index):
tenant = fakes.tenants["bar"] tenant = fakes.tenants["bar"]
mock_index.return_value = fakes.fake_network_link_occi(
fakes.network_links[tenant['id']]
)
for url in ("/networklink/", "/networklink"): for url in ("/networklink/", "/networklink"):
req = self._build_req(url, tenant["id"], method="GET") req = self._build_req(url, tenant["id"], method="GET")
@@ -54,14 +45,8 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
self.assertExpectedResult(expected_result, resp) self.assertExpectedResult(expected_result, resp)
self.assertEqual(204, resp.status_code) self.assertEqual(204, resp.status_code)
@mock.patch.object(network_link.Controller, "index") def test_list_ifaces(self):
def test_list_ifaces(self, mock_index): tenant = fakes.tenants["baz"]
tenant = fakes.tenants["foo"]
mock_index.return_value = collection.Collection(
fakes.fake_network_link_occi(
fakes.network_links[tenant['id']]
)
)
for url in ("/networklink/", "/networklink"): for url in ("/networklink/", "/networklink"):
req = self._build_req(url, tenant["id"], method="GET") req = self._build_req(url, tenant["id"], method="GET")
@@ -70,62 +55,67 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
expected = [] expected = []
for ip in fakes.network_links[tenant["id"]]: float_list = {}
if ip["instance_id"]: for floating_ip in fakes.floating_ips[tenant["id"]]:
# fixme(jorgesece): test in case of instance None if floating_ip["instance_id"]:
link_id = '_'.join([ip["instance_id"], float_list.update({floating_ip['fixed_ip']: floating_ip})
ip["network_id"], instance_vm = fakes.linked_vm_id
ip["ip"]]) 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( expected.append(
("X-OCCI-Location", ("X-OCCI-Location",
utils.join_url(self.application_url + "/", utils.join_url(self.application_url + "/",
"networklink/%s" % link_id)) "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) self.assertExpectedResult(expected, resp)
@mock.patch.object(network_link.Controller, "show") def test_show_iface(self):
def test_show_iface(self, m_show): tenant = fakes.tenants["baz"]
tenant = fakes.tenants["foo"] instance_vm = fakes.linked_vm_id
m_show.return_value = fakes.fake_network_link_occi( for p in fakes.ports[tenant["id"]]:
fakes.network_links[tenant['id']] for ip in p["fixed_ips"]:
) link_id = '_'.join([instance_vm,
p["net_id"],
for ip in fakes.network_links[tenant["id"]]: ip["ip_address"]]
if ip["instance_id"] is not None:
link_id = '_'.join([ip["instance_id"],
ip["network_id"],
ip["ip"]]
) )
req = self._build_req("/networklink/%s" % link_id, req = self._build_req("/networklink/%s" % link_id,
tenant["id"], method="GET") tenant["id"], method="GET")
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertContentType(resp) self.assertContentType(resp)
# net_id = uuid.uuid4().hex
source = utils.join_url(self.application_url + "/", source = utils.join_url(self.application_url + "/",
"compute/%s" % ip["instance_id"]) "compute/%s" % instance_vm)
target = utils.join_url(self.application_url + "/", target = utils.join_url(self.application_url + "/",
"network/%s" % ip['network_id']) "network/%s" % p["net_id"])
self.assertResultIncludesLinkAttr(link_id, source, target, self.assertResultIncludesLinkAttr(link_id, source, target,
resp) resp)
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
@mock.patch.object(network_link.Controller, "show") def test_show_invalid_id(self):
def test_show_invalid_id(self, m_show):
tenant = fakes.tenants["foo"] tenant = fakes.tenants["foo"]
link_id = uuid.uuid4().hex link_id = uuid.uuid4().hex
m_show.side_effect = exception.LinkNotFound(link_id=link_id)
req = self._build_req("/networklink/%s" % link_id, req = self._build_req("/networklink/%s" % link_id,
tenant["id"], method="GET") tenant["id"], method="GET")
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertEqual(404, resp.status_code) self.assertEqual(404, resp.status_code)
@mock.patch.object(network_link.Controller, "create") def test_create_link_invalid(self):
def test_create_link_invalid(self, m_create):
tenant = fakes.tenants["foo"] tenant = fakes.tenants["foo"]
m_create.side_effect = exception.Invalid net_id = fakes.ports[tenant['id']][0]['net_id']
net_id = fakes.network_links[tenant['id']][0]['network_id']
occi_net_id = utils.join_url(self.application_url + "/", occi_net_id = utils.join_url(self.application_url + "/",
"network/%s" % net_id) "network/%s" % net_id)
headers = { headers = {
@@ -133,23 +123,21 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
'networkinterface;' 'networkinterface;'
'scheme="http://schemas.ogf.org/occi/infrastructure#";' 'scheme="http://schemas.ogf.org/occi/infrastructure#";'
'class="kind"'), 'class="kind"'),
'X-OCCI-Attribute': ( 'X-OCCI-Attribute': ('occi.core.source="foo", '
'occi.core.source="foo", ' 'occi.core.target="%s"'
'occi.core.target="%s"' ) % occi_net_id
) % occi_net_id
} }
req = self._build_req("/networklink", None, method="POST", req = self._build_req("/networklink", None, method="POST",
headers=headers) headers=headers)
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertEqual(400, resp.status_code) self.assertEqual(400, resp.status_code)
@mock.patch.object(network_link.Controller, "create") def test_create_link_no_pool(self):
def test_create_link_no_pool(self, m_create):
tenant = fakes.tenants["foo"] tenant = fakes.tenants["foo"]
m_create.return_value = fakes.fake_network_link_occi( net_id = fakes.ports[tenant['id']][0]['net_id']
fakes.network_links[tenant['id']] occi_compute_id = utils.join_url(
)[0] self.application_url + "/",
net_id = fakes.network_links[tenant['id']][0]['network_id'] "compute/%s" % fakes.linked_vm_id)
occi_net_id = utils.join_url(self.application_url + "/", occi_net_id = utils.join_url(self.application_url + "/",
"network/%s" % net_id) "network/%s" % net_id)
headers = { headers = {
@@ -157,50 +145,44 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
'networkinterface;' 'networkinterface;'
'scheme="http://schemas.ogf.org/occi/infrastructure#";' 'scheme="http://schemas.ogf.org/occi/infrastructure#";'
'class="kind"'), 'class="kind"'),
'X-OCCI-Attribute': ( 'X-OCCI-Attribute': ('occi.core.source="%s", '
'occi.core.source="foo", ' 'occi.core.target="%s"'
'occi.core.target="%s"' ) % (occi_compute_id, occi_net_id)
) % occi_net_id
} }
req = self._build_req("/networklink", None, method="POST", req = self._build_req("/networklink", tenant["id"], method="POST",
headers=headers) headers=headers)
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
@mock.patch.object(network_link.Controller, "create") def test_create_link_with_pool(self):
def test_create_link_with_pool(self, m_create): tenant = fakes.tenants["baz"]
tenant = fakes.tenants["foo"] link_info = fakes.ports[tenant['id']][0]
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]
server_url = utils.join_url(self.application_url + "/", 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 + "/", net_url = utils.join_url(self.application_url + "/",
"network/%s" % link_info['network_id']) "network/%s" % link_info['net_id'])
pool_name = 'pool' pool_name = 'pool'
headers = { headers = {
'Category': ( 'Category': ('networkinterface;'
'networkinterface;' 'scheme="http://schemas.ogf.org/occi/'
'scheme="http://schemas.ogf.org/occi/infrastructure#";' 'infrastructure#";'
'class="kind",' 'class="kind",'
'%s;' '%s;'
'scheme="http://schemas.openstack.org/network/' 'scheme="http://schemas.openstack.org/network/'
'floatingippool#"; class="mixin"') % pool_name, 'floatingippool#"; class="mixin"') % pool_name,
'X-OCCI-Attribute': ( 'X-OCCI-Attribute': ('occi.core.source="%s", '
'occi.core.source="%s", ' 'occi.core.target="%s"'
'occi.core.target="%s"' ) % (server_url, net_url)
) % (server_url, net_url)
} }
req = self._build_req("/networklink", tenant["id"], method="POST", req = self._build_req("/networklink", tenant["id"], method="POST",
headers=headers) headers=headers)
resp = req.get_response(self.app) resp = req.get_response(self.app)
link_id = '_'.join([link_info['instance_id'], link_id = '_'.join([link_info['server_id'],
link_info['network_id'], link_info['net_id'],
link_info['ip']]) link_info['fixed_ips'][0]
["ip_address"]])
expected = [("X-OCCI-Location", expected = [("X-OCCI-Location",
utils.join_url(self.application_url + "/", utils.join_url(self.application_url + "/",
"networklink/%s" % link_id))] "networklink/%s" % link_id))]
@@ -208,39 +190,35 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
self.assertExpectedResult(expected, resp) self.assertExpectedResult(expected, resp)
self.assertDefaults(resp) self.assertDefaults(resp)
@mock.patch.object(network_link.Controller, "delete") def test_delete_fixed(self):
def test_delete_fixed(self, m_delete): tenant = fakes.tenants["baz"]
tenant = fakes.tenants["foo"]
m_delete.return_value = []
for n in fakes.network_links[tenant["id"]]: for n in fakes.ports[tenant["id"]]:
if n["network_id"] != "PUBLIC": if n["net_id"] != "PUBLIC":
if n["instance_id"]: if n["server_id"]:
link_id = '_'.join([n["instance_id"], link_id = '_'.join([n["server_id"],
n["network_id"], n["net_id"],
n["ip"]]) n["fixed_ips"]
req = self._build_req("/networklink/%s" % link_id, [0]["ip_address"]])
tenant["id"], method="DELETE") req = self._build_req(
"/networklink/%s" % link_id,
tenant["id"], method="DELETE")
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertContentType(resp) self.assertContentType(resp)
self.assertEqual(204, resp.status_code) self.assertEqual(204, resp.status_code)
@mock.patch.object(network_link.Controller, "delete") def test_delete_public(self):
def test_delete_public(self, m_delete): tenant = fakes.tenants["baz"]
tenant = fakes.tenants["public"] for n in fakes.floating_ips[tenant["id"]]:
m_delete.return_value = [] if n["instance_id"]:
link_id = '_'.join([n["instance_id"],
for n in fakes.network_links[tenant["id"]]: "PUBLIC",
if n["network_id"] != "PUBLIC": n["ip"]])
if n["instance_id"]: req = self._build_req("/networklink/%s" % link_id,
link_id = '_'.join([n["instance_id"], tenant["id"], method="DELETE")
n["network_id"], resp = req.get_response(self.app)
n["ip"]]) self.assertContentType(resp)
req = self._build_req("/networklink/%s" % link_id, self.assertEqual(204, resp.status_code)
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): def test_show_non_existant_compute(self):
tenant = fakes.tenants["foo"] tenant = fakes.tenants["foo"]
@@ -260,10 +238,9 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
'networkinterface;' 'networkinterface;'
'scheme="http://schemas.ogf.org/occi/infrastructure#";' 'scheme="http://schemas.ogf.org/occi/infrastructure#";'
'class="kind"'), 'class="kind"'),
'X-OCCI-Attribute': ( 'X-OCCI-Attribute': ('occi.core.source="foo", '
'occi.core.source="foo", ' 'occi.core.target="%s"'
'occi.core.target="%s"' ) % net_id
) % net_id
} }
req = self._build_req("/networklink", None, method="POST", req = self._build_req("/networklink", None, method="POST",
headers=headers) headers=headers)
@@ -279,10 +256,9 @@ class TestNetInterfaceController(test_middleware.TestMiddleware):
'networkinterface;' 'networkinterface;'
'scheme="http://schemas.ogf.org/occi/infrastructure#";' 'scheme="http://schemas.ogf.org/occi/infrastructure#";'
'class="kind"'), 'class="kind"'),
'X-OCCI-Attribute': ( 'X-OCCI-Attribute': ('occi.core.source="%s", '
'occi.core.source="%s", ' 'occi.core.target="bar"'
'occi.core.target="bar"' ) % server_id
) % server_id
} }
req = self._build_req("/networklink", None, method="POST", req = self._build_req("/networklink", None, method="POST",
headers=headers) headers=headers)

313
ooi/tests/functional/middleware/test_network_controller.py
View File

@@ -17,37 +17,37 @@
import mock import mock
from ooi.api import helpers_neutron import uuid
from ooi.api import network import warnings
from ooi.occi.core import collection
from oslo_config import cfg
import webob
from ooi.api import helpers
from ooi.tests import fakes_network as fakes from ooi.tests import fakes_network as fakes
from ooi.tests.functional.middleware import test_middleware from ooi.tests.functional.middleware import test_middleware
from ooi import utils from ooi import utils
from ooi import wsgi from ooi import wsgi
def create_occi_results(data): class TestNetNeutronController(test_middleware.TestMiddleware):
return network.Controller(None)._get_network_resources(data) """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): 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.accept = self.content_type = None
self.application_url = fakes.application_url 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): with mock.patch.object(cfg.ConfigOpts, "__getattr__",
"""Test OCCI compute controller.""" side_effect=mock_endpoint,
autospec=True):
def setUp(self): self.app = wsgi.OCCIMiddleware(None)
super(TestNetworkController, self).setUp()
def assertExpectedResult(self, expected, result): def assertExpectedResult(self, expected, result):
expected = ["%s: %s" % e for e in expected] expected = ["%s: %s" % e for e in expected]
@@ -55,34 +55,43 @@ class TestNetworkController(TestMiddlewareNeutron):
results = str(result.text).splitlines() results = str(result.text).splitlines()
self.assertItemsEqual(expected, results) 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): def test_list_networks_empty(self, m):
tenant = fakes.tenants["bar"] tenant = fakes.tenants["bar"]
headers = { out = fakes.create_fake_json_resp(
'Category': 'network; scheme="http://schema#";class="kind";', {"networks": fakes.networks[tenant['id']]}, 200)
'X_OCCI_Attribute': 'project=%s' % tenant["id"], m.return_value.get_response.return_value = out
}
url = "/network" req = self._build_req(path="/network",
req = self._build_req(path=url, tenant_id='X', method="GET")
tenant_id='X',
method="GET",
headers=headers, content_type="text/occi")
m.return_value = collection.Collection(
create_occi_results(fakes.networks[tenant['id']]))
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertEqual(204, resp.status_code) self.assertEqual(204, resp.status_code)
expected_result = "" expected_result = ""
self.assertExpectedResult(expected_result, resp) self.assertExpectedResult(expected_result, resp)
self.assertDefaults(resp) self.assertDefaults(resp)
@mock.patch.object(network.Controller, "index") @mock.patch.object(helpers.BaseHelper, "_get_req")
def test_list_networks(self, m): def test_list_networks(self, m):
tenant = fakes.tenants["foo"] tenant = fakes.tenants["foo"]
ooi_net = helpers_neutron.OpenStackNeutron._build_networks( out = fakes.create_fake_json_resp(
fakes.networks[tenant['id']] {"networks": fakes.networks[tenant['id']]}, 200)
) m.return_value.get_response.return_value = out
m.return_value = collection.Collection(
create_occi_results(ooi_net))
req = self._build_req(path="/network", req = self._build_req(path="/network",
tenant_id='X', method="GET") tenant_id='X', method="GET")
resp = req.get_response(self.app) resp = req.get_response(self.app)
@@ -98,38 +107,80 @@ class TestNetworkController(TestMiddlewareNeutron):
self.assertDefaults(resp) self.assertDefaults(resp)
self.assertExpectedResult(expected, resp) self.assertExpectedResult(expected, resp)
@mock.patch.object(network.Controller, "create") @mock.patch.object(helpers.BaseHelper, "_get_req")
def test_create(self, m): def test_create(self, m):
tenant = fakes.tenants["foo"] 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 = { headers = {
'Category': 'network; scheme="http://schema#";class="kind",' + 'Category': 'network;'
'mixinID;' ' scheme='
'scheme="http://schemas.openstack.org/template/os#";' '"http://schemas.ogf.org/occi/infrastructure#";'
' class=mixin', 'class="kind",'
'X_Occi_Attribute': 'project=%s' % tenant["id"], '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", req = self._build_req(path="/network",
tenant_id='X', tenant_id='X',
method="POST", method="POST",
headers=headers) headers=headers)
fake_net = fakes.fake_network_occi(
fakes.networks[tenant['id']] m.return_value = fakes.networks[tenant['id']][0]
)[0]
m.return_value = collection.Collection([fake_net])
resp = req.get_response(self.app) resp = req.get_response(self.app)
self.assertEqual(200, resp.status_code) self.assertEqual(200, resp.status_code)
expected = [("X-OCCI-Location", expected = [("X-OCCI-Location",
utils.join_url(self.application_url + "/", utils.join_url(self.application_url + "/",
"network/%s" % fake_net.id))] "network/%s" % net_id))]
self.assertExpectedResult(expected, resp) self.assertExpectedResult(expected, resp)
@mock.patch.object(network.Controller, "show") @mock.patch.object(helpers.BaseHelper, "_get_req")
def test_show_networks(self, m): def test_show_networks(self, m):
tenant = fakes.tenants["foo"] tenant = fakes.tenants["foo"]
for n in fakes.networks[tenant["id"]]: for n in fakes.networks[tenant["id"]]:
ooi_net = helpers_neutron.OpenStackNeutron._build_networks([n])[0] net_out = fakes.create_fake_json_resp(
m.return_value = create_occi_results([ooi_net])[0] {"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"], req = self._build_req(path="/network/%s" % n["id"],
tenant_id='X', tenant_id='X',
method="GET") method="GET")
@@ -139,11 +190,159 @@ class TestNetworkController(TestMiddlewareNeutron):
self.assertDefaults(resp) self.assertDefaults(resp)
self.assertExpectedResult(expected, resp) self.assertExpectedResult(expected, resp)
@mock.patch.object(network.Controller, "delete") @mock.patch.object(helpers.BaseHelper, "_get_req")
def test_delete_networks(self, m): def test_delete_networks(self, m):
tenant = fakes.tenants["foo"] 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"]]: 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"], req = self._build_req(path="/network/%s" % n["id"],
tenant_id='X', tenant_id='X',
method="DELETE") method="DELETE")
@@ -184,11 +383,11 @@ class TestNetworkController(TestMiddlewareNeutron):
self.assertEqual(400, resp.status_code) self.assertEqual(400, resp.status_code)
class NetworkControllerTextPlain(test_middleware.TestMiddlewareTextPlain, class NetworkNovaControllerTextPlain(test_middleware.TestMiddlewareTextPlain,
TestNetworkController): TestNetNovaController):
"""Test OCCI network controller with Accept: text/plain.""" """Test OCCI network controller with Accept: text/plain."""
class NetworkControllerTextOcci(test_middleware.TestMiddlewareTextOcci, class NetworkNovaControllerTextOcci(test_middleware.TestMiddlewareTextOcci,
TestNetworkController): TestNetNovaController):
"""Test OCCI network controller with Accept: text/occi.""" """Test OCCI network controller with Accept: text/occi."""

233
ooi/tests/middleware/test_functional_netinterface.py
View File

@@ -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."""

351
ooi/tests/middleware/test_functional_network.py
View File

@@ -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)