neutron-14.0.3.tar.gz: .../test_mech_openvswitch.py

"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)