"Fossies" - the Fresh Open Source Software Archive

Member "neutron-14.0.3/neutron/tests/unit/plugins/ml2/drivers/openvswitch/mech_driver/test_mech_openvswitch.py" (22 Oct 2019, 16332 Bytes) of package /linux/misc/openstack/neutron-14.0.3.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the last Fossies "Diffs" side-by-side code changes report for "test_mech_openvswitch.py": 14.0.2_vs_15.0.0.

    1 # Copyright (c) 2013 OpenStack Foundation
    2 # All Rights Reserved.
    3 #
    4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
    5 #    not use this file except in compliance with the License. You may obtain
    6 #    a copy of the License at
    7 #
    8 #         http://www.apache.org/licenses/LICENSE-2.0
    9 #
   10 #    Unless required by applicable law or agreed to in writing, software
   11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   13 #    License for the specific language governing permissions and limitations
   14 #    under the License.
   15 
   16 import mock
   17 from neutron_lib.api.definitions import portbindings
   18 from neutron_lib.callbacks import events
   19 from neutron_lib.callbacks import registry
   20 from neutron_lib import constants
   21 from neutron_lib.plugins.ml2 import api
   22 from oslo_config import cfg
   23 
   24 from neutron.conf.plugins.ml2.drivers.openvswitch import mech_ovs_conf
   25 from neutron.plugins.ml2.drivers.openvswitch.agent.common import (
   26     constants as a_const)
   27 from neutron.plugins.ml2.drivers.openvswitch.mech_driver import (
   28     mech_openvswitch)
   29 from neutron.tests.unit.plugins.ml2 import _test_mech_agent as base
   30 
   31 
   32 class OpenvswitchMechanismBaseTestCase(base.AgentMechanismBaseTestCase):
   33     VIF_TYPE = portbindings.VIF_TYPE_OVS
   34     VIF_DETAILS = {'bridge_name': 'br-int',
   35                    portbindings.OVS_DATAPATH_TYPE: 'system',
   36                    portbindings.CAP_PORT_FILTER: True,
   37                    portbindings.OVS_HYBRID_PLUG: True}
   38     AGENT_TYPE = constants.AGENT_TYPE_OVS
   39 
   40     GOOD_MAPPINGS = {'fake_physical_network': 'fake_bridge'}
   41     GOOD_TUNNEL_TYPES = ['gre', 'vxlan']
   42     GOOD_CONFIGS = {'bridge_mappings': GOOD_MAPPINGS,
   43                     'integration_bridge': 'br-int',
   44                     'tunnel_types': GOOD_TUNNEL_TYPES}
   45 
   46     BAD_MAPPINGS = {'wrong_physical_network': 'wrong_bridge'}
   47     BAD_TUNNEL_TYPES = ['bad_tunnel_type']
   48     BAD_CONFIGS = {'bridge_mappings': BAD_MAPPINGS,
   49                    'integration_bridge': 'br-int',
   50                    'tunnel_types': BAD_TUNNEL_TYPES}
   51 
   52     AGENTS = [{'alive': True,
   53                'configurations': GOOD_CONFIGS,
   54                'host': 'host'}]
   55     AGENTS_DEAD = [{'alive': False,
   56                     'configurations': GOOD_CONFIGS,
   57                     'host': 'dead_host'}]
   58     AGENTS_BAD = [{'alive': False,
   59                    'configurations': GOOD_CONFIGS,
   60                    'host': 'bad_host_1'},
   61                   {'alive': True,
   62                    'configurations': BAD_CONFIGS,
   63                    'host': 'bad_host_2'}]
   64 
   65     def setUp(self):
   66         super(OpenvswitchMechanismBaseTestCase, self).setUp()
   67         cfg.CONF.set_override('firewall_driver', 'iptables_hybrid',
   68                               'SECURITYGROUP')
   69         self.driver = mech_openvswitch.OpenvswitchMechanismDriver()
   70         self.driver.initialize()
   71 
   72     def test__set_bridge_name_notify(self):
   73 
   74         def fake_callback(resource, event, trigger, payload=None):
   75             trigger('fake-br-name')
   76 
   77         def noop_callback(resource, event, trigger, payload=None):
   78             pass
   79 
   80         # hardcode callback to override bridge name
   81         registry.subscribe(fake_callback, a_const.OVS_BRIDGE_NAME,
   82                            events.BEFORE_READ)
   83         fake_vif_details = {}
   84         fake_agent = {'configurations': {'integration_bridge': 'fake-br'}}
   85         old_fake_agent = {'configurations': {}}
   86         self.driver._set_bridge_name('foo', fake_vif_details, fake_agent)
   87         # assert that callback value is used
   88         self.assertEqual(
   89             'fake-br-name',
   90             fake_vif_details.get(portbindings.VIF_DETAILS_BRIDGE_NAME, ''))
   91         # replace callback with noop
   92         registry.unsubscribe(fake_callback, a_const.OVS_BRIDGE_NAME,
   93                            events.BEFORE_READ)
   94         registry.subscribe(noop_callback, a_const.OVS_BRIDGE_NAME,
   95                            events.BEFORE_READ)
   96         fake_vif_details = {}
   97         self.driver._set_bridge_name('foo', fake_vif_details, fake_agent)
   98         # assert that agent config value is used
   99         self.assertEqual(
  100             'fake-br',
  101             fake_vif_details.get(portbindings.VIF_DETAILS_BRIDGE_NAME, ''))
  102         fake_vif_details = {}
  103         self.driver._set_bridge_name('foo', fake_vif_details, old_fake_agent)
  104         # assert that if agent does not supply integration_bridge bridge_name
  105         # is not set in vif:binding-details
  106         self.assertIsNone(
  107             fake_vif_details.get(portbindings.VIF_DETAILS_BRIDGE_NAME))
  108 
  109 
  110 class OpenvswitchMechanismSGDisabledBaseTestCase(
  111     OpenvswitchMechanismBaseTestCase):
  112     VIF_DETAILS = {'bridge_name': 'br-int',
  113                    portbindings.OVS_DATAPATH_TYPE: 'system',
  114                    portbindings.CAP_PORT_FILTER: False,
  115                    portbindings.OVS_HYBRID_PLUG: False}
  116 
  117     def setUp(self):
  118         cfg.CONF.set_override('enable_security_group',
  119                               False,
  120                               group='SECURITYGROUP')
  121         super(OpenvswitchMechanismSGDisabledBaseTestCase, self).setUp()
  122 
  123 
  124 class OpenvswitchMechanismHybridPlugTestCase(OpenvswitchMechanismBaseTestCase):
  125 
  126     def _make_port_ctx(self, agents):
  127         segments = [{api.ID: 'local_segment_id', api.NETWORK_TYPE: 'local'}]
  128         return base.FakePortContext(self.AGENT_TYPE, agents, segments,
  129                                     vnic_type=self.VNIC_TYPE)
  130 
  131     def test_backward_compat_with_unreporting_agent(self):
  132         hybrid = portbindings.OVS_HYBRID_PLUG
  133         # agent didn't report so it should be hybrid based on server config
  134         context = self._make_port_ctx(self.AGENTS)
  135         self.driver.bind_port(context)
  136         self.assertTrue(context._bound_vif_details[hybrid])
  137         self.driver.vif_details[hybrid] = False
  138         context = self._make_port_ctx(self.AGENTS)
  139         self.driver.bind_port(context)
  140         self.assertFalse(context._bound_vif_details[hybrid])
  141 
  142     def test_hybrid_plug_true_if_agent_requests(self):
  143         hybrid = portbindings.OVS_HYBRID_PLUG
  144         # set server side default to false and ensure that hybrid becomes
  145         # true if requested by the agent
  146         self.driver.vif_details[hybrid] = False
  147         agents = [{'alive': True,
  148                    'configurations': {hybrid: True},
  149                    'host': 'host'}]
  150         context = self._make_port_ctx(agents)
  151         self.driver.bind_port(context)
  152         self.assertTrue(context._bound_vif_details[hybrid])
  153 
  154     def test_hybrid_plug_false_if_agent_requests(self):
  155         hybrid = portbindings.OVS_HYBRID_PLUG
  156         # set server side default to true and ensure that hybrid becomes
  157         # false if requested by the agent
  158         self.driver.vif_details[hybrid] = True
  159         agents = [{'alive': True,
  160                    'configurations': {hybrid: False},
  161                    'host': 'host'}]
  162         context = self._make_port_ctx(agents)
  163         self.driver.bind_port(context)
  164         self.assertFalse(context._bound_vif_details[hybrid])
  165 
  166 
  167 class OpenvswitchMechanismGenericTestCase(OpenvswitchMechanismBaseTestCase,
  168                                           base.AgentMechanismGenericTestCase):
  169     def test_driver_responsible_for_ports_allocation(self):
  170         agents = [
  171             {'agent_type': 'Open vSwitch agent',
  172              'configurations': {'resource_provider_bandwidths': {'eth0': {}}},
  173              'id': '1',
  174              'host': 'host'}
  175         ]
  176         segments = []
  177         # uuid -v5 87ee7d5c-73bb-11e8-9008-c4d987b2a692 host:eth0
  178         profile = {'allocation': '13cc0ed9-e802-5eaa-b4c7-3441855e31f2'}
  179 
  180         port_ctx = base.FakePortContext(
  181             self.AGENT_TYPE,
  182             agents,
  183             segments,
  184             vnic_type=portbindings.VNIC_NORMAL,
  185             profile=profile)
  186         with mock.patch.object(self.driver, '_possible_agents_for_port',
  187                                return_value=agents):
  188             self.assertTrue(
  189                 self.driver.responsible_for_ports_allocation(port_ctx))
  190 
  191 
  192 class OpenvswitchMechanismLocalTestCase(OpenvswitchMechanismBaseTestCase,
  193                                         base.AgentMechanismLocalTestCase):
  194     pass
  195 
  196 
  197 class OpenvswitchMechanismFlatTestCase(OpenvswitchMechanismBaseTestCase,
  198                                        base.AgentMechanismFlatTestCase):
  199     pass
  200 
  201 
  202 class OpenvswitchMechanismVlanTestCase(OpenvswitchMechanismBaseTestCase,
  203                                        base.AgentMechanismVlanTestCase):
  204     pass
  205 
  206 
  207 class OpenvswitchMechanismGreTestCase(OpenvswitchMechanismBaseTestCase,
  208                                       base.AgentMechanismGreTestCase):
  209     pass
  210 
  211 
  212 class OpenvswitchMechanismSGDisabledLocalTestCase(
  213     OpenvswitchMechanismSGDisabledBaseTestCase,
  214     base.AgentMechanismLocalTestCase):
  215     pass
  216 
  217 
  218 class OpenvswitchMechanismFirewallUndefinedTestCase(
  219     OpenvswitchMechanismBaseTestCase, base.AgentMechanismLocalTestCase):
  220 
  221     VIF_DETAILS = {'bridge_name': 'br-int',
  222                    portbindings.OVS_DATAPATH_TYPE: 'system',
  223                    portbindings.CAP_PORT_FILTER: True,
  224                    portbindings.OVS_HYBRID_PLUG: True}
  225 
  226     def setUp(self):
  227         # this simple test case just ensures backward compatibility where
  228         # the server has no firewall driver configured, which should result
  229         # in hybrid plugging.
  230         super(OpenvswitchMechanismFirewallUndefinedTestCase, self).setUp()
  231         cfg.CONF.set_override('firewall_driver', '', 'SECURITYGROUP')
  232         self.driver = mech_openvswitch.OpenvswitchMechanismDriver()
  233         self.driver.initialize()
  234 
  235 
  236 class OpenvswitchMechanismDPDKTestCase(OpenvswitchMechanismBaseTestCase):
  237 
  238     GOOD_MAPPINGS = {'fake_physical_network': 'fake_bridge'}
  239 
  240     GOOD_TUNNEL_TYPES = ['gre', 'vxlan']
  241 
  242     VHOST_CONFIGS = {'bridge_mappings': GOOD_MAPPINGS,
  243                     'integration_bridge': 'br-int',
  244                     'tunnel_types': GOOD_TUNNEL_TYPES,
  245                     'datapath_type': a_const.OVS_DATAPATH_NETDEV,
  246                     'ovs_capabilities': {
  247                         'iface_types': [a_const.OVS_DPDK_VHOST_USER]}}
  248 
  249     VHOST_SERVER_CONFIGS = {'bridge_mappings': GOOD_MAPPINGS,
  250                     'integration_bridge': 'br-int',
  251                     'tunnel_types': GOOD_TUNNEL_TYPES,
  252                     'datapath_type': a_const.OVS_DATAPATH_NETDEV,
  253                     'ovs_capabilities': {
  254                         'iface_types': [a_const.OVS_DPDK_VHOST_USER_CLIENT]}}
  255 
  256     SYSTEM_CONFIGS = {'bridge_mappings': GOOD_MAPPINGS,
  257                       'integration_bridge': 'br-int',
  258                       'tunnel_types': GOOD_TUNNEL_TYPES,
  259                       'datapath_type': a_const.OVS_DATAPATH_SYSTEM,
  260                       'ovs_capabilities': {'iface_types': []}}
  261 
  262     AGENT = {'alive': True,
  263              'configurations': VHOST_CONFIGS,
  264              'host': 'host'}
  265 
  266     AGENT_SERVER = {'alive': True,
  267                     'configurations': VHOST_SERVER_CONFIGS,
  268                     'host': 'host'}
  269 
  270     AGENT_SYSTEM = {'alive': True,
  271                     'configurations': SYSTEM_CONFIGS,
  272                     'host': 'host'}
  273 
  274     def test_get_vhost_mode(self):
  275         ifaces = []
  276         result = self.driver.get_vhost_mode(ifaces)
  277         self.assertEqual(portbindings.VHOST_USER_MODE_CLIENT, result)
  278 
  279         ifaces = [a_const.OVS_DPDK_VHOST_USER]
  280         result = self.driver.get_vhost_mode(ifaces)
  281         self.assertEqual(portbindings.VHOST_USER_MODE_CLIENT, result)
  282 
  283         ifaces = [a_const.OVS_DPDK_VHOST_USER_CLIENT]
  284         result = self.driver.get_vhost_mode(ifaces)
  285         self.assertEqual(portbindings.VHOST_USER_MODE_SERVER, result)
  286 
  287     def test_get_vif_type(self):
  288         result = self.driver.get_vif_type(None, self.AGENT, None)
  289         self.assertEqual(portbindings.VIF_TYPE_VHOST_USER, result)
  290 
  291         result = self.driver.get_vif_type(None, self.AGENT_SERVER, None)
  292         self.assertEqual(portbindings.VIF_TYPE_VHOST_USER, result)
  293 
  294         result = self.driver.get_vif_type(None, self.AGENT_SYSTEM, None)
  295         self.assertEqual(portbindings.VIF_TYPE_OVS, result)
  296 
  297 
  298 class OpenvswitchMechanismSRIOVTestCase(OpenvswitchMechanismBaseTestCase):
  299 
  300     def _make_port_ctx(self, agents, profile=None):
  301         segments = [{api.ID: 'local_segment_id', api.NETWORK_TYPE: 'local'}]
  302         return base.FakePortContext(self.AGENT_TYPE, agents, segments,
  303                                     vnic_type=portbindings.VNIC_DIRECT,
  304                                     profile=profile)
  305 
  306     @mock.patch('neutron.plugins.ml2.drivers.mech_agent.'
  307                 'SimpleAgentMechanismDriverBase.bind_port')
  308     def test_bind_port_sriov_legacy(self, mocked_bind_port):
  309         context = self._make_port_ctx(self.AGENTS)
  310         self.driver.bind_port(context)
  311         mocked_bind_port.assert_not_called()
  312 
  313     @mock.patch('neutron.plugins.ml2.drivers.mech_agent.'
  314                 'SimpleAgentMechanismDriverBase.bind_port')
  315     def test_bind_port_sriov_switchdev(self, mocked_bind_port):
  316         profile = {'capabilities': ['switchdev']}
  317         context = self._make_port_ctx(self.AGENTS, profile=profile)
  318         self.driver.bind_port(context)
  319         mocked_bind_port.assert_called()
  320 
  321 
  322 class OpenvswitchMechVnicTypesTestCase(OpenvswitchMechanismBaseTestCase):
  323     def setUp(self):
  324         self.blacklist_cfg = {
  325             'OVS_DRIVER': {
  326                 'vnic_type_blacklist': []
  327             }
  328         }
  329         self.default_supported_vnics = [portbindings.VNIC_NORMAL,
  330                                         portbindings.VNIC_DIRECT]
  331         super(OpenvswitchMechVnicTypesTestCase, self).setUp()
  332 
  333     def test_default_vnic_types(self):
  334         self.assertEqual([portbindings.VNIC_NORMAL,
  335                           portbindings.VNIC_DIRECT],
  336                          self.driver.supported_vnic_types)
  337 
  338     def test_vnic_type_blacklist_valid_item(self):
  339         self.blacklist_cfg['OVS_DRIVER']['vnic_type_blacklist'] = \
  340             [portbindings.VNIC_DIRECT]
  341 
  342         fake_conf = cfg.CONF
  343         fake_conf_fixture = base.MechDriverConfFixture(
  344             fake_conf, self.blacklist_cfg,
  345             mech_ovs_conf.register_ovs_mech_driver_opts)
  346         self.useFixture(fake_conf_fixture)
  347 
  348         test_driver = mech_openvswitch.OpenvswitchMechanismDriver()
  349 
  350         supported_vnic_types = test_driver.supported_vnic_types
  351         self.assertNotIn(portbindings.VNIC_DIRECT, supported_vnic_types)
  352         self.assertEqual(len(self.default_supported_vnics) - 1,
  353                          len(supported_vnic_types))
  354 
  355     def test_vnic_type_blacklist_not_valid_item(self):
  356         self.blacklist_cfg['OVS_DRIVER']['vnic_type_blacklist'] = ['foo']
  357 
  358         fake_conf = cfg.CONF
  359         fake_conf_fixture = base.MechDriverConfFixture(
  360             fake_conf, self.blacklist_cfg,
  361             mech_ovs_conf.register_ovs_mech_driver_opts)
  362         self.useFixture(fake_conf_fixture)
  363 
  364         self.assertRaises(ValueError,
  365                           mech_openvswitch.OpenvswitchMechanismDriver)
  366 
  367     def test_vnic_type_blacklist_all_items(self):
  368         self.blacklist_cfg['OVS_DRIVER']['vnic_type_blacklist'] = \
  369             [portbindings.VNIC_NORMAL, portbindings.VNIC_DIRECT]
  370         fake_conf = cfg.CONF
  371         fake_conf_fixture = base.MechDriverConfFixture(
  372             fake_conf, self.blacklist_cfg,
  373             mech_ovs_conf.register_ovs_mech_driver_opts)
  374         self.useFixture(fake_conf_fixture)
  375 
  376         self.assertRaises(ValueError,
  377                           mech_openvswitch.OpenvswitchMechanismDriver)
  378 
  379 
  380 class OpenvswitchMechDeviceMappingsTestCase(OpenvswitchMechanismBaseTestCase):
  381 
  382     def test_standard_device_mappings(self):
  383         mappings = self.driver.get_standard_device_mappings(self.AGENTS[0])
  384         self.assertEqual(
  385             len(self.GOOD_CONFIGS['bridge_mappings']),
  386             len(mappings))
  387         for ph_orig, br_orig in self.GOOD_CONFIGS['bridge_mappings'].items():
  388             self.assertIn(ph_orig, mappings)
  389             self.assertEqual([br_orig], mappings[ph_orig])
  390 
  391     def test_standard_device_mappings_negative(self):
  392         fake_agent = {'agent_type': constants.AGENT_TYPE_OVS,
  393                       'configurations': {}}
  394         self.assertRaises(ValueError, self.driver.get_standard_device_mappings,
  395                           fake_agent)