e2207d7c12
List the routers of the tenant and use the first router with external gateway, thus It is linked to the public network. In case no router with those features is found, a new one is created. Change-Id: I696ebe04426fab7e210d86f30084c4fee64cbce8 Closes-bug: #1637497
465 lines
19 KiB
Python
465 lines
19 KiB
Python
# Copyright 2015 Spanish National Research Council
|
|
# Copyright 2015 LIP - INDIGO-DataCloud
|
|
#
|
|
# 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 TestNetNeutronController(test_middleware.TestMiddleware):
|
|
"""Test OCCI network controller."""
|
|
|
|
def setUp(self):
|
|
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
|
|
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
|
|
list_router_out = fakes.create_fake_json_resp(
|
|
{"routers": []},
|
|
200)
|
|
mock_list_router = mock.Mock(webob.Request)
|
|
mock_list_router.get_response.return_value = list_router_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_list_router, 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_create_router_exists(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
|
|
list_router_out = fakes.create_fake_json_resp(
|
|
{"routers": [
|
|
{"id": uuid.uuid4().hex,
|
|
"external_gateway_info": None},
|
|
{"id": uuid.uuid4().hex,
|
|
"external_gateway_info": {"network_id": uuid.uuid4().hex}},
|
|
]
|
|
},
|
|
200)
|
|
mock_list_router = mock.Mock(webob.Request)
|
|
mock_list_router.get_response.return_value = list_router_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_list_router, 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 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")
|
|
resp = req.get_response(self.app)
|
|
self.assertEqual(204, resp.status_code)
|
|
self.assertDefaults(resp)
|
|
|
|
def test_action_net(self):
|
|
tenant = fakes.tenants["foo"]
|
|
|
|
for action in ("up", "down"):
|
|
headers = {
|
|
'Category': (
|
|
'%s;'
|
|
'scheme="http://schemas.ogf.org/occi/infrastructure/'
|
|
'network/action#";'
|
|
'class="action"' % action)
|
|
}
|
|
for net in fakes.networks[tenant["id"]]:
|
|
req = self._build_req("/network/%s?action=%s" % (net["id"],
|
|
action),
|
|
tenant_id=tenant["id"], method="POST",
|
|
headers=headers)
|
|
resp = req.get_response(self.app)
|
|
self.assertDefaults(resp)
|
|
self.assertEqual(501, resp.status_code)
|
|
|
|
def test_invalid_action(self):
|
|
tenant = fakes.tenants["foo"]
|
|
|
|
action = "foo"
|
|
for net in fakes.networks[tenant["id"]]:
|
|
req = self._build_req("/network/%s?action=%s" % (net["id"],
|
|
action),
|
|
tenant["id"], method="POST")
|
|
resp = req.get_response(self.app)
|
|
self.assertDefaults(resp)
|
|
self.assertEqual(400, resp.status_code)
|
|
|
|
|
|
class NetworkNovaControllerTextPlain(test_middleware.TestMiddlewareTextPlain,
|
|
TestNetNovaController):
|
|
"""Test OCCI network controller with Accept: text/plain."""
|
|
|
|
|
|
class NetworkNovaControllerTextOcci(test_middleware.TestMiddlewareTextOcci,
|
|
TestNetNovaController):
|
|
"""Test OCCI network controller with Accept: text/occi."""
|