Merge "Add functional tests for network"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
4a3b5904d7
|
|
@ -157,24 +157,26 @@ networks = {
|
|||
ports = {
|
||||
tenants["foo"]["id"]: [
|
||||
{
|
||||
"id": uuid.uuid4().hex,
|
||||
"port_id": uuid.uuid4().hex,
|
||||
"fixed_ips":
|
||||
[{"ip_address": uuid.uuid4().hex}],
|
||||
"mac_addr": uuid.uuid4().hex,
|
||||
"port_state": "DOWN",
|
||||
"net_id": uuid.uuid4().hex
|
||||
"net_id": uuid.uuid4().hex,
|
||||
"server_id": linked_vm_id
|
||||
},
|
||||
],
|
||||
tenants["bar"]["id"]: [],
|
||||
tenants["baz"]["id"]: [
|
||||
{
|
||||
"id": uuid.uuid4().hex,
|
||||
"port_id": uuid.uuid4().hex,
|
||||
"fixed_ips": [
|
||||
{"ip_address": uuid.uuid4().hex}
|
||||
{"ip_address": "192.168.253.1"}
|
||||
],
|
||||
"mac_addr": uuid.uuid4().hex,
|
||||
"port_state": "ACTIVE",
|
||||
"net_id": uuid.uuid4().hex
|
||||
"net_id": uuid.uuid4().hex,
|
||||
"server_id": linked_vm_id
|
||||
},
|
||||
|
||||
],
|
||||
|
|
@ -563,6 +565,26 @@ class FakeApp(object):
|
|||
ip = {"floating_ip": {"ip": allocated_ip, "id": 1}}
|
||||
return create_fake_json_resp(ip, 202)
|
||||
|
||||
def _do_create_port(self, req):
|
||||
tenant = req.path_info.split('/')[1]
|
||||
body = req.json_body.copy()
|
||||
server = body["interfaceAttachment"]["server_id"]
|
||||
net = body["interfaceAttachment"]["net_id"]
|
||||
port = ports[tenant]
|
||||
p = {"interfaceAttachment": {
|
||||
"port_id": uuid.uuid4().hex,
|
||||
"fixed_ips":
|
||||
[{"ip_address":
|
||||
port[0]["fixed_ips"]
|
||||
[0]["ip_address"]
|
||||
}],
|
||||
"mac_addr": port[0]["mac_addr"],
|
||||
"port_state": "DOWN",
|
||||
"net_id": net,
|
||||
"server_id": server
|
||||
}}
|
||||
return create_fake_json_resp(p, 200)
|
||||
|
||||
def _do_post(self, req):
|
||||
if req.path_info.endswith("servers"):
|
||||
return self._do_create_server(req)
|
||||
|
|
@ -578,6 +600,8 @@ class FakeApp(object):
|
|||
return self._do_create_attachment(req)
|
||||
elif req.path_info.endswith("os-floating-ips"):
|
||||
return self._do_allocate_ip(req)
|
||||
elif req.path_info.endswith("os-interface"):
|
||||
return self._do_create_port(req)
|
||||
raise Exception
|
||||
|
||||
def _do_delete(self, req):
|
||||
|
|
@ -587,6 +611,7 @@ class FakeApp(object):
|
|||
r"/[^/]+/os-floating-ips/[^/]+$": 202,
|
||||
r"/[^/]+/servers/[^/]+$": 204,
|
||||
r"/[^/]+/os-volumes/[^/]+$": 204,
|
||||
r"/[^/]+/servers/[^/]+/os-interface/[^/]+$": 204,
|
||||
}
|
||||
for p, st in tested_paths.items():
|
||||
if re.match(p, req.path_info):
|
||||
|
|
|
|||
|
|
@ -71,6 +71,26 @@ networks = {
|
|||
]
|
||||
}
|
||||
|
||||
networks_nova = {
|
||||
tenants["bar"]["id"]: [],
|
||||
tenants["foo"]["id"]: [
|
||||
{
|
||||
"id": uuid.uuid4().hex,
|
||||
"label": "foo",
|
||||
"gateway": "33.0.0.1",
|
||||
"cidr": "33.0.0.1/24",
|
||||
"status": "ACTIVE",
|
||||
},
|
||||
{
|
||||
"id": uuid.uuid4().hex,
|
||||
"label": "bar",
|
||||
"gateway": "44.0.0.1",
|
||||
"cidr": "44.0.0.1/24",
|
||||
"status": "DOWN",
|
||||
},
|
||||
]
|
||||
}
|
||||
|
||||
pools = {
|
||||
tenants["bar"]["id"]: [],
|
||||
tenants["foo"]["id"]: [
|
||||
|
|
@ -90,6 +110,15 @@ linked_net_id = uuid.uuid4().hex
|
|||
|
||||
allocated_ip = "192.168.253.23"
|
||||
|
||||
ports = {
|
||||
tenants["foo"]["id"]: [
|
||||
{"id": uuid.uuid4().hex,
|
||||
"device_id": uuid.uuid4().hex,
|
||||
"device_owner": uuid.uuid4().hex
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
network_links = {
|
||||
tenants["bar"]["id"]: [],
|
||||
tenants["foo"]["id"]: [
|
||||
|
|
@ -236,4 +265,112 @@ def fake_network_occi(os_list_net):
|
|||
list_nets = []
|
||||
for n in os_list_net:
|
||||
list_nets.append(fake_build_net(n['name'], id=n['id']))
|
||||
return network.Controller._get_network_resources(list_nets)
|
||||
return network.Controller._get_network_resources(list_nets)
|
||||
|
||||
|
||||
def build_occi_network(network):
|
||||
name = network["name"]
|
||||
network_id = network["id"]
|
||||
subnet_info = network["subnet_info"]
|
||||
status = network["status"].upper()
|
||||
if status in ("ACTIVE",):
|
||||
status = "active"
|
||||
else:
|
||||
status = "inactive"
|
||||
|
||||
app_url = application_url
|
||||
cats = []
|
||||
cats.append('network; '
|
||||
'scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
' class="kind"; title="network resource";'
|
||||
' rel='
|
||||
'"http://schemas.ogf.org/occi/core#resource";'
|
||||
' location="%s/network/"' % app_url)
|
||||
cats.append('ipnetwork; '
|
||||
'scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure/network#";'
|
||||
' class="mixin"; title="IP Networking Mixin"')
|
||||
cats.append('osnetwork; '
|
||||
'scheme='
|
||||
'"http://schemas.openstack.org/infrastructure/network#";'
|
||||
' class="mixin"; title="openstack network"')
|
||||
|
||||
links = []
|
||||
links.append('<%s/network/%s?action=up>; '
|
||||
'rel="http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network/action#up"' %
|
||||
(application_url, network_id))
|
||||
links.append('<%s/network/%s?action=down>; '
|
||||
'rel="http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network/action#down"' %
|
||||
(application_url, network_id))
|
||||
|
||||
attrs = [
|
||||
'occi.core.id="%s"' % network_id,
|
||||
'occi.core.title="%s"' % name,
|
||||
'occi.network.state="%s"' % status,
|
||||
'org.openstack.network.ip_version="%s"' % subnet_info["ip_version"],
|
||||
'occi.network.address="%s"' % subnet_info["cidr"],
|
||||
'occi.network.gateway="%s"' % subnet_info["gateway_ip"],
|
||||
]
|
||||
result = []
|
||||
for c in cats:
|
||||
result.append(("Category", c))
|
||||
for a in attrs:
|
||||
result.append(("X-OCCI-Attribute", a))
|
||||
for l in links:
|
||||
result.append(("Link", l))
|
||||
return result
|
||||
|
||||
|
||||
def build_occi_nova(network):
|
||||
name = network["label"]
|
||||
network_id = network["id"]
|
||||
gateway = network["gateway"]
|
||||
cidr = network["cidr"]
|
||||
status = "active"
|
||||
|
||||
app_url = application_url
|
||||
cats = []
|
||||
cats.append('network; '
|
||||
'scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
' class="kind"; title="network resource";'
|
||||
' rel='
|
||||
'"http://schemas.ogf.org/occi/core#resource";'
|
||||
' location="%s/network/"' % app_url)
|
||||
cats.append('ipnetwork; '
|
||||
'scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure/network#";'
|
||||
' class="mixin"; title="IP Networking Mixin"')
|
||||
cats.append('osnetwork; '
|
||||
'scheme='
|
||||
'"http://schemas.openstack.org/infrastructure/network#";'
|
||||
' class="mixin"; title="openstack network"')
|
||||
|
||||
links = []
|
||||
links.append('<%s/network/%s?action=up>; '
|
||||
'rel="http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network/action#up"' %
|
||||
(application_url, network_id))
|
||||
links.append('<%s/network/%s?action=down>; '
|
||||
'rel="http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network/action#down"' %
|
||||
(application_url, network_id))
|
||||
|
||||
attrs = [
|
||||
'occi.core.id="%s"' % network_id,
|
||||
'occi.core.title="%s"' % name,
|
||||
'occi.network.state="%s"' % status,
|
||||
'occi.network.address="%s"' % cidr,
|
||||
'occi.network.gateway="%s"' % gateway,
|
||||
]
|
||||
result = []
|
||||
for c in cats:
|
||||
result.append(("Category", c))
|
||||
for a in attrs:
|
||||
result.append(("X-OCCI-Attribute", a))
|
||||
for l in links:
|
||||
result.append(("Link", l))
|
||||
return result
|
||||
|
|
|
|||
|
|
@ -0,0 +1,233 @@
|
|||
# -*- 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.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."""
|
||||
|
|
@ -0,0 +1,330 @@
|
|||
# -*- 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 webob
|
||||
|
||||
from ooi.api import helpers
|
||||
from ooi.tests import fakes_network as fakes
|
||||
from ooi.tests.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
|
||||
self.neutron_endpoint = "foo"
|
||||
self.app = wsgi.OCCIMiddleware(
|
||||
None,
|
||||
neutron_ooi_endpoint=self.neutron_endpoint)
|
||||
|
||||
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[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,
|
||||
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)
|
||||
|
|
@ -26,62 +26,6 @@ from ooi import utils
|
|||
from ooi import wsgi
|
||||
|
||||
|
||||
def build_occi_network(network):
|
||||
name = network["name"]
|
||||
network_id = network["id"]
|
||||
subnet_info = network["subnet_info"]
|
||||
status = network["status"].upper()
|
||||
if status in ("ACTIVE",):
|
||||
status = "active"
|
||||
else:
|
||||
status = "inactive"
|
||||
|
||||
app_url = fakes.application_url
|
||||
cats = []
|
||||
cats.append('network; '
|
||||
'scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure#";'
|
||||
' class="kind"; title="network resource";'
|
||||
' rel='
|
||||
'"http://schemas.ogf.org/occi/core#resource";'
|
||||
' location="%s/network/"' % app_url)
|
||||
cats.append('ipnetwork; '
|
||||
'scheme='
|
||||
'"http://schemas.ogf.org/occi/infrastructure/network#";'
|
||||
' class="mixin"; title="IP Networking Mixin"')
|
||||
cats.append('osnetwork; '
|
||||
'scheme='
|
||||
'"http://schemas.openstack.org/infrastructure/network#";'
|
||||
' class="mixin"; title="openstack network"')
|
||||
|
||||
links = []
|
||||
links.append('<%s/network/%s?action=up>; '
|
||||
'rel="http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network/action#up"' %
|
||||
(fakes.application_url, network_id))
|
||||
links.append('<%s/network/%s?action=down>; '
|
||||
'rel="http://schemas.ogf.org/occi/'
|
||||
'infrastructure/network/action#down"' %
|
||||
(fakes.application_url, network_id))
|
||||
|
||||
attrs = [
|
||||
'occi.core.id="%s"' % network_id,
|
||||
'occi.core.title="%s"' % name,
|
||||
'occi.network.state="%s"' % status,
|
||||
'org.openstack.network.ip_version="%s"' % subnet_info["ip_version"],
|
||||
'occi.network.address="%s"' % subnet_info["cidr"],
|
||||
'occi.network.gateway="%s"' % subnet_info["gateway_ip"],
|
||||
]
|
||||
result = []
|
||||
for c in cats:
|
||||
result.append(("Category", c))
|
||||
for a in attrs:
|
||||
result.append(("X-OCCI-Attribute", a))
|
||||
for l in links:
|
||||
result.append(("Link", l))
|
||||
return result
|
||||
|
||||
|
||||
def create_occi_results(data):
|
||||
return network.Controller(None)._get_network_resources(data)
|
||||
|
||||
|
|
@ -190,7 +134,7 @@ class TestNetworkController(TestMiddlewareNeutron):
|
|||
tenant_id='X',
|
||||
method="GET")
|
||||
resp = req.get_response(self.app)
|
||||
expected = build_occi_network(n)
|
||||
expected = fakes.build_occi_network(n)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertDefaults(resp)
|
||||
self.assertExpectedResult(expected, resp)
|
||||
|
|
|
|||
Loading…
Reference in New Issue