Implement OpenFlow Agent mechanism driver

This adds ML2 mechanism driver controlling OpenFlow switches
and an agent using Ryu as OpenFlow Python library.
- An agent acts as an OpenFlow controller on each compute nodes.
- OpenFlow 1.3 (vendor agnostic unlike OVS extensions).

Implements: blueprint ryu-ml2-driver
Change-Id: I6a8168d24f911996639179d91c4da49151751057
This commit is contained in:
fumihiko kakuma 2014-01-29 10:54:12 +09:00
parent 2964b8483a
commit f8e49358da
19 changed files with 2470 additions and 54 deletions

View File

@ -0,0 +1,13 @@
# Defines configuration options specific to the OpenFlow Agent Mechanism Driver
[ovs]
# Please refer to configuration options to the OpenvSwitch
[agent]
# (IntOpt) Number of seconds to retry acquiring an Open vSwitch datapath.
# This is an optional parameter, default value is 60 seconds.
#
# get_datapath_retry_times =
# Example: get_datapath_retry_times = 30
# Please refer to configuration options to the OpenvSwitch else the above.

View File

@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import distutils.version as dist_version
import re
from oslo.config import cfg
@ -106,6 +107,27 @@ class OVSBridge(BaseOVS):
self.defer_apply_flows = False
self.deferred_flows = {'add': '', 'mod': '', 'del': ''}
def set_controller(self, controller_names):
vsctl_command = ['--', 'set-controller', self.br_name]
vsctl_command.extend(controller_names)
self.run_vsctl(vsctl_command, check_error=True)
def del_controller(self):
self.run_vsctl(['--', 'del-controller', self.br_name],
check_error=True)
def get_controller(self):
res = self.run_vsctl(['--', 'get-controller', self.br_name],
check_error=True)
if res:
return res.strip().split('\n')
return res
def set_protocols(self, protocols):
self.run_vsctl(['--', 'set', 'bridge', self.br_name,
"protocols=%s" % protocols],
check_error=True)
def create(self):
self.add_bridge(self.br_name)
@ -401,7 +423,7 @@ class OVSBridge(BaseOVS):
ofport = data[ofport_idx]
# ofport must be integer otherwise return None
if not isinstance(ofport, int) or ofport == -1:
LOG.warn(_("ofport: %(ofport)s for VIF: %(vif)s is not a"
LOG.warn(_("ofport: %(ofport)s for VIF: %(vif)s is not a "
"positive integer"), {'ofport': ofport,
'vif': port_id})
return
@ -483,3 +505,44 @@ def get_bridge_external_bridge_id(root_helper, bridge):
except Exception:
LOG.exception(_("Bridge %s not found."), bridge)
return None
def _compare_installed_and_required_version(
installed_version, required_version, check_type, version_type):
if installed_version:
if dist_version.StrictVersion(
installed_version) < dist_version.StrictVersion(
required_version):
msg = (_('Failed %(ctype)s version check for Open '
'vSwitch with %(vtype)s support. To use '
'%(vtype)s tunnels with OVS, please ensure '
'the OVS version is %(required)s or newer!') %
{'ctype': check_type, 'vtype': version_type,
'required': required_version})
raise SystemError(msg)
else:
msg = (_('Unable to determine %(ctype)s version for Open '
'vSwitch with %(vtype)s support. To use '
'%(vtype)s tunnels with OVS, please ensure '
'that the version is %(required)s or newer!') %
{'ctype': check_type, 'vtype': version_type,
'required': required_version})
raise SystemError(msg)
def check_ovs_vxlan_version(root_helper):
min_required_version = constants.MINIMUM_OVS_VXLAN_VERSION
installed_klm_version = get_installed_ovs_klm_version()
installed_usr_version = get_installed_ovs_usr_version(root_helper)
LOG.debug(_("Checking OVS version for VXLAN support "
"installed klm version is %s "), installed_klm_version)
LOG.debug(_("Checking OVS version for VXLAN support "
"installed usr version is %s"), installed_usr_version)
# First check the userspace version
_compare_installed_and_required_version(installed_usr_version,
min_required_version,
'userspace', 'VXLAN')
# Now check the kernel version
_compare_installed_and_required_version(installed_klm_version,
min_required_version,
'kernel', 'VXLAN')

View File

@ -67,6 +67,7 @@ AGENT_TYPE_OVS = 'Open vSwitch agent'
AGENT_TYPE_LINUXBRIDGE = 'Linux bridge agent'
AGENT_TYPE_HYPERV = 'HyperV agent'
AGENT_TYPE_NEC = 'NEC plugin agent'
AGENT_TYPE_OFA = 'OFA driver agent'
AGENT_TYPE_L3 = 'L3 agent'
AGENT_TYPE_LOADBALANCER = 'Loadbalancer agent'
AGENT_TYPE_MLNX = 'Mellanox plugin agent'

View File

@ -0,0 +1,60 @@
# Copyright (C) 2014 VA Linux Systems Japan K.K.
# Based on openvswitch mechanism driver.
#
# Copyright (c) 2013 OpenStack Foundation
# All Rights Reserved.
#
# 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.
# @author: Fumihiko Kakuma, VA Linux Systems Japan K.K.
from neutron.common import constants
from neutron.extensions import portbindings
from neutron.openstack.common import log
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import mech_agent
LOG = log.getLogger(__name__)
class OfagentMechanismDriver(mech_agent.SimpleAgentMechanismDriverBase):
"""Attach to networks using ofagent L2 agent.
The OfagentMechanismDriver integrates the ml2 plugin with the
ofagent L2 agent. Port binding with this driver requires the
ofagent agent to be running on the port's host, and that agent
to have connectivity to at least one segment of the port's
network.
"""
def __init__(self):
super(OfagentMechanismDriver, self).__init__(
constants.AGENT_TYPE_OFA,
portbindings.VIF_TYPE_OVS,
{portbindings.CAP_PORT_FILTER: True})
def check_segment_for_agent(self, segment, agent):
mappings = agent['configurations'].get('bridge_mappings', {})
tunnel_types = agent['configurations'].get('tunnel_types', [])
LOG.debug(_("Checking segment: %(segment)s "
"for mappings: %(mappings)s "
"with tunnel_types: %(tunnel_types)s"),
{'segment': segment, 'mappings': mappings,
'tunnel_types': tunnel_types})
network_type = segment[api.NETWORK_TYPE]
return (
network_type == p_const.TYPE_LOCAL or
network_type in tunnel_types or
(network_type in [p_const.TYPE_FLAT, p_const.TYPE_VLAN] and
segment[api.PHYSICAL_NETWORK] in mappings)
)

View File

@ -0,0 +1,21 @@
This directory includes agent for OpenFlow Agent mechanism driver.
# -- Installation
For how to install/set up ML2 mechanism driver for OpenFlow Agent, please refer to
https://github.com/osrg/ryu/wiki/OpenStack
# -- Ryu General
For general Ryu stuff, please refer to
http://www.osrg.net/ryu/
Ryu is available at github
git://github.com/osrg/ryu.git
https://github.com/osrg/ryu
The mailing is at
ryu-devel@lists.sourceforge.net
https://lists.sourceforge.net/lists/listinfo/ryu-devel
Enjoy!

View File

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,33 @@
# Copyright (C) 2014 VA Linux Systems Japan K.K.
#
# 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.
# @author: Fumihiko Kakuma, VA Linux Systems Japan K.K.
from oslo.config import cfg
from neutron.agent.common import config
from neutron.plugins.openvswitch.common import config as ovs_config
agent_opts = [
cfg.IntOpt('get_datapath_retry_times', default=60,
help=_("Number of seconds to retry acquiring "
"an Open vSwitch datapath")),
]
cfg.CONF.register_opts(ovs_config.ovs_opts, 'OVS')
cfg.CONF.register_opts(ovs_config.agent_opts, 'AGENT')
cfg.CONF.register_opts(agent_opts, 'AGENT')
config.register_agent_state_opts_helper(cfg.CONF)
config.register_root_helper(cfg.CONF)

View File

@ -14,7 +14,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import distutils.version as dist_version
import signal
import sys
import time
@ -227,8 +226,11 @@ class OVSNeutronAgent(sg_rpc.SecurityGroupAgentRpcCallbackMixin,
def _check_ovs_version(self):
if p_const.TYPE_VXLAN in self.tunnel_types:
check_ovs_version(constants.MINIMUM_OVS_VXLAN_VERSION,
self.root_helper)
try:
ovs_lib.check_ovs_vxlan_version(self.root_helper)
except SystemError:
LOG.exception(_("Agent terminated"))
raise SystemExit(1)
def _report_state(self):
# How many devices are likely used by a VM
@ -1250,44 +1252,6 @@ def handle_sigterm(signum, frame):
sys.exit(1)
def check_ovs_version(min_required_version, root_helper):
LOG.debug(_("Checking OVS version for VXLAN support"))
installed_klm_version = ovs_lib.get_installed_ovs_klm_version()
installed_usr_version = ovs_lib.get_installed_ovs_usr_version(root_helper)
# First check the userspace version
if installed_usr_version:
if dist_version.StrictVersion(
installed_usr_version) < dist_version.StrictVersion(
min_required_version):
LOG.error(_('Failed userspace version check for Open '
'vSwitch with VXLAN support. To use '
'VXLAN tunnels with OVS, please ensure '
'the OVS version is %s '
'or newer!'), min_required_version)
sys.exit(1)
# Now check the kernel version
if installed_klm_version:
if dist_version.StrictVersion(
installed_klm_version) < dist_version.StrictVersion(
min_required_version):
LOG.error(_('Failed kernel version check for Open '
'vSwitch with VXLAN support. To use '
'VXLAN tunnels with OVS, please ensure '
'the OVS version is %s or newer!'),
min_required_version)
raise SystemExit(1)
else:
LOG.warning(_('Cannot determine kernel Open vSwitch version, '
'please ensure your Open vSwitch kernel module '
'is at least version %s to support VXLAN '
'tunnels.'), min_required_version)
else:
LOG.warning(_('Unable to determine Open vSwitch version. Please '
'ensure that its version is %s or newer to use VXLAN '
'tunnels with OVS.'), min_required_version)
raise SystemExit(1)
def create_agent_config_map(config):
"""Create a map of agent config parameters.

View File

@ -0,0 +1,74 @@
# Copyright (c) 2014 OpenStack Foundation
# All Rights Reserved.
#
# 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.
from neutron.common import constants
from neutron.extensions import portbindings
from neutron.plugins.ml2.drivers import mech_ofagent
from neutron.tests.unit.ml2 import _test_mech_agent as base
class OfagentMechanismBaseTestCase(base.AgentMechanismBaseTestCase):
VIF_TYPE = portbindings.VIF_TYPE_OVS
CAP_PORT_FILTER = True
AGENT_TYPE = constants.AGENT_TYPE_OFA
GOOD_MAPPINGS = {'fake_physical_network': 'fake_bridge'}
GOOD_TUNNEL_TYPES = ['gre', 'vxlan']
GOOD_CONFIGS = {'bridge_mappings': GOOD_MAPPINGS,
'tunnel_types': GOOD_TUNNEL_TYPES}
BAD_MAPPINGS = {'wrong_physical_network': 'wrong_bridge'}
BAD_TUNNEL_TYPES = ['bad_tunnel_type']
BAD_CONFIGS = {'bridge_mappings': BAD_MAPPINGS,
'tunnel_types': BAD_TUNNEL_TYPES}
AGENTS = [{'alive': True,
'configurations': GOOD_CONFIGS}]
AGENTS_DEAD = [{'alive': False,
'configurations': GOOD_CONFIGS}]
AGENTS_BAD = [{'alive': False,
'configurations': GOOD_CONFIGS},
{'alive': True,
'configurations': BAD_CONFIGS}]
def setUp(self):
super(OfagentMechanismBaseTestCase, self).setUp()
self.driver = mech_ofagent.OfagentMechanismDriver()
self.driver.initialize()
class OfagentMechanismGenericTestCase(OfagentMechanismBaseTestCase,
base.AgentMechanismGenericTestCase):
pass
class OfagentMechanismLocalTestCase(OfagentMechanismBaseTestCase,
base.AgentMechanismLocalTestCase):
pass
class OfagentMechanismFlatTestCase(OfagentMechanismBaseTestCase,
base.AgentMechanismFlatTestCase):
pass
class OfagentMechanismVlanTestCase(OfagentMechanismBaseTestCase,
base.AgentMechanismVlanTestCase):
pass
class OfagentMechanismGreTestCase(OfagentMechanismBaseTestCase,
base.AgentMechanismGreTestCase):
pass

View File

View File

@ -0,0 +1,43 @@
# Copyright (C) 2014 VA Linux Systems Japan K.K.
#
# 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.
# @author: Fumihiko Kakuma, VA Linux Systems Japan K.K.
import mock
def patch_fake_oflib_of():
ryu_mod = mock.Mock()
ryu_base_mod = ryu_mod.base
ryu_lib_mod = ryu_mod.lib
ryu_lib_hub = ryu_lib_mod.hub
ryu_ofproto_mod = ryu_mod.ofproto
ryu_ofproto_of13 = ryu_ofproto_mod.ofproto_v1_3
ryu_ofproto_of13.OFPTT_ALL = 0xff
ryu_ofproto_of13.OFPG_ANY = 0xffffffff
ryu_ofproto_of13.OFPP_ANY = 0xffffffff
ryu_ofproto_of13.OFPFC_ADD = 0
ryu_ofproto_of13.OFPFC_DELETE = 3
ryu_app_mod = ryu_mod.app
ryu_app_ofctl_mod = ryu_app_mod.ofctl
ryu_ofctl_api = ryu_app_ofctl_mod.api
return mock.patch.dict('sys.modules',
{'ryu': ryu_mod,
'ryu.base': ryu_base_mod,
'ryu.lib': ryu_lib_mod,
'ryu.lib.hub': ryu_lib_hub,
'ryu.ofproto': ryu_ofproto_mod,
'ryu.ofproto.ofproto_v1_3': ryu_ofproto_of13,
'ryu.app': ryu_app_mod,
'ryu.app.ofctl': ryu_app_ofctl_mod,
'ryu.app.ofctl.api': ryu_ofctl_api})

View File

@ -0,0 +1,25 @@
# Copyright (C) 2014 VA Linux Systems Japan K.K.
#
# 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.
# @author: Fumihiko Kakuma, VA Linux Systems Japan K.K.
from oslo.config import cfg
from neutron.plugins.ofagent.common import config # noqa
from neutron.tests import base
class ConfigurationTest(base.BaseTestCase):
"""Configuration file Tests."""
def test_ml2_defaults(self):
self.assertEqual(60, cfg.CONF.AGENT.get_datapath_retry_times)

View File

@ -0,0 +1,660 @@
# Copyright (C) 2014 VA Linux Systems Japan K.K.
# Based on test for openvswitch agent(test_ovs_neutron_agent.py).
#
# Copyright (c) 2012 OpenStack Foundation.
#
# 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.
# @author: Fumihiko Kakuma, VA Linux Systems Japan K.K.
import contextlib
import mock
from oslo.config import cfg
import testtools
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
from neutron.openstack.common import importutils
from neutron.openstack.common.rpc import common as rpc_common
from neutron.plugins.common import constants as p_const
from neutron.plugins.openvswitch.common import constants
from neutron.tests import base
from neutron.tests.unit.ofagent import fake_oflib
NOTIFIER = ('neutron.plugins.ml2.rpc.AgentNotifierApi')
class OFAAgentTestCase(base.BaseTestCase):
_AGENT_NAME = 'neutron.plugins.ofagent.agent.ofa_neutron_agent'
def setUp(self):
super(OFAAgentTestCase, self).setUp()
self.addCleanup(mock.patch.stopall)
self.fake_oflib_of = fake_oflib.patch_fake_oflib_of().start()
self.mod_agent = importutils.import_module(self._AGENT_NAME)
self.ryuapp = mock.Mock()
cfg.CONF.register_cli_opts([
cfg.StrOpt('ofp-listen-host', default='',
help='openflow listen host'),
cfg.IntOpt('ofp-tcp-listen-port', default=6633,
help='openflow tcp listen port')
])
cfg.CONF.set_override('root_helper', 'fake_helper', group='AGENT')
class CreateAgentConfigMap(OFAAgentTestCase):
def test_create_agent_config_map_succeeds(self):
self.assertTrue(self.mod_agent.create_agent_config_map(cfg.CONF))
def test_create_agent_config_map_fails_for_invalid_tunnel_config(self):
# An ip address is required for tunneling but there is no default,
# verify this for both gre and vxlan tunnels.
cfg.CONF.set_override('tunnel_types', [p_const.TYPE_GRE],
group='AGENT')
with testtools.ExpectedException(ValueError):
self.mod_agent.create_agent_config_map(cfg.CONF)
cfg.CONF.set_override('tunnel_types', [p_const.TYPE_VXLAN],
group='AGENT')
with testtools.ExpectedException(ValueError):
self.mod_agent.create_agent_config_map(cfg.CONF)
def test_create_agent_config_map_enable_tunneling(self):
# Verify setting only enable_tunneling will default tunnel_type to GRE
cfg.CONF.set_override('tunnel_types', None, group='AGENT')
cfg.CONF.set_override('enable_tunneling', True, group='OVS')
cfg.CONF.set_override('local_ip', '10.10.10.10', group='OVS')
cfgmap = self.mod_agent.create_agent_config_map(cfg.CONF)
self.assertEqual(cfgmap['tunnel_types'], [p_const.TYPE_GRE])
def test_create_agent_config_map_fails_no_local_ip(self):
# An ip address is required for tunneling but there is no default
cfg.CONF.set_override('enable_tunneling', True, group='OVS')
with testtools.ExpectedException(ValueError):
self.mod_agent.create_agent_config_map(cfg.CONF)
def test_create_agent_config_map_fails_for_invalid_tunnel_type(self):
cfg.CONF.set_override('tunnel_types', ['foobar'], group='AGENT')
with testtools.ExpectedException(ValueError):
self.mod_agent.create_agent_config_map(cfg.CONF)
def test_create_agent_config_map_multiple_tunnel_types(self):
cfg.CONF.set_override('local_ip', '10.10.10.10', group='OVS')
cfg.CONF.set_override('tunnel_types', [p_const.TYPE_GRE,
p_const.TYPE_VXLAN], group='AGENT')
cfgmap = self.mod_agent.create_agent_config_map(cfg.CONF)
self.assertEqual(cfgmap['tunnel_types'],
[p_const.TYPE_GRE, p_const.TYPE_VXLAN])
class TestOFANeutronAgentOVSBridge(OFAAgentTestCase):
def setUp(self):
super(TestOFANeutronAgentOVSBridge, self).setUp()
self.br_name = 'bridge1'
self.root_helper = 'fake_helper'
self.ovs = self.mod_agent.OVSBridge(
self.br_name, self.root_helper, self.ryuapp)
def test_find_datapath_id(self):
with mock.patch.object(self.ovs, 'get_datapath_id',
return_value='12345'):
self.ovs.find_datapath_id()
self.assertEqual(self.ovs.datapath_id, '12345')
def _fake_get_datapath(self, app, datapath_id):
if self.ovs.retry_count >= 2:
datapath = mock.Mock()
datapath.ofproto_parser = mock.Mock()
return datapath
self.ovs.retry_count += 1
return None
def test_get_datapath_normal(self):
self.ovs.retry_count = 0
with mock.patch.object(self.mod_agent.ryu_api, 'get_datapath',
new=self._fake_get_datapath):
self.ovs.datapath_id = '0x64'
self.ovs.get_datapath(retry_max=4)
self.assertEqual(self.ovs.retry_count, 2)
def test_get_datapath_retry_out_by_default_time(self):
cfg.CONF.set_override('get_datapath_retry_times', 3, group='AGENT')
with mock.patch.object(self.mod_agent.ryu_api, 'get_datapath',
return_value=None) as mock_get_datapath:
with testtools.ExpectedException(SystemExit):
self.ovs.datapath_id = '0x64'
self.ovs.get_datapath(retry_max=3)
self.assertEqual(mock_get_datapath.call_count, 3)
def test_get_datapath_retry_out_by_specified_time(self):
with mock.patch.object(self.mod_agent.ryu_api, 'get_datapath',
return_value=None) as mock_get_datapath:
with testtools.ExpectedException(SystemExit):
self.ovs.datapath_id = '0x64'
self.ovs.get_datapath(retry_max=2)
self.assertEqual(mock_get_datapath.call_count, 2)
def test_setup_ofp_default_par(self):
with contextlib.nested(
mock.patch.object(self.ovs, 'set_protocols'),
mock.patch.object(self.ovs, 'set_controller'),
mock.patch.object(self.ovs, 'find_datapath_id'),
mock.patch.object(self.ovs, 'get_datapath'),
) as (mock_set_protocols, mock_set_controller,
mock_find_datapath_id, mock_get_datapath):
self.ovs.setup_ofp()
mock_set_protocols.assert_called_with('OpenFlow13')
mock_set_controller.assert_called_with(['tcp:127.0.0.1:6633'])
mock_get_datapath.assert_called_with(
cfg.CONF.AGENT.get_datapath_retry_times)
self.assertEqual(mock_find_datapath_id.call_count, 1)
def test_setup_ofp_specify_par(self):
controller_names = ['tcp:192.168.10.10:1234', 'tcp:172.17.16.20:5555']
with contextlib.nested(
mock.patch.object(self.ovs, 'set_protocols'),
mock.patch.object(self.ovs, 'set_controller'),
mock.patch.object(self.ovs, 'find_datapath_id'),
mock.patch.object(self.ovs, 'get_datapath'),
) as (mock_set_protocols, mock_set_controller,
mock_find_datapath_id, mock_get_datapath):
self.ovs.setup_ofp(controller_names=controller_names,
protocols='OpenFlow133',
retry_max=11)
mock_set_protocols.assert_called_with('OpenFlow133')
mock_set_controller.assert_called_with(controller_names)
mock_get_datapath.assert_called_with(11)
self.assertEqual(mock_find_datapath_id.call_count, 1)
def test_setup_ofp_with_except(self):
with contextlib.nested(
mock.patch.object(self.ovs, 'set_protocols',
side_effect=RuntimeError),
mock.patch.object(self.ovs, 'set_controller'),
mock.patch.object(self.ovs, 'find_datapath_id'),
mock.patch.object(self.ovs, 'get_datapath'),
) as (mock_set_protocols, mock_set_controller,
mock_find_datapath_id, mock_get_datapath):
with testtools.ExpectedException(SystemExit):
self.ovs.setup_ofp()
class TestOFANeutronAgent(OFAAgentTestCase):
def setUp(self):
super(TestOFANeutronAgent, self).setUp()
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
notifier_cls.return_value = self.notifier
# Avoid rpc initialization for unit tests
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
kwargs = self.mod_agent.create_agent_config_map(cfg.CONF)
class MockFixedIntervalLoopingCall(object):
def __init__(self, f):
self.f = f
def start(self, interval=0):
self.f()
with contextlib.nested(
mock.patch.object(self.mod_agent.OFANeutronAgent,
'setup_integration_br',
return_value=mock.Mock()),
mock.patch.object(self.mod_agent.OFANeutronAgent,
'setup_ancillary_bridges',
return_value=[]),
mock.patch.object(self.mod_agent.OVSBridge,
'get_local_port_mac',
return_value='00:00:00:00:00:01'),
mock.patch('neutron.agent.linux.utils.get_interface_mac',
return_value='00:00:00:00:00:01'),
mock.patch('neutron.openstack.common.loopingcall.'
'FixedIntervalLoopingCall',
new=MockFixedIntervalLoopingCall)):
self.agent = self.mod_agent.OFANeutronAgent(self.ryuapp, **kwargs)
self.agent.tun_br = mock.Mock()
self.datapath = mock.Mock()
self.ofparser = mock.Mock()
self.datapath.ofparser = self.ofparser
self.ofparser.OFPMatch = mock.Mock()
self.ofparser.OFPMatch.return_value = mock.Mock()
self.ofparser.OFPFlowMod = mock.Mock()
self.ofparser.OFPFlowMod.return_value = mock.Mock()
self.agent.int_br.ofparser = self.ofparser
self.agent.sg_agent = mock.Mock()
def _mock_port_bound(self, ofport=None):
port = mock.Mock()
port.ofport = ofport
net_uuid = 'my-net-uuid'
with mock.patch.object(self.mod_agent.OVSBridge,
'set_db_attribute',
return_value=True):
with mock.patch.object(self.agent,
'ryu_send_msg') as ryu_send_msg_func:
self.agent.port_bound(port, net_uuid, 'local', None, None)
self.assertEqual(ryu_send_msg_func.called, ofport != -1)
def test_port_bound_deletes_flows_for_valid_ofport(self):
self._mock_port_bound(ofport=1)
def test_port_bound_ignores_flows_for_invalid_ofport(self):
self._mock_port_bound(ofport=-1)
def test_port_dead(self):
with mock.patch.object(self.mod_agent.OVSBridge,
'set_db_attribute',
return_value=True):
with mock.patch.object(self.agent,
'ryu_send_msg') as ryu_send_msg_func:
port = mock.Mock()
port.ofport = 2
self.agent.port_dead(port)
self.assertTrue(ryu_send_msg_func.called)
def mock_update_ports(self, vif_port_set=None, registered_ports=None):
with mock.patch.object(self.agent.int_br, 'get_vif_port_set',
return_value=vif_port_set):
return self.agent.update_ports(registered_ports)
def test_update_ports_returns_none_for_unchanged_ports(self):
self.assertIsNone(self.mock_update_ports())
def test_update_ports_returns_port_changes(self):
vif_port_set = set([1, 3])
registered_ports = set([1, 2])
expected = dict(current=vif_port_set, added=set([3]), removed=set([2]))
actual = self.mock_update_ports(vif_port_set, registered_ports)
self.assertEqual(expected, actual)
def test_treat_devices_added_returns_true_for_missing_device(self):
with mock.patch.object(self.agent.plugin_rpc, 'get_device_details',
side_effect=Exception()):
self.assertTrue(self.agent.treat_devices_added([{}]))
def _mock_treat_devices_added(self, details, port, func_name):
"""Mock treat devices added.
:param details: the details to return for the device
:param port: the port that get_vif_port_by_id should return
:param func_name: the function that should be called
:returns: whether the named function was called
"""
with contextlib.nested(
mock.patch.object(self.agent.plugin_rpc, 'get_device_details',
return_value=details),
mock.patch.object(self.agent.int_br, 'get_vif_port_by_id',
return_value=port),
mock.patch.object(self.agent.plugin_rpc, 'update_device_up'),
mock.patch.object(self.agent, func_name)
) as (get_dev_fn, get_vif_func, upd_dev_up, func):
self.assertFalse(self.agent.treat_devices_added([{}]))
return func.called
def test_treat_devices_added_ignores_invalid_ofport(self):
port = mock.Mock()
port.ofport = -1
self.assertFalse(self._mock_treat_devices_added(mock.MagicMock(), port,
'port_dead'))
def test_treat_devices_added_marks_unknown_port_as_dead(self):
port = mock.Mock()
port.ofport = 1
self.assertTrue(self._mock_treat_devices_added(mock.MagicMock(), port,
'port_dead'))
def test_treat_devices_added_updates_known_port(self):
details = mock.MagicMock()
details.__contains__.side_effect = lambda x: True
self.assertTrue(self._mock_treat_devices_added(details,
mock.Mock(),
'treat_vif_port'))
def test_treat_devices_removed_returns_true_for_missing_device(self):
with mock.patch.object(self.agent.plugin_rpc, 'update_device_down',
side_effect=Exception()):
self.assertTrue(self.agent.treat_devices_removed([{}]))
def _mock_treat_devices_removed(self, port_exists):
details = dict(exists=port_exists)
with mock.patch.object(self.agent.plugin_rpc, 'update_device_down',
return_value=details):
with mock.patch.object(self.agent, 'port_unbound') as port_unbound:
self.assertFalse(self.agent.treat_devices_removed([{}]))
self.assertTrue(port_unbound.called)
def test_treat_devices_removed_unbinds_port(self):
self._mock_treat_devices_removed(True)
def test_treat_devices_removed_ignores_missing_port(self):
self._mock_treat_devices_removed(False)
def test_process_network_ports(self):
reply = {'current': set(['tap0']),
'removed': set(['eth0']),
'added': set(['eth1'])}
with mock.patch.object(self.agent, 'treat_devices_added',
return_value=False) as device_added:
with mock.patch.object(self.agent, 'treat_devices_removed',
return_value=False) as device_removed:
self.assertFalse(self.agent.process_network_ports(reply))
device_added.assert_called_once_with(set(['eth1']))
device_removed.assert_called_once_with(set(['eth0']))
def test_report_state(self):
with mock.patch.object(self.agent.state_rpc,
"report_state") as report_st:
self.agent.int_br_device_count = 5
self.agent._report_state()
report_st.assert_called_with(self.agent.context,
self.agent.agent_state)
self.assertNotIn("start_flag", self.agent.agent_state)
self.assertEqual(
self.agent.agent_state["configurations"]["devices"],
self.agent.int_br_device_count
)
def test_network_delete(self):
with contextlib.nested(
mock.patch.object(self.agent, "reclaim_local_vlan"),
mock.patch.object(self.agent.tun_br, "cleanup_tunnel_port")
) as (recl_fn, clean_tun_fn):
self.agent.network_delete("unused_context",
network_id="123")
self.assertFalse(recl_fn.called)
self.agent.local_vlan_map["123"] = "LVM object"
self.agent.network_delete("unused_context",
network_id="123")
self.assertFalse(clean_tun_fn.called)
recl_fn.assert_called_with("123")
def test_port_update(self):
with contextlib.nested(
mock.patch.object(self.agent.int_br, "get_vif_port_by_id"),
mock.patch.object(self.agent, "treat_vif_port"),
mock.patch.object(self.agent.plugin_rpc, "update_device_up"),
mock.patch.object(self.agent.plugin_rpc, "update_device_down")
) as (getvif_fn, treatvif_fn, updup_fn, upddown_fn):
port = {"id": "123",
"network_id": "124",
"admin_state_up": False}
getvif_fn.return_value = "vif_port_obj"
self.agent.port_update("unused_context",
port=port,
network_type="vlan",
segmentation_id="1",
physical_network="physnet")
treatvif_fn.assert_called_with("vif_port_obj", "123",
"124", "vlan", "physnet",
"1", False)
upddown_fn.assert_called_with(self.agent.context,
"123", self.agent.agent_id,
cfg.CONF.host)
port["admin_state_up"] = True
self.agent.port_update("unused_context",
port=port,
network_type="vlan",
segmentation_id="1",
physical_network="physnet")
updup_fn.assert_called_with(self.agent.context,
"123", self.agent.agent_id,
cfg.CONF.host)
def test_port_update_plugin_rpc_failed(self):
port = {'id': 1,
'network_id': 1,
'admin_state_up': True}
with contextlib.nested(
mock.patch.object(self.mod_agent.LOG, 'error'),
mock.patch.object(self.agent.int_br, "get_vif_port_by_id"),
mock.patch.object(self.agent.plugin_rpc, 'update_device_up'),
mock.patch.object(self.agent, 'port_bound'),
mock.patch.object(self.agent.plugin_rpc, 'update_device_down'),
mock.patch.object(self.agent, 'port_dead')
) as (log, _, device_up, _, device_down, _):
device_up.side_effect = rpc_common.Timeout
self.agent.port_update(mock.Mock(), port=port)
self.assertTrue(device_up.called)
self.assertEqual(log.call_count, 1)
log.reset_mock()
port['admin_state_up'] = False
device_down.side_effect = rpc_common.Timeout
self.agent.port_update(mock.Mock(), port=port)
self.assertTrue(device_down.called)
self.assertEqual(log.call_count, 1)
def test_setup_physical_bridges(self):
with contextlib.nested(
mock.patch.object(ip_lib, "device_exists"),
mock.patch.object(utils, "execute"),
mock.patch.object(self.mod_agent.OVSBridge, "add_port"),
mock.patch.object(self.mod_agent.OVSBridge, "delete_port"),
mock.patch.object(self.mod_agent.OVSBridge, "set_protocols"),
mock.patch.object(self.mod_agent.OVSBridge, "set_controller"),
mock.patch.object(self.mod_agent.OVSBridge, "get_datapath_id",
return_value='0xa'),
mock.patch.object(self.agent.int_br, "add_port"),
mock.patch.object(self.agent.int_br, "delete_port"),
mock.patch.object(ip_lib.IPWrapper, "add_veth"),
mock.patch.object(ip_lib.IpLinkCommand, "delete"),
mock.patch.object(ip_lib.IpLinkCommand, "set_up"),
mock.patch.object(ip_lib.IpLinkCommand, "set_mtu"),
mock.patch.object(self.mod_agent.ryu_api, "get_datapath",
return_value=self.datapath)
) as (devex_fn, utilsexec_fn,
ovs_addport_fn, ovs_delport_fn, ovs_set_protocols_fn,
ovs_set_controller_fn, ovs_datapath_id_fn, br_addport_fn,
br_delport_fn, addveth_fn, linkdel_fn, linkset_fn, linkmtu_fn,
ryu_api_fn):
devex_fn.return_value = True
parent = mock.MagicMock()
parent.attach_mock(utilsexec_fn, 'utils_execute')
parent.attach_mock(linkdel_fn, 'link_delete')
parent.attach_mock(addveth_fn, 'add_veth')
addveth_fn.return_value = (ip_lib.IPDevice("int-br-eth1"),
ip_lib.IPDevice("phy-br-eth1"))
ovs_addport_fn.return_value = "25"
br_addport_fn.return_value = "11"
self.agent.setup_physical_bridges({"physnet1": "br-eth"})
expected_calls = [mock.call.link_delete(),
mock.call.utils_execute(['/sbin/udevadm',
'settle',
'--timeout=10']),
mock.call.add_veth('int-br-eth',
'phy-br-eth')]
parent.assert_has_calls(expected_calls, any_order=False)
self.assertEqual(self.agent.int_ofports["physnet1"],
"11")
self.assertEqual(self.agent.phys_ofports["physnet1"],
"25")
def test_port_unbound(self):
with mock.patch.object(self.agent, "reclaim_local_vlan") as reclvl_fn:
self.agent.enable_tunneling = True
lvm = mock.Mock()
lvm.network_type = "gre"
lvm.vif_ports = {"vif1": mock.Mock()}
self.agent.local_vlan_map["netuid12345"] = lvm
self.agent.port_unbound("vif1", "netuid12345")
self.assertTrue(reclvl_fn.called)
reclvl_fn.called = False
lvm.vif_ports = {}
self.agent.port_unbound("vif1", "netuid12345")
self.assertEqual(reclvl_fn.call_count, 2)
lvm.vif_ports = {"vif1": mock.Mock()}
self.agent.port_unbound("vif3", "netuid12345")
self.assertEqual(reclvl_fn.call_count, 2)
def _check_ovs_vxlan_version(self, installed_usr_version,
installed_klm_version,
expecting_ok):
with mock.patch(
'neutron.agent.linux.ovs_lib.get_installed_ovs_klm_version'
) as klm_cmd:
with mock.patch(
'neutron.agent.linux.ovs_lib.get_installed_ovs_usr_version'
) as usr_cmd:
try:
klm_cmd.return_value = installed_klm_version
usr_cmd.return_value = installed_usr_version
self.agent.tunnel_types = 'vxlan'
self.agent._check_ovs_version()
version_ok = True
except SystemExit as e:
self.assertEqual(e.code, 1)
version_ok = False
self.assertEqual(version_ok, expecting_ok)
def test_check_minimum_version(self):
min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION
self._check_ovs_vxlan_version(min_vxlan_ver, min_vxlan_ver,
expecting_ok=True)
def test_check_future_version(self):
install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) + 0.01)
self._check_ovs_vxlan_version(install_ver, install_ver,
expecting_ok=True)
def test_check_fail_version(self):
install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) - 0.01)
self._check_ovs_vxlan_version(install_ver, install_ver,
expecting_ok=False)
def test_check_fail_no_version(self):
self._check_ovs_vxlan_version(None, None,
expecting_ok=False)
def test_check_fail_klm_version(self):
min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION
install_ver = str(float(min_vxlan_ver) - 0.01)
self._check_ovs_vxlan_version(min_vxlan_ver, install_ver,
expecting_ok=False)
def test_daemon_loop_uses_polling_manager(self):
with mock.patch(
'neutron.agent.linux.polling.get_polling_manager'
) as mock_get_pm:
fake_pm = mock.Mock()
mock_get_pm.return_value = fake_pm
fake_pm.__enter__ = mock.Mock()
fake_pm.__exit__ = mock.Mock()
with mock.patch.object(
self.agent, 'ovsdb_monitor_loop'
) as mock_loop:
self.agent.daemon_loop()
mock_get_pm.assert_called_once_with(True, 'fake_helper',
constants.DEFAULT_OVSDBMON_RESPAWN)
mock_loop.assert_called_once_with(polling_manager=fake_pm.__enter__())
def test_setup_tunnel_port_error_negative(self):
with contextlib.nested(
mock.patch.object(self.agent.tun_br, 'add_tunnel_port',
return_value='-1'),
mock.patch.object(self.mod_agent.LOG, 'error')
) as (add_tunnel_port_fn, log_error_fn):
ofport = self.agent.setup_tunnel_port(
'gre-1', 'remote_ip', p_const.TYPE_GRE)
add_tunnel_port_fn.assert_called_once_with(
'gre-1', 'remote_ip', self.agent.local_ip, p_const.TYPE_GRE,
self.agent.vxlan_udp_port)
log_error_fn.assert_called_once_with(
_("Failed to set-up %(type)s tunnel port to %(ip)s"),
{'type': p_const.TYPE_GRE, 'ip': 'remote_ip'})
self.assertEqual(ofport, 0)
def test_setup_tunnel_port_error_not_int(self):
with contextlib.nested(
mock.patch.object(self.agent.tun_br, 'add_tunnel_port',
return_value=None),
mock.patch.object(self.mod_agent.LOG, 'exception'),
mock.patch.object(self.mod_agent.LOG, 'error')
) as (add_tunnel_port_fn, log_exc_fn, log_error_fn):
ofport = self.agent.setup_tunnel_port(
'gre-1', 'remote_ip', p_const.TYPE_GRE)
add_tunnel_port_fn.assert_called_once_with(
'gre-1', 'remote_ip', self.agent.local_ip, p_const.TYPE_GRE,
self.agent.vxlan_udp_port)
log_exc_fn.assert_called_once_with(
_("ofport should have a value that can be "
"interpreted as an integer"))
log_error_fn.assert_called_once_with(
_("Failed to set-up %(type)s tunnel port to %(ip)s"),
{'type': p_const.TYPE_GRE, 'ip': 'remote_ip'})
self.assertEqual(ofport, 0)
class AncillaryBridgesTest(OFAAgentTestCase):
def setUp(self):
super(AncillaryBridgesTest, self).setUp()
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
notifier_cls.return_value = self.notifier
# Avoid rpc initialization for unit tests
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
cfg.CONF.set_override('report_interval', 0, 'AGENT')
self.kwargs = self.mod_agent.create_agent_config_map(cfg.CONF)
def _test_ancillary_bridges(self, bridges, ancillary):
device_ids = ancillary[:]
def pullup_side_effect(self, *args):
result = device_ids.pop(0)
return result
with contextlib.nested(
mock.patch.object(self.mod_agent.OFANeutronAgent,
'setup_integration_br',
return_value=mock.Mock()),
mock.patch('neutron.agent.linux.utils.get_interface_mac',
return_value='00:00:00:00:00:01'),
mock.patch.object(self.mod_agent.OVSBridge,
'get_local_port_mac',
return_value='00:00:00:00:00:01'),
mock.patch('neutron.agent.linux.ovs_lib.get_bridges',
return_value=bridges),
mock.patch(
'neutron.agent.linux.ovs_lib.get_bridge_external_bridge_id',
side_effect=pullup_side_effect)):
self.agent = self.mod_agent.OFANeutronAgent(
self.ryuapp, **self.kwargs)
self.assertEqual(len(ancillary), len(self.agent.ancillary_brs))
if ancillary:
bridges = [br.br_name for br in self.agent.ancillary_brs]
for br in ancillary:
self.assertIn(br, bridges)
def test_ancillary_bridges_single(self):
bridges = ['br-int', 'br-ex']
self._test_ancillary_bridges(bridges, ['br-ex'])
def test_ancillary_bridges_none(self):
bridges = ['br-int']
self._test_ancillary_bridges(bridges, [])
def test_ancillary_bridges_multiple(self):
bridges = ['br-int', 'br-ex1', 'br-ex2']
self._test_ancillary_bridges(bridges, ['br-ex1', 'br-ex2'])

View File

@ -20,6 +20,7 @@ from neutron.agent.linux import ovs_lib
from neutron.agent.linux import utils
from neutron.openstack.common import jsonutils
from neutron.openstack.common import uuidutils
from neutron.plugins.openvswitch.common import constants
from neutron.tests import base
from neutron.tests import tools
@ -131,6 +132,37 @@ class OVS_Lib_Test(base.BaseTestCase):
# test __str__
str(port)
def test_set_controller(self):
controller_names = ['tcp:127.0.0.1:6633', 'tcp:172.17.16.10:5555']
self.br.set_controller(controller_names)
self.execute.assert_called_once_with(
['ovs-vsctl', self.TO, '--', 'set-controller', self.BR_NAME,
'tcp:127.0.0.1:6633', 'tcp:172.17.16.10:5555'],
root_helper=self.root_helper)
def test_del_controller(self):
self.br.del_controller()
self.execute.assert_called_once_with(
['ovs-vsctl', self.TO, '--', 'del-controller', self.BR_NAME],
root_helper=self.root_helper)
def test_get_controller(self):
self.execute.return_value = 'tcp:127.0.0.1:6633\ntcp:172.17.16.10:5555'
names = self.br.get_controller()
self.assertEqual(names,
['tcp:127.0.0.1:6633', 'tcp:172.17.16.10:5555'])
self.execute.assert_called_once_with(
['ovs-vsctl', self.TO, '--', 'get-controller', self.BR_NAME],
root_helper=self.root_helper)
def test_set_protocols(self):
protocols = 'OpenFlow13'
self.br.set_protocols(protocols)
self.execute.assert_called_once_with(
['ovs-vsctl', self.TO, '--', 'set', 'bridge', self.BR_NAME,
"protocols=%s" % protocols],
root_helper=self.root_helper)
def test_create(self):
self.br.add_bridge(self.BR_NAME)
@ -666,3 +698,47 @@ class OVS_Lib_Test(base.BaseTestCase):
data = [[["map", external_ids], "tap99", 1]]
self.assertIsNone(self._test_get_vif_port_by_id('tap99id', data,
"br-ext"))
def _check_ovs_vxlan_version(self, installed_usr_version,
installed_klm_version,
expecting_ok):
with mock.patch(
'neutron.agent.linux.ovs_lib.get_installed_ovs_klm_version'
) as klm_cmd:
with mock.patch(
'neutron.agent.linux.ovs_lib.get_installed_ovs_usr_version'
) as usr_cmd:
try:
klm_cmd.return_value = installed_klm_version
usr_cmd.return_value = installed_usr_version
ovs_lib.check_ovs_vxlan_version(root_helper='sudo')
version_ok = True
except SystemError:
version_ok = False
self.assertEqual(version_ok, expecting_ok)
def test_check_minimum_version(self):
min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION
self._check_ovs_vxlan_version(min_vxlan_ver, min_vxlan_ver,
expecting_ok=True)
def test_check_future_version(self):
install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) + 0.01)
self._check_ovs_vxlan_version(install_ver, install_ver,
expecting_ok=True)
def test_check_fail_version(self):
install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) - 0.01)
self._check_ovs_vxlan_version(install_ver, install_ver,
expecting_ok=False)
def test_check_fail_no_version(self):
self._check_ovs_vxlan_version(None, None,
expecting_ok=False)
def test_check_fail_klm_version(self):
min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION
install_ver = str(float(min_vxlan_ver) - 0.01)
self._check_ovs_vxlan_version(min_vxlan_ver,
install_ver,
expecting_ok=False)

View File

@ -468,7 +468,7 @@ class TestOvsNeutronAgent(base.BaseTestCase):
self.assertEqual(reclvl_fn.call_count, 2)
def _check_ovs_vxlan_version(self, installed_usr_version,
installed_klm_version, min_vers,
installed_klm_version,
expecting_ok):
with mock.patch(
'neutron.agent.linux.ovs_lib.get_installed_ovs_klm_version'
@ -480,8 +480,7 @@ class TestOvsNeutronAgent(base.BaseTestCase):
klm_cmd.return_value = installed_klm_version
usr_cmd.return_value = installed_usr_version
self.agent.tunnel_types = 'vxlan'
ovs_neutron_agent.check_ovs_version(min_vers,
root_helper='sudo')
self.agent._check_ovs_version()
version_ok = True
except SystemExit as e:
self.assertEqual(e.code, 1)
@ -489,28 +488,28 @@ class TestOvsNeutronAgent(base.BaseTestCase):
self.assertEqual(version_ok, expecting_ok)
def test_check_minimum_version(self):
self._check_ovs_vxlan_version('1.10', '1.10',
constants.MINIMUM_OVS_VXLAN_VERSION,
min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION
self._check_ovs_vxlan_version(min_vxlan_ver, min_vxlan_ver,
expecting_ok=True)
def test_check_future_version(self):
self._check_ovs_vxlan_version('1.11', '1.11',
constants.MINIMUM_OVS_VXLAN_VERSION,
install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) + 0.01)
self._check_ovs_vxlan_version(install_ver, install_ver,
expecting_ok=True)
def test_check_fail_version(self):
self._check_ovs_vxlan_version('1.9', '1.9',
constants.MINIMUM_OVS_VXLAN_VERSION,
install_ver = str(float(constants.MINIMUM_OVS_VXLAN_VERSION) - 0.01)
self._check_ovs_vxlan_version(install_ver, install_ver,
expecting_ok=False)
def test_check_fail_no_version(self):
self._check_ovs_vxlan_version(None, None,
constants.MINIMUM_OVS_VXLAN_VERSION,
expecting_ok=False)
def test_check_fail_klm_version(self):
self._check_ovs_vxlan_version('1.10', '1.9',
constants.MINIMUM_OVS_VXLAN_VERSION,
min_vxlan_ver = constants.MINIMUM_OVS_VXLAN_VERSION
install_ver = str(float(min_vxlan_ver) - 0.01)
self._check_ovs_vxlan_version(min_vxlan_ver, install_ver,
expecting_ok=False)
def _prepare_l2_pop_ofports(self):

View File

@ -60,6 +60,7 @@ data_files =
etc/neutron/plugins/ml2/ml2_conf_arista.ini
etc/neutron/plugins/ml2/ml2_conf_cisco.ini
etc/neutron/plugins/bigswitch/restproxy.ini
etc/neutron/plugins/ml2/ml2_conf_ofa.ini
etc/neutron/plugins/mlnx = etc/neutron/plugins/mlnx/mlnx_conf.ini
etc/neutron/plugins/nec = etc/neutron/plugins/nec/nec.ini
etc/neutron/plugins/nicira = etc/neutron/plugins/nicira/nvp.ini
@ -125,6 +126,7 @@ console_scripts =
quantum-rootwrap = oslo.rootwrap.cmd:main
quantum-usage-audit = neutron.cmd.usage_audit:main
neutron-metering-agent = neutron.services.metering.agents.metering_agent:main
neutron-ofagent-agent = ryu.cmd.ofa_neutron_agent:main
neutron.core_plugins =
bigswitch = neutron.plugins.bigswitch.plugin:NeutronRestProxyV2
brocade = neutron.plugins.brocade.NeutronPlugin:BrocadePluginV2
@ -167,6 +169,7 @@ neutron.ml2.mechanism_drivers =
cisco_nexus = neutron.plugins.ml2.drivers.cisco.nexus.mech_cisco_nexus:CiscoNexusMechanismDriver
l2population = neutron.plugins.ml2.drivers.l2pop.mech_driver:L2populationMechanismDriver
bigswitch = neutron.plugins.ml2.drivers.mech_bigswitch.driver:BigSwitchMechanismDriver
ofagent = neutron.plugins.ml2.drivers.mech_ofagent:OfagentMechanismDriver
neutron.openstack.common.cache.backends =
memory = neutron.openstack.common.cache._backends.memory:MemoryBackend