x
/
ooi
Code Issues Proposed changes

Create OCCI and Controller network for neutron 

Add controller for networks. It manages neutron or nova network
depending of existing the parameter neutro_ooi_endpoint.
Create OCCI for network in ooi/openstack/network.py.
It adds a new mixin with one parameter (IP version), even
this mixin can be used for the future to add more attributes.
Add tests for those changes.

Change-Id: I3c49ae9fcab9ac7f0adbaf559263a9d608793ce1
This commit is contained in:
jorgesece 2016-05-17 10:54:13 +01:00
parent 6c4246499b
commit a28f1436e3
5 changed files with 733 additions and 202 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

195
ooi/api/network.py
View File

@ -1,6 +1,7 @@
# -*- 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
@ -15,51 +16,187 @@
# under the License.
from ooi.api import base
import ooi.api.helpers
from ooi.api import helpers
from ooi.api import helpers_neutron
from ooi import exception
from ooi.occi.core import collection
from ooi.occi.infrastructure import network
from ooi.occi import validator as occi_validator
from ooi.openstack import network as os_network
FLOATING_PREFIX = "floating"
FIXED_PREFIX = "fixed"
def _build_network(name, prefix=None):
if prefix:
network_id = '/'.join([prefix, name])
def parse_validate_schema(req, scheme=None,
required_attr=None):
"""Parse attributes and validate scheme
Returns attributes from request
If scheme is specified, it validates the OCCI scheme:
-Raises exception in case of being invalid
:param req: request
:param: scheme: scheme to validate
:param: required_attr: attributes required
"""
parser = req.get_parser()(req.headers, req.body)
if scheme:
attributes = parser.parse()
validator = occi_validator.Validator(attributes)
validator.validate(scheme)
validator.validate_attributes(required_attr)
else:
network_id = name
return network.NetworkResource(title=name,
id=network_id,
state="active",
mixins=[network.ip_network])
attributes = parser.parse_attributes(req.headers)
return attributes
def process_parameters(req, scheme=None,
required_attr=None):
"""Get attributes from request parameters
:param req: request
:param: scheme: scheme to validate
:param: required_attr: attributes required
"""
parameters = parse_validate_schema(req, scheme, required_attr)
try:
attributes = {}
if 'X_PROJECT_ID' in req.headers:
attributes["X_PROJECT_ID"] = req.headers["X_PROJECT_ID"]
if "attributes" in parameters:
for k, v in parameters.get("attributes", None).items():
attributes[k.strip()] = v.strip()
if not attributes:
attributes = None
except Exception:
raise exception.Invalid
return attributes
class Controller(base.Controller):
def __init__(self, *args, **kwargs):
super(Controller, self).__init__(*args, **kwargs)
self.os_helper = ooi.api.helpers.OpenStackHelper(
self.app,
self.openstack_version
)
def __init__(self, app=None, openstack_version=None,
neutron_ooi_endpoint=None):
"""Network controller initialization
def _floating_index(self, req):
pools = self.os_helper.get_floating_ip_pools(req)
:param app: application
:param: openstack_version: nova version
:param: neutron_ooi_endpoint: This parameter
indicates the Neutron endpoint to load the Neutron Helper.
If it is None, nova-network is used.
"""
super(Controller, self).__init__(
app=app,
openstack_version=openstack_version)
if neutron_ooi_endpoint:
self.os_helper = helpers_neutron.OpenStackNeutron(
neutron_ooi_endpoint
)
else:
self.os_helper = helpers.OpenStackHelper(
self.app,
self.openstack_version
)
@staticmethod
def _get_network_resources(networks_list):
"""Create network instances from network in json format
:param networks_list: networks objects provides by
the cloud infrastructure
"""
occi_network_resources = []
if pools:
occi_network_resources.append(_build_network(FLOATING_PREFIX))
if networks_list:
for s in networks_list:
n_state = s['state']
n_id = s["id"]
n_name = s["name"]
n_address = s.get("address", None)
n_ip_version = s.get("ip_version", None)
n_gateway = s.get("gateway", None)
s = os_network.OSNetworkResource(title=n_name,
id=n_id, state=n_state,
ip_version=n_ip_version,
address=n_address,
gateway=n_gateway)
occi_network_resources.append(s)
return occi_network_resources
def index(self, req):
occi_network_resources = self._floating_index(req)
occi_network_resources.append(_build_network(FIXED_PREFIX))
return collection.Collection(resources=occi_network_resources)
"""List networks
:param req: request object
"""
occi_networks = self.os_helper.list_networks(req)
occi_network_resources = self._get_network_resources(
occi_networks)
return collection.Collection(
resources=occi_network_resources)
def show(self, req, id):
if id == FIXED_PREFIX:
return [_build_network(id)]
elif id == FLOATING_PREFIX:
pools = self.os_helper.get_floating_ip_pools(req)
if pools:
return [_build_network(id)]
raise exception.NetworkNotFound(resource_id=id)
"""Get network details
:param req: request object
:param id: network identification
"""
resp = self.os_helper.get_network_details(req, id)
occi_network_resources = self._get_network_resources(
[resp])
return occi_network_resources[0]
def create(self, req, body=None):
"""Create a network instance in the cloud
:param req: request object
:param body: body request (not used)
"""
scheme = {
"category": network.NetworkResource.kind,
"mixins": [
network.ip_network,
],
"optional_mixins": [
os_network.OSNetwork()
]
}
required = ["occi.core.title",
"occi.network.address",
]
attributes = process_parameters(req, scheme, required)
name = attributes.get('occi.core.title')
cidr = attributes.get('occi.network.address')
gateway = attributes.get('occi.network.gateway', None)
ip_version = attributes.get('org.openstack.network.ip_version', None)
net = self.os_helper.create_network(req, name=name,
cidr=cidr,
gateway=gateway,
ip_version=ip_version)
occi_network_resources = self._get_network_resources([net])
return collection.Collection(
resources=occi_network_resources)
def delete(self, req, id):
"""delete networks which satisfy the parameters
:param req: current request
:param id: identification
"""
response = self.os_helper.delete_network(req, id)
return response
def run_action(self, req, id, body):
"""Run action over the network
:param req: current request
:param id: network identification
:param body: body
"""
action = req.GET.get("action", None)
occi_actions = [a.term for a in network.NetworkResource.actions]
if action is None or action not in occi_actions:
raise exception.InvalidAction(action=action)
raise exception.NotImplemented("Network actions are not implemented")

