Merge "Improve unit test coverage for Cisco plugin model code"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
07a467da2c
@ -49,7 +49,6 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
|
||||
"""
|
||||
__native_bulk_support = True
|
||||
supported_extension_aliases = ["provider", "binding"]
|
||||
_plugins = {}
|
||||
_methods_to_delegate = ['create_network_bulk',
|
||||
'get_network', 'get_networks',
|
||||
'create_port_bulk',
|
||||
@ -67,6 +66,7 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
|
||||
"""
|
||||
conf.CiscoConfigOptions()
|
||||
|
||||
self._plugins = {}
|
||||
for key in conf.CISCO_PLUGINS.keys():
|
||||
plugin_obj = conf.CISCO_PLUGINS[key]
|
||||
if plugin_obj is not None:
|
||||
@ -138,28 +138,12 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
|
||||
LOG.info(_("No %s Plugin loaded"), plugin_key)
|
||||
LOG.info(_("%(plugin_key)s: %(function_name)s with args %(args)s "
|
||||
"ignored"),
|
||||
{'plugin_key': plugin_key, 'function_name': function_name,
|
||||
{'plugin_key': plugin_key,
|
||||
'function_name': function_name,
|
||||
'args': args})
|
||||
return
|
||||
return [self._invoke_plugin(plugin_key, function_name, args, kwargs)]
|
||||
|
||||
def _invoke_plugin(self, plugin_key, function_name, args, kwargs):
|
||||
"""Invoke plugin.
|
||||
|
||||
Invokes the relevant function on a device plugin's
|
||||
implementation for completing this operation.
|
||||
"""
|
||||
func = getattr(self._plugins[plugin_key], function_name)
|
||||
func_args_len = int(inspect.getargspec(func).args.__len__()) - 1
|
||||
if args.__len__() > func_args_len:
|
||||
func_args = args[:func_args_len]
|
||||
extra_args = args[func_args_len:]
|
||||
for dict_arg in extra_args:
|
||||
for k, v in dict_arg.iteritems():
|
||||
kwargs[k] = v
|
||||
return func(*func_args, **kwargs)
|
||||
else:
|
||||
return func(*args, **kwargs)
|
||||
func = getattr(self._plugins[plugin_key], function_name)
|
||||
return [func(*args, **kwargs)]
|
||||
|
||||
def _get_segmentation_id(self, network_id):
|
||||
binding_seg_id = odb.get_network_binding(None, network_id)
|
||||
@ -243,12 +227,19 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
|
||||
return ovs_output[0]
|
||||
|
||||
def get_network(self, context, id, fields=None):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Get network. This method is delegated to the vswitch plugin.
|
||||
|
||||
def get_networks(self, context, filters=None, fields=None):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def get_networks(self, context, filters=None, fields=None,
|
||||
sorts=None, limit=None, marker=None, page_reverse=False):
|
||||
"""Get networks. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def _invoke_nexus_for_net_create(self, context, tenant_id, net_id,
|
||||
instance_id, host_id):
|
||||
@ -328,12 +319,18 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
|
||||
return ovs_output[0]
|
||||
|
||||
def get_port(self, context, id, fields=None):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Get port. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def get_ports(self, context, filters=None, fields=None):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Get ports. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def _check_nexus_net_create_needed(self, new_port, old_port):
|
||||
"""Check if nexus plugin should be invoked for net_create.
|
||||
@ -521,21 +518,37 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
|
||||
n_args)
|
||||
|
||||
def create_subnet(self, context, subnet):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Create subnet. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def update_subnet(self, context, id, subnet):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Update subnet. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def get_subnet(self, context, id, fields=None):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Get subnet. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def delete_subnet(self, context, id, kwargs):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
"""Delete subnet. This method is delegated to the vswitch plugin.
|
||||
|
||||
def get_subnets(self, context, filters=None, fields=None):
|
||||
"""For this model this method will be delegated to vswitch plugin."""
|
||||
pass
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
def get_subnets(self, context, filters=None, fields=None,
|
||||
sorts=None, limit=None, marker=None, page_reverse=False):
|
||||
"""Get subnets. This method is delegated to the vswitch plugin.
|
||||
|
||||
This method is included here to satisfy abstract method requirements.
|
||||
"""
|
||||
pass # pragma no cover
|
||||
|
||||
@ -45,22 +45,25 @@ class PluginV2(db_base_plugin_v2.NeutronDbPluginV2):
|
||||
'get_subnet', 'get_subnets', ]
|
||||
|
||||
CISCO_FAULT_MAP = {
|
||||
cexc.NetworkSegmentIDNotFound: wexc.HTTPNotFound,
|
||||
cexc.NoMoreNics: wexc.HTTPBadRequest,
|
||||
cexc.NetworkVlanBindingAlreadyExists: wexc.HTTPBadRequest,
|
||||
cexc.VlanIDNotFound: wexc.HTTPNotFound,
|
||||
cexc.VlanIDNotAvailable: wexc.HTTPNotFound,
|
||||
cexc.QosNotFound: wexc.HTTPNotFound,
|
||||
cexc.QosNameAlreadyExists: wexc.HTTPBadRequest,
|
||||
cexc.CredentialNotFound: wexc.HTTPNotFound,
|
||||
cexc.CredentialNameNotFound: wexc.HTTPNotFound,
|
||||
cexc.CredentialAlreadyExists: wexc.HTTPBadRequest,
|
||||
cexc.CredentialNameNotFound: wexc.HTTPNotFound,
|
||||
cexc.CredentialNotFound: wexc.HTTPNotFound,
|
||||
cexc.NetworkSegmentIDNotFound: wexc.HTTPNotFound,
|
||||
cexc.NetworkVlanBindingAlreadyExists: wexc.HTTPBadRequest,
|
||||
cexc.NexusComputeHostNotConfigured: wexc.HTTPNotFound,
|
||||
cexc.NexusConnectFailed: wexc.HTTPServiceUnavailable,
|
||||
cexc.NexusConfigFailed: wexc.HTTPBadRequest,
|
||||
cexc.NexusConnectFailed: wexc.HTTPServiceUnavailable,
|
||||
cexc.NexusPortBindingNotFound: wexc.HTTPNotFound,
|
||||
cexc.NoMoreNics: wexc.HTTPBadRequest,
|
||||
cexc.PortIdForNexusSvi: wexc.HTTPBadRequest,
|
||||
cexc.PortVnicBindingAlreadyExists: wexc.HTTPBadRequest,
|
||||
cexc.PortVnicNotFound: wexc.HTTPNotFound}
|
||||
cexc.PortVnicNotFound: wexc.HTTPNotFound,
|
||||
cexc.QosNameAlreadyExists: wexc.HTTPBadRequest,
|
||||
cexc.QosNotFound: wexc.HTTPNotFound,
|
||||
cexc.SubnetNotSpecified: wexc.HTTPBadRequest,
|
||||
cexc.VlanIDNotAvailable: wexc.HTTPNotFound,
|
||||
cexc.VlanIDNotFound: wexc.HTTPNotFound,
|
||||
}
|
||||
|
||||
def __init__(self):
|
||||
"""Load the model class."""
|
||||
|
||||
@ -212,6 +212,23 @@ class CiscoNetworkPluginV2TestCase(test_db_plugin.NeutronDbPluginV2TestCase):
|
||||
return (vlan_deleted == vlan_deletion_expected and
|
||||
vlan_untrunked == vlan_untrunk_expected)
|
||||
|
||||
def _assertExpectedHTTP(self, status, exc):
|
||||
"""Confirm that an HTTP status corresponds to an expected exception.
|
||||
|
||||
Confirm that an HTTP status which has been returned for an
|
||||
neutron API request matches the HTTP status corresponding
|
||||
to an expected exception.
|
||||
|
||||
:param status: HTTP status
|
||||
:param exc: Expected exception
|
||||
|
||||
"""
|
||||
if exc in base.FAULT_MAP:
|
||||
expected_http = base.FAULT_MAP[exc].code
|
||||
else:
|
||||
expected_http = wexc.HTTPInternalServerError.code
|
||||
self.assertEqual(status, expected_http)
|
||||
|
||||
|
||||
class TestCiscoBasicGet(CiscoNetworkPluginV2TestCase,
|
||||
test_db_plugin.TestBasicGet):
|
||||
@ -259,23 +276,6 @@ class TestCiscoPortsV2(CiscoNetworkPluginV2TestCase,
|
||||
if do_delete:
|
||||
self._delete('ports', port['port']['id'])
|
||||
|
||||
def _assertExpectedHTTP(self, status, exc):
|
||||
"""Confirm that an HTTP status corresponds to an expected exception.
|
||||
|
||||
Confirm that an HTTP status which has been returned for an
|
||||
neutron API request matches the HTTP status corresponding
|
||||
to an expected exception.
|
||||
|
||||
:param status: HTTP status
|
||||
:param exc: Expected exception
|
||||
|
||||
"""
|
||||
if exc in base.FAULT_MAP:
|
||||
expected_http = base.FAULT_MAP[exc].code
|
||||
else:
|
||||
expected_http = wexc.HTTPInternalServerError.code
|
||||
self.assertEqual(status, expected_http)
|
||||
|
||||
def test_create_ports_bulk_emulated_plugin_failure(self):
|
||||
real_has_attr = hasattr
|
||||
|
||||
@ -911,23 +911,41 @@ class TestCiscoNetworksV2(CiscoNetworkPluginV2TestCase,
|
||||
'networks',
|
||||
wexc.HTTPInternalServerError.code)
|
||||
|
||||
def test_create_provider_vlan_network(self):
|
||||
@contextlib.contextmanager
|
||||
def _provider_vlan_network(self, phys_net, segment_id, net_name):
|
||||
provider_attrs = {provider.NETWORK_TYPE: 'vlan',
|
||||
provider.PHYSICAL_NETWORK: PHYS_NET,
|
||||
provider.SEGMENTATION_ID: '1234'}
|
||||
provider.PHYSICAL_NETWORK: phys_net,
|
||||
provider.SEGMENTATION_ID: segment_id}
|
||||
arg_list = tuple(provider_attrs.keys())
|
||||
res = self._create_network(self.fmt, 'pvnet1', True,
|
||||
res = self._create_network(self.fmt, net_name, True,
|
||||
arg_list=arg_list, **provider_attrs)
|
||||
net = self.deserialize(self.fmt, res)
|
||||
expected = [('name', 'pvnet1'),
|
||||
('admin_state_up', True),
|
||||
('status', 'ACTIVE'),
|
||||
('shared', False),
|
||||
(provider.NETWORK_TYPE, 'vlan'),
|
||||
(provider.PHYSICAL_NETWORK, PHYS_NET),
|
||||
(provider.SEGMENTATION_ID, 1234)]
|
||||
for k, v in expected:
|
||||
self.assertEqual(net['network'][k], v)
|
||||
network = self.deserialize(self.fmt, res)['network']
|
||||
try:
|
||||
yield network
|
||||
finally:
|
||||
req = self.new_delete_request('networks', network['id'])
|
||||
req.get_response(self.api)
|
||||
|
||||
def test_create_provider_vlan_network(self):
|
||||
with self._provider_vlan_network(PHYS_NET, '1234',
|
||||
'pvnet1') as network:
|
||||
expected = [('name', 'pvnet1'),
|
||||
('admin_state_up', True),
|
||||
('status', 'ACTIVE'),
|
||||
('shared', False),
|
||||
(provider.NETWORK_TYPE, 'vlan'),
|
||||
(provider.PHYSICAL_NETWORK, PHYS_NET),
|
||||
(provider.SEGMENTATION_ID, 1234)]
|
||||
for k, v in expected:
|
||||
self.assertEqual(network[k], v)
|
||||
self.assertTrue(network_db_v2.is_provider_network(network['id']))
|
||||
|
||||
def test_delete_provider_vlan_network(self):
|
||||
with self._provider_vlan_network(PHYS_NET, '1234',
|
||||
'pvnet1') as network:
|
||||
network_id = network['id']
|
||||
# Provider network should now be deleted
|
||||
self.assertFalse(network_db_v2.is_provider_network(network_id))
|
||||
|
||||
|
||||
class TestCiscoSubnetsV2(CiscoNetworkPluginV2TestCase,
|
||||
@ -995,65 +1013,106 @@ class TestCiscoRouterInterfacesV2(CiscoNetworkPluginV2TestCase):
|
||||
super(TestCiscoRouterInterfacesV2, self).setUp()
|
||||
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
|
||||
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _router(self, subnet):
|
||||
"""Create a virtual router, yield it for testing, then delete it."""
|
||||
data = {'router': {'tenant_id': 'test_tenant_id'}}
|
||||
router_req = self.new_create_request('routers', data, self.fmt)
|
||||
res = router_req.get_response(self.ext_api)
|
||||
router = self.deserialize(self.fmt, res)
|
||||
try:
|
||||
yield router
|
||||
finally:
|
||||
self._delete('routers', router['router']['id'])
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _router_interface(self, router, subnet):
|
||||
"""Create a router interface, yield for testing, then delete it."""
|
||||
interface_data = {'subnet_id': subnet['subnet']['id']}
|
||||
req = self.new_action_request('routers', interface_data,
|
||||
router['router']['id'],
|
||||
'add_router_interface')
|
||||
req.get_response(self.ext_api)
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
req = self.new_action_request('routers', interface_data,
|
||||
router['router']['id'],
|
||||
'remove_router_interface')
|
||||
req.get_response(self.ext_api)
|
||||
|
||||
def test_nexus_l3_enable_config(self):
|
||||
"""Verify proper operation of the Nexus L3 enable configuration."""
|
||||
self.addCleanup(cisco_config.CONF.reset)
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _network_subnet_router(self):
|
||||
"""Context mgr for creating/deleting a net, subnet, and router."""
|
||||
with self.network() as network:
|
||||
with self.subnet(network=network) as subnet:
|
||||
with self._router(subnet) as router:
|
||||
# With 'nexus_l3_enable' configured to True, confirm that
|
||||
# a switched virtual interface (SVI) is created/deleted
|
||||
# on the Nexus switch when a virtual router interface is
|
||||
# created/deleted.
|
||||
cisco_config.CONF.set_override('nexus_l3_enable',
|
||||
True, 'CISCO')
|
||||
with self._router_interface(router, subnet):
|
||||
self.assertTrue(self._is_in_last_nexus_cfg(
|
||||
['interface', 'vlan', 'ip', 'address']))
|
||||
self.assertTrue(self._is_in_nexus_cfg(
|
||||
['no', 'interface', 'vlan']))
|
||||
self.assertTrue(self._is_in_last_nexus_cfg(
|
||||
['no', 'vlan']))
|
||||
data = {'router': {'tenant_id': 'test_tenant_id'}}
|
||||
request = self.new_create_request('routers', data, self.fmt)
|
||||
response = request.get_response(self.ext_api)
|
||||
router = self.deserialize(self.fmt, response)
|
||||
try:
|
||||
yield network, subnet, router
|
||||
finally:
|
||||
self._delete('routers', router['router']['id'])
|
||||
|
||||
# With 'nexus_l3_enable' configured to False, confirm
|
||||
# that no changes are made to the Nexus switch running
|
||||
# configuration when a virtual router interface is
|
||||
# created and then deleted.
|
||||
cisco_config.CONF.set_override('nexus_l3_enable',
|
||||
False, 'CISCO')
|
||||
self.mock_ncclient.reset_mock()
|
||||
self._router_interface(router, subnet)
|
||||
self.assertFalse(self.mock_ncclient.manager.connect.
|
||||
return_value.edit_config.called)
|
||||
@contextlib.contextmanager
|
||||
def _router_interface(self, router, subnet, **kwargs):
|
||||
"""Create a router interface, yield the response, then delete it."""
|
||||
interface_data = {}
|
||||
if subnet:
|
||||
interface_data['subnet_id'] = subnet['subnet']['id']
|
||||
interface_data.update(kwargs)
|
||||
request = self.new_action_request('routers', interface_data,
|
||||
router['router']['id'],
|
||||
'add_router_interface')
|
||||
response = request.get_response(self.ext_api)
|
||||
try:
|
||||
yield response
|
||||
finally:
|
||||
# If router interface was created successfully, delete it now.
|
||||
if response.status_int == wexc.HTTPOk.code:
|
||||
request = self.new_action_request('routers', interface_data,
|
||||
router['router']['id'],
|
||||
'remove_router_interface')
|
||||
request.get_response(self.ext_api)
|
||||
|
||||
@contextlib.contextmanager
|
||||
def _network_subnet_router_interface(self, **kwargs):
|
||||
"""Context mgr for create/deleting a net, subnet, router and intf."""
|
||||
with self._network_subnet_router() as (network, subnet, router):
|
||||
with self._router_interface(router, subnet,
|
||||
**kwargs) as response:
|
||||
yield response
|
||||
|
||||
def test_add_remove_router_intf_with_nexus_l3_enabled(self):
|
||||
"""Verifies proper add/remove intf operation with Nexus L3 enabled.
|
||||
|
||||
With 'nexus_l3_enable' configured to True, confirm that a switched
|
||||
virtual interface (SVI) is created/deleted on the Nexus switch when
|
||||
a virtual router interface is created/deleted.
|
||||
"""
|
||||
cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
|
||||
with self._network_subnet_router_interface():
|
||||
self.assertTrue(self._is_in_last_nexus_cfg(
|
||||
['interface', 'vlan', 'ip', 'address']))
|
||||
# Clear list of calls made to mock ncclient
|
||||
self.mock_ncclient.reset()
|
||||
# Router interface is now deleted. Confirm that SVI
|
||||
# has been deleted from the Nexus switch.
|
||||
self.assertTrue(self._is_in_nexus_cfg(['no', 'interface', 'vlan']))
|
||||
self.assertTrue(self._is_in_last_nexus_cfg(['no', 'vlan']))
|
||||
|
||||
def test_add_remove_router_intf_with_nexus_l3_disabled(self):
|
||||
"""Verifies proper add/remove intf operation with Nexus L3 disabled.
|
||||
|
||||
With 'nexus_l3_enable' configured to False, confirm that no changes
|
||||
are made to the Nexus switch running configuration when a virtual
|
||||
router interface is created and then deleted.
|
||||
"""
|
||||
cisco_config.CONF.set_override('nexus_l3_enable', False, 'CISCO')
|
||||
with self._network_subnet_router_interface():
|
||||
self.assertFalse(self.mock_ncclient.manager.connect.
|
||||
return_value.edit_config.called)
|
||||
|
||||
def test_create_svi_but_subnet_not_specified_exception(self):
|
||||
"""Tests raising of SubnetNotSpecified exception.
|
||||
|
||||
Tests that a SubnetNotSpecified exception is raised when an
|
||||
add_router_interface request is made for creating a switch virtual
|
||||
interface (SVI), but the request does not specify a subnet.
|
||||
"""
|
||||
cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
|
||||
with self._network_subnet_router() as (network, subnet, router):
|
||||
with self._router_interface(router, subnet=None) as response:
|
||||
self._assertExpectedHTTP(response.status_int,
|
||||
c_exc.SubnetNotSpecified)
|
||||
|
||||
def test_create_svi_but_port_id_included_exception(self):
|
||||
"""Tests raising of PortIdForNexusSvi exception.
|
||||
|
||||
Tests that a PortIdForNexusSvi exception is raised when an
|
||||
add_router_interface request is made for creating a switch virtual
|
||||
interface (SVI), but the request includes a virtual port ID.
|
||||
"""
|
||||
cisco_config.CONF.set_override('nexus_l3_enable', True, 'CISCO')
|
||||
with self._network_subnet_router_interface(
|
||||
port_id='my_port_id') as response:
|
||||
self._assertExpectedHTTP(response.status_int,
|
||||
c_exc.PortIdForNexusSvi)
|
||||
|
||||
|
||||
class TestCiscoPortsV2XML(TestCiscoPortsV2):
|
||||
|
||||
68
neutron/tests/unit/cisco/test_plugin_model.py
Executable file
68
neutron/tests/unit/cisco/test_plugin_model.py
Executable file
@ -0,0 +1,68 @@
|
||||
# Copyright 2014 Cisco Systems, Inc.
|
||||
#
|
||||
# Licensed under the Apache License, Version 2.0 (the "License");
|
||||
# you may not use this file except in compliance with the License.
|
||||
# You may obtain a copy of the License at
|
||||
#
|
||||
# http://www.apache.org/licenses/LICENSE-2.0
|
||||
#
|
||||
# Unless required by applicable law or agreed to in writing, software
|
||||
# distributed under the License is distributed on an "AS IS" BASIS,
|
||||
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
|
||||
# implied.
|
||||
# See the License for the specific language governing permissions and
|
||||
# limitations under the License.
|
||||
|
||||
import sys
|
||||
|
||||
import mock
|
||||
|
||||
from neutron.common import config
|
||||
from neutron import context
|
||||
from neutron.plugins.cisco.common import cisco_constants as const
|
||||
from neutron.plugins.cisco.common import config as cisco_config
|
||||
from neutron.plugins.cisco.models import virt_phy_sw_v2
|
||||
from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
|
||||
from neutron.tests import base
|
||||
from neutron.tests.unit import test_api_v2
|
||||
|
||||
|
||||
class TestCiscoPluginModel(base.BaseTestCase):
|
||||
|
||||
def setUp(self):
|
||||
# Point config file to: neutron/tests/etc/neutron.conf.test
|
||||
args = ['--config-file', test_api_v2.etcdir('neutron.conf.test')]
|
||||
config.parse(args=args)
|
||||
|
||||
super(TestCiscoPluginModel, self).setUp()
|
||||
|
||||
self.addCleanup(cisco_config.CONF.reset)
|
||||
|
||||
def test_non_nexus_device_driver(self):
|
||||
"""Tests handling of an non-Nexus device driver being configured."""
|
||||
with mock.patch.dict(sys.modules, {'mock_driver': mock.Mock()}):
|
||||
cisco_config.CONF.set_override('nexus_driver',
|
||||
'mock_driver.Non_Nexus_Driver',
|
||||
'CISCO')
|
||||
# Plugin model instance should have is_nexus_plugin set to False
|
||||
model = virt_phy_sw_v2.VirtualPhysicalSwitchModelV2()
|
||||
self.assertFalse(model.is_nexus_plugin)
|
||||
|
||||
# Model's _invoke_nexus_for_net_create should just return False
|
||||
user_id = 'user_id'
|
||||
tenant_id = 'tenant_id'
|
||||
ctx = context.Context(user_id, tenant_id)
|
||||
self.assertFalse(model._invoke_nexus_for_net_create(
|
||||
ctx, tenant_id, net_id='net_id',
|
||||
instance_id='instance_id', host_id='host_id'))
|
||||
|
||||
def test_nexus_plugin_calls_ignored_if_plugin_not_loaded(self):
|
||||
"""Verifies Nexus plugin calls are ignored if plugin is not loaded."""
|
||||
cisco_config.CONF.set_override(const.NEXUS_PLUGIN,
|
||||
None, 'CISCO_PLUGINS')
|
||||
with mock.patch.object(cisco_nexus_plugin_v2.NexusPlugin,
|
||||
'create_network') as mock_create_network:
|
||||
model = virt_phy_sw_v2.VirtualPhysicalSwitchModelV2()
|
||||
model._invoke_plugin_per_device(model, const.NEXUS_PLUGIN,
|
||||
'create_network')
|
||||
self.assertFalse(mock_create_network.called)
|
||||
Loading…
Reference in New Issue
Block a user