13
ooi/occi/validator.py
View File

@ -83,6 +83,19 @@ class Validator(object):
raise exception.OCCISchemaMismatch(expected=expected_types,
found=l['rel'])
def validate_attributes(self, required):
"""Validate required attributes
:param required: required attributes
"""
attr = self.parsed_obj.get("attributes", {})
if required:
for at in required:
if at not in attr:
raise exception.Invalid(
"Expecting %s attribute" % at
)
def validate(self, schema):
if "category" in schema:
self._validate_category(schema["category"])

92
ooi/openstack/network.py
View File

@ -1,6 +1,7 @@
# -*- 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
@ -16,6 +17,7 @@
from ooi.occi.core import attribute as attr
from ooi.occi.core import mixin
from ooi.occi.infrastructure import network
from ooi.occi.infrastructure import network_link
from ooi.openstack import helpers
@ -32,14 +34,15 @@ class OSNetworkInterface(network_link.NetworkInterface):
"occi.networkinterface.gateway",
"occi.networkinterface.allocation"])
def __init__(self, source, target, mac, address, ip_id=None, pool=None):
def __init__(self, source, target, mac, address, ip_id=None,
pool=None, state='active'):
link_id = '_'.join([source.id, address])
mixins = [network_link.ip_network_interface]
if pool:
mixins.append(OSFloatingIPPool(pool))
super(OSNetworkInterface, self).__init__(mixins, source, target,
link_id, "eth0", mac,
"active")
state)
self.ip_id = ip_id
self.attributes["occi.networkinterface.address"] = (
attr.MutableAttribute("occi.networkinterface.address", address))
@ -72,3 +75,88 @@ class OSNetworkInterface(network_link.NetworkInterface):
@allocation.setter
def allocation(self, value):
self.attributes["occi.networkinterface.allocation"].value = value
class OSNetwork(mixin.Mixin):
scheme = helpers.build_scheme("infrastructure/network")
def __init__(self, pool=None):
term = "osnetwork"
title = "openstack network"
super(OSNetwork, self).__init__(
scheme=self.scheme,
term=term,
title=title,
attributes=attr.AttributeCollection([
"org.openstack.network.ip_version"
])
)
os_network = OSNetwork()
class OSNetworkResource(network.NetworkResource):
attributes = attr.AttributeCollection([
"org.openstack.network.ip_version",
"occi.network.address",
"occi.network.gateway",
"occi.network.allocation",
])
def __init__(self, title=None, summary=None,
id=None, vlan=None, label=None, state=None,
address=None, gateway=None, ip_version=None, allocation=None):
super(OSNetworkResource,
self).__init__(title=title,
summary=summary, id=id, vlan=vlan,
label=label, state=state,
mixins=[network.ip_network, OSNetwork()])
# subnet
self.attributes["org.openstack.network.ip_version"] = (
attr.MutableAttribute(
"org.openstack.network.ip_version", ip_version))
self.attributes["occi.network.address"] = (
attr.MutableAttribute(
"occi.network.address", address))
self.attributes["occi.network.gateway"] = (
attr.MutableAttribute(
"occi.network.gateway", gateway))
self.attributes["occi.network.allocation"] = (
attr.MutableAttribute(
"occi.network.allocation", allocation))
@property
def ip_version(self):
return self.attributes["org.openstack.network.ip_version"].value
@ip_version.setter
def ip_version(self, value):
self.attributes["org.openstack.network.ip_version"].value = value
@property
def address(self):
return self.attributes["occi.network.address"].value
@address.setter
def address(self, value):
self.attributes["occi.network.address"].value = value
@property
def gateway(self):
return self.attributes["occi.network.gateway"].value
@gateway.setter
def gateway(self, value):
self.attributes["occi.network.gateway"].value = value
@property
def allocation(self):
return self.attributes["occi.network.network.allocation"].value
@allocation.setter
def allocation(self, value):
self.attributes["occi.network.network.allocation"] = value

356
ooi/tests/controllers/test_network.py
View File

@ -1,6 +1,7 @@
# -*- 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
@ -19,102 +20,291 @@ import uuid
import mock
from ooi.api import helpers
from ooi.api import helpers_neutron
from ooi.api import network as network_api
from ooi import exception
from ooi.occi.core import collection
from ooi.occi.infrastructure import network
from ooi.occi.infrastructure import network as occi_network
from ooi.tests import base
from ooi.tests import fakes
from ooi.tests import fakes_network as fakes
class TestController(base.TestController):
class TestNetworkControllerNeutron(base.TestController):
def setUp(self):
super(TestController, self).setUp()
self.controller = network_api.Controller(mock.MagicMock(), None)
super(TestNetworkControllerNeutron, self).setUp()
self.controller = network_api.Controller(neutron_ooi_endpoint="ff")
@mock.patch("ooi.api.network._build_network")
@mock.patch.object(network_api.Controller, "_floating_index")
def test_index(self, m_float, m_build):
res = network.NetworkResource(title="foo",
id="foo",
state="active",
mixins=[network.ip_network])
res_fixed = network.NetworkResource(title="fixed",
id="fixed",
state="active",
mixins=[network.ip_network])
@mock.patch.object(helpers_neutron.OpenStackNeutron, "list_resources")
def test_index(self, m_index):
test_networks = [
fakes.networks[fakes.tenants["bar"]["id"]],
fakes.networks[fakes.tenants["foo"]["id"]]
]
req = fakes.create_req_test(None, None)
for nets in test_networks:
ooi_net = helpers_neutron.OpenStackNeutron._build_networks(nets)
m_index.return_value = ooi_net
result = self.controller.index(req)
expected = self.controller._get_network_resources(ooi_net)
self.assertEqual(result.resources.__len__(),
expected.__len__())
m_index.assert_called_with(req, 'networks')
m_float.return_value = [res]
m_build.return_value = res_fixed
ret = self.controller.index(None)
self.assertIsInstance(ret, collection.Collection)
self.assertEqual([res, res_fixed], ret.resources)
m_float.assert_called_with(None)
m_build.assert_called_with("fixed")
@mock.patch.object(helpers_neutron.OpenStackNeutron, "get_network_details")
def test_show(self, m_network):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for net in test_networks:
ret = self.controller.show(None, net["id"])
self.assertIsInstance(ret, occi_network.NetworkResource)
def test_build(self):
ret = network_api._build_network("foo")
self.assertIsInstance(ret, network.NetworkResource)
self.assertEqual("foo", ret.title)
self.assertEqual("foo", ret.id)
self.assertEqual([network.ip_network], ret.mixins)
@mock.patch.object(helpers_neutron.OpenStackNeutron, "create_network")
def test_create(self, m):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for test_net in test_networks:
parameters = {"occi.core.title": test_net["name"],
"org.openstack.network.ip_version": 4,
"occi.network.address": "0.0.0.0",
}
categories = {occi_network.NetworkResource.kind,
occi_network.ip_network}
req = fakes.create_req_test_occi(parameters, categories)
fake_net = fakes.fake_build_net(
parameters['occi.core.title'],
parameters['org.openstack.network.ip_version'],
parameters['occi.network.address']
)
m.return_value = fake_net
ret = self.controller.create(req)
net = ret.resources.pop()
self.assertIsInstance(net, occi_network.NetworkResource)
self.assertEqual(net.title, test_net['name'])
def test_build_with_prefix(self):
ret = network_api._build_network("foo", prefix="bar")
self.assertIsInstance(ret, network.NetworkResource)
self.assertEqual("foo", ret.title)
self.assertEqual("bar/foo", ret.id)
self.assertEqual([network.ip_network], ret.mixins)
@mock.patch.object(helpers_neutron.OpenStackNeutron, "create_resource")
def test_create_error(self, m):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
schema1 = occi_network.NetworkResource.kind.scheme
net = test_networks[0]
schemes = {schema1: net}
parameters = {"occi.core.title": "name",
}
req = fakes.create_req_test(parameters, schemes)
@mock.patch.object(helpers.OpenStackHelper, "get_floating_ip_pools")
def test_show(self, m_pools):
for tenant in fakes.tenants.values():
pools = fakes.pools[tenant["id"]]
if not pools:
continue
m_pools.return_value = pools
ret = self.controller.show(None, "floating")[0]
self.assertIsInstance(ret, network.NetworkResource)
self.assertEqual("floating", ret.title)
self.assertEqual("floating", ret.id)
self.assertEqual([network.ip_network], ret.mixins)
m_pools.assert_called_with(None)
self.assertRaises(exception.Invalid, self.controller.create, req)
@mock.patch.object(helpers.OpenStackHelper, "get_floating_ip_pools")
def test_show_not_found(self, m_pools):
@mock.patch.object(helpers_neutron.OpenStackNeutron, "create_network")
def test_create_no_ip_mixin(self, m):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for test_net in test_networks:
parameters = {"occi.core.title": test_net["name"],
"org.openstack.network.ip_version": 4,
"occi.network.address": "0.0.0.0",
}
categories = {occi_network.NetworkResource.kind}
req = fakes.create_req_test_occi(parameters, categories)
fake_net = fakes.fake_build_net(
parameters['occi.core.title'],
parameters['org.openstack.network.ip_version'],
parameters['occi.network.address']
)
m.return_value = fake_net
self.assertRaises(exception.OCCIMissingType,
self.controller.create, req)
@mock.patch.object(helpers_neutron.OpenStackNeutron, "delete_network")
def test_delete(self, m_network):
m_network.return_value = []
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for net in test_networks:
ret = self.controller.delete(None, net["id"])
self.assertEqual(ret, [])
self.assertEqual(ret.__len__(), 0)
def test_get_network_resources(self):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
subnet = fakes.subnets
for net in test_networks:
net["subnet_info"] = subnet[0]
ooi_net = (
helpers_neutron.OpenStackNeutron._build_networks(
test_networks))
ret = self.controller._get_network_resources(ooi_net)
self.assertIsInstance(ret, list)
self.assertIsNot(ret.__len__(), 0)
for net_ret in ret:
self.assertIsInstance(net_ret, occi_network.NetworkResource)
def test_filter_attributes(self):
parameters = {"occi.core.title": 'name',
"org.openstack.network.ip_version": '4',
"occi.network.address": '00001/24',
"occi.network.gateway": '00001',
}
categories = {occi_network.NetworkResource.kind}
req = fakes.create_req_test_occi(parameters, categories)
occi_scheme = {
"category": occi_network.NetworkResource.kind,
"optional_mixins": [
occi_network.ip_network,
]
}
ret = network_api.process_parameters(req, occi_scheme)
self.assertIsNotNone(ret)
self.assertEqual(parameters, ret)
def test_filter_attributes_empty(self):
categories = {occi_network.NetworkResource.kind}
req = fakes.create_req_test_occi(None, categories)
occi_scheme = {
"category": occi_network.NetworkResource.kind,
"optional_mixins": [
occi_network.ip_network,
]
}
attributes = network_api.process_parameters(req, occi_scheme)
self.assertIsNone(attributes)
def test_run_action_invalid(self):
tenant = fakes.tenants["foo"]
pools = fakes.pools[tenant["id"]]
m_pools.return_value = pools
self.assertRaises(exception.NetworkNotFound,
self.controller.show,
None, uuid.uuid4().hex)
req = self._build_req(tenant["id"], path="/network?action=start")
server_uuid = uuid.uuid4().hex
self.assertRaises(exception.InvalidAction,
self.controller.run_action,
req,
server_uuid,
None)
@mock.patch.object(helpers.OpenStackHelper, "get_floating_ip_pools")
def test_show_empty_floating(self, m_pools):
m_pools.return_value = []
self.assertRaises(exception.NetworkNotFound,
self.controller.show,
None, "floating")
m_pools.assert_called_with(None)
def test_run_action_up(self):
tenant = fakes.tenants["foo"]
req = self._build_req(tenant["id"], path="/network?action=up")
server_uuid = uuid.uuid4().hex
self.assertRaises(exception.NotImplemented,
self.controller.run_action,
req,
server_uuid,
None)
@mock.patch.object(helpers.OpenStackHelper, "get_floating_ip_pools")
def test_show_non_existent(self, m_pools):
m_pools.return_value = []
self.assertRaises(exception.NetworkNotFound,
self.controller.show,
None, None)
@mock.patch.object(helpers.OpenStackHelper, "get_floating_ip_pools")
def test_floating_ips(self, m_pools):
for tenant in fakes.tenants.values():
pools = fakes.pools[tenant["id"]]
m_pools.return_value = pools
ret = self.controller._floating_index(None)
if pools:
self.assertEqual(1, len(ret))
self.assertIsInstance(ret[0], network.NetworkResource)
self.assertEqual("floating", ret[0].title)
self.assertEqual("floating", ret[0].id)
else:
self.assertEqual(0, len(ret))
m_pools.assert_called_with(None)
class TestNetworkControllerNova(base.TestController):
def setUp(self):
super(TestNetworkControllerNova, self).setUp()
self.controller = network_api.Controller(None)
@mock.patch.object(helpers.OpenStackHelper, "list_networks")
def test_index(self, m_index):
test_networks = [
fakes.networks[fakes.tenants["bar"]["id"]],
fakes.networks[fakes.tenants["foo"]["id"]]
]
req = fakes.create_req_test(None, None)
for nets in test_networks:
ooi_net = helpers.OpenStackHelper._build_networks(nets)
m_index.return_value = ooi_net
result = self.controller.index(req)
expected = self.controller._get_network_resources(ooi_net)
self.assertEqual(result.resources.__len__(),
expected.__len__())
m_index.assert_called_with(req)
@mock.patch.object(helpers.OpenStackHelper, "get_network_details")
def test_show(self, m_network):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for net in test_networks:
ret = self.controller.show(None, net["id"])
self.assertIsInstance(ret, occi_network.NetworkResource)
@mock.patch.object(helpers.OpenStackHelper, "create_network")
def test_create(self, m):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for test_net in test_networks:
parameters = {"occi.core.title": test_net["name"],
"org.openstack.network.ip_version": 4,
"occi.network.address": "0.0.0.0",
}
categories = {occi_network.NetworkResource.kind,
occi_network.ip_network}
req = fakes.create_req_test_occi(parameters, categories)
fake_net = fakes.fake_build_net(
parameters['occi.core.title'],
parameters['org.openstack.network.ip_version'],
parameters['occi.network.address']
)
m.return_value = fake_net
ret = self.controller.create(req)
net = ret.resources.pop()
self.assertIsInstance(net, occi_network.NetworkResource)
self.assertEqual(net.title, test_net['name'])
@mock.patch.object(helpers.OpenStackHelper, "create_network")
def test_create_error(self, m):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
schema1 = occi_network.NetworkResource.kind.scheme
net = test_networks[0]
schemes = {schema1: net}
parameters = {"occi.core.title": "name",
}
req = fakes.create_req_test(parameters, schemes)
self.assertRaises(exception.Invalid, self.controller.create, req)
@mock.patch.object(helpers.OpenStackHelper, "create_network")
def test_create_no_ip_mixin(self, m):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for test_net in test_networks:
parameters = {"occi.core.title": test_net["name"],
"org.openstack.network.ip_version": 4,
"occi.network.address": "0.0.0.0",
}
categories = {occi_network.NetworkResource.kind}
req = fakes.create_req_test_occi(parameters, categories)
fake_net = fakes.fake_build_net(
parameters['occi.core.title'],
parameters['org.openstack.network.ip_version'],
parameters['occi.network.address']
)
m.return_value = fake_net
self.assertRaises(exception.OCCIMissingType,
self.controller.create, req)
@mock.patch.object(helpers.OpenStackHelper, "delete_network")
def test_delete(self, m_network):
m_network.return_value = []
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
for net in test_networks:
ret = self.controller.delete(None, net["id"])
self.assertEqual(ret, [])
self.assertEqual(ret.__len__(), 0)
def test_get_network_resources(self):
test_networks = fakes.networks[fakes.tenants["foo"]["id"]]
subnet = fakes.subnets
for net in test_networks:
net["subnet_info"] = subnet[0]
ooi_net = (
helpers_neutron.OpenStackNeutron._build_networks(
test_networks))
ret = self.controller._get_network_resources(ooi_net)
self.assertIsInstance(ret, list)
self.assertIsNot(ret.__len__(), 0)
for net_ret in ret:
self.assertIsInstance(net_ret, occi_network.NetworkResource)
def test_run_action_invalid(self):
tenant = fakes.tenants["foo"]
req = self._build_req(tenant["id"], path="/network?action=start")
server_uuid = uuid.uuid4().hex
self.assertRaises(exception.InvalidAction,
self.controller.run_action,
req,
server_uuid,
None)
def test_run_action_up(self):
tenant = fakes.tenants["foo"]
req = self._build_req(tenant["id"], path="/network?action=up")
server_uuid = uuid.uuid4().hex
self.assertRaises(exception.NotImplemented,
self.controller.run_action,
req,
server_uuid,
None)

279
ooi/tests/middleware/test_network_controller.py
View File

@ -1,6 +1,7 @@
# -*- 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
@ -14,127 +15,229 @@
# License for the specific language governing permissions and limitations
# under the License.
import uuid
import mock
from ooi.tests import fakes
from ooi.api import helpers_neutron
from ooi.api import network
from ooi.occi.core import collection
from ooi.tests import fakes_network as fakes
from ooi.tests.middleware import test_middleware
from ooi import utils
from ooi import wsgi
def build_occi_network(pool_name):
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/"' % fakes.application_url)
'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"')
attrs = [
'occi.core.title="%s"' % pool_name,
'occi.network.state="active"',
'occi.core.id="%s"' % pool_name,
]
'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, pool_name))
(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, pool_name))
(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 l in links:
result.append(("Link", l))
for a in attrs:
result.append(("X-OCCI-Attribute", a))
for l in links:
result.append(("Link", l))
return result
class TestNetworkController(test_middleware.TestMiddleware):
"""Test OCCI network controller."""
def create_occi_results(data):
return network.Controller(None)._get_network_resources(data)
def test_list_pools_empty(self):
class TestMiddlewareNeutron(test_middleware.TestMiddleware):
"""OCCI middleware test for Neutron middleware.
According to the OCCI HTTP rendering, no Accept header
means text/plain.
"""
def setUp(self):
super(TestMiddlewareNeutron, self).setUp()
self.accept = self.content_type = None
self.application_url = fakes.application_url
self.app = wsgi.OCCIMiddleware(None)
class TestNetworkController(TestMiddlewareNeutron):
"""Test OCCI compute controller."""
def setUp(self):
super(TestNetworkController, self).setUp()
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(network.Controller, "index")
def test_list_networks_empty(self, m):
tenant = fakes.tenants["bar"]
app = self.get_app()
headers = {
'Category': 'network; scheme="http://schema#";class="kind";',
'X_OCCI_Attribute': 'project=%s' % tenant["id"],
}
url = "/network"
req = self._build_req(path=url,
tenant_id='X',
method="GET",
headers=headers, content_type="text/occi")
m.return_value = collection.Collection(
create_occi_results(fakes.networks[tenant['id']]))
resp = req.get_response(self.app)
self.assertEqual(204, resp.status_code)
expected_result = ""
self.assertExpectedResult(expected_result, resp)
self.assertDefaults(resp)
for url in ("/network", "/network/"):
req = self._build_req(url, tenant["id"], method="GET")
@mock.patch.object(network.Controller, "index")
def test_list_networks(self, m):
tenant = fakes.tenants["foo"]
ooi_net = helpers_neutron.OpenStackNeutron._build_networks(
fakes.networks[tenant['id']]
)
m.return_value = collection.Collection(
create_occi_results(ooi_net))
req = self._build_req(path="/network",
tenant_id='X', method="GET")
resp = req.get_response(self.app)
req.environ["HTTP_X_PROJECT_ID"] = tenant["id"]
resp = req.get_response(app)
resp = req.get_response(app)
expected = [
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/fixed"))
]
self.assertDefaults(resp)
self.assertExpectedResult(expected, resp)
self.assertEqual(200, resp.status_code)
def test_list_pools(self):
tenant = fakes.tenants["foo"]
app = self.get_app()
for url in ("/network", "/network/"):
req = self._build_req(url, tenant["id"], method="GET")
resp = req.get_response(app)
self.assertEqual(200, resp.status_code)
expected = [
("X-OCCI-Location",
utils.join_url(self.application_url + "/", "network/fixed"))
]
if fakes.pools[tenant["id"]]:
expected.append(
("X-OCCI-Location",
utils.join_url(self.application_url + "/",
"network/floating"))
)
self.assertDefaults(resp)
self.assertExpectedResult(expected, resp)
def test_show_floating_pool(self):
tenant = fakes.tenants["foo"]
app = self.get_app()
for pool in fakes.pools[tenant["id"]]:
req = self._build_req("/network/floating", tenant["id"],
method="GET")
resp = req.get_response(app)
expected = build_occi_network("floating")
self.assertDefaults(resp)
self.assertExpectedResult(expected, resp)
self.assertEqual(200, resp.status_code)
def test_show_fixed(self):
tenant = fakes.tenants["foo"]
app = self.get_app()
req = self._build_req("/network/fixed", tenant["id"], method="GET")
resp = req.get_response(app)
expected = build_occi_network("fixed")
utils.join_url(self.application_url + "/",
"network/%s" % s["id"]))
)
self.assertDefaults(resp)
self.assertExpectedResult(expected, resp)
self.assertEqual(200, resp.status_code)
def test_pool_not_found(self):
@mock.patch.object(network.Controller, "create")
def test_create(self, m):
tenant = fakes.tenants["foo"]
headers = {
'Category': 'network; scheme="http://schema#";class="kind",' +
'mixinID;'
'scheme="http://schemas.openstack.org/template/os#";'
' class=mixin',
'X_Occi_Attribute': 'project=%s' % tenant["id"],
}
req = self._build_req(path="/network",
tenant_id='X',
method="POST",
headers=headers)
fake_net = fakes.fake_network_occi(
fakes.networks[tenant['id']]
)[0]
m.return_value = collection.Collection([fake_net])
resp = req.get_response(self.app)
self.assertEqual(200, resp.status_code)
expected = [("X-OCCI-Location",
utils.join_url(self.application_url + "/",
"network/%s" % fake_net.id))]
self.assertExpectedResult(expected, resp)
@mock.patch.object(network.Controller, "show")
def test_show_networks(self, m):
tenant = fakes.tenants["foo"]
app = self.get_app()
req = self._build_req("/network/%s" % uuid.uuid4().hex,
tenant["id"], method="GET")
resp = req.get_response(app)
self.assertEqual(404, resp.status_code)
for n in fakes.networks[tenant["id"]]:
ooi_net = helpers_neutron.OpenStackNeutron._build_networks([n])[0]
m.return_value = create_occi_results([ooi_net])[0]
req = self._build_req(path="/network/%s" % n["id"],
tenant_id='X',
method="GET")
resp = req.get_response(self.app)
expected = build_occi_network(n)
self.assertEqual(200, resp.status_code)
self.assertDefaults(resp)
self.assertExpectedResult(expected, resp)
@mock.patch.object(network.Controller, "delete")
def test_delete_networks(self, m):
tenant = fakes.tenants["foo"]
for n in fakes.networks[tenant["id"]]:
m.return_value = create_occi_results([])
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 NetworkControllerTextPlain(test_middleware.TestMiddlewareTextPlain,