# @author: Abhishek Raut, Cisco Systems Inc.
from mock import patch
-from oslo.config import cfg
from neutron.api import extensions as neutron_extensions
from neutron.api.v2 import attributes
n1kv_neutron_plugin.N1kvNeutronPluginV2._setup_vsm = _fake_setup_vsm
neutron_extensions.append_api_extensions_path(extensions.__path__)
- self.addCleanup(cfg.CONF.reset)
ext_mgr = NetworkProfileTestExtensionManager()
# Save the original RESOURCE_ATTRIBUTE_MAP
for group in config[module]:
for opt, val in config[module][group].items():
module.cfg.CONF.set_override(opt, val, group)
- self.addCleanup(module.cfg.CONF.reset)
# Configure the Nexus switch dictionary
# TODO(Henry): add tests for other devices
super(TestCiscoRouterInterfacesV2, self).setUp()
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
- self.addCleanup(cisco_config.CONF.reset)
@contextlib.contextmanager
def _network_subnet_router(self):
'provider_vlan_auto_create', auto_create, 'CISCO')
cfg.CONF.set_override(
'provider_vlan_auto_trunk', auto_trunk, 'CISCO')
- self.addCleanup(cfg.CONF.reset)
with mock.patch.object(cdb, 'is_provider_vlan',
return_value=True) as mock_db:
# Create a provider network
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()}):
mock_lbaas_agent = mock.MagicMock()
get_lbaas_agent_patcher.start().return_value = mock_lbaas_agent
mock_lbaas_agent.__getitem__.return_value = {'host': 'host'}
- self.addCleanup(cfg.CONF.reset)
self._subnet_id = _subnet_id
class TestLoadBalancer(LoadBalancerPluginDbTestCase):
- def setUp(self):
- self.addCleanup(cfg.CONF.reset)
- super(TestLoadBalancer, self).setUp()
def test_create_vip(self, **extras):
expected = {
def setUp(self):
cfg.CONF.set_override('admin_password', "admin123", 'heleos')
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(db.clear_db)
super(TestEmbraneL3NatDBTestCase, self).setUp()
def setUp(self):
cfg.CONF.set_override('admin_password', "admin123", 'heleos')
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(db.clear_db)
super(EmbranePluginV2TestCase, self).setUp(self._plugin_name)
def setUp(self):
super(TestHyperVNeutronAgent, self).setUp()
- self.addCleanup(cfg.CONF.reset)
# Avoid rpc initialization for unit tests
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
self.assertTrue(sdnve_neutron_agent.create_agent_config_map(cfg.CONF))
def test_create_agent_config_using_controller_ips(self):
- self.addCleanup(cfg.CONF.reset)
cfg.CONF.set_override('controller_ips',
['10.10.10.1', '10.10.10.2'], group='SDNVE')
cfgmap = sdnve_neutron_agent.create_agent_config_map(cfg.CONF)
self.assertEqual(cfgmap['controller_ip'], '10.10.10.1')
def test_create_agent_config_using_interface_mappings(self):
- self.addCleanup(cfg.CONF.reset)
cfg.CONF.set_override('interface_mappings',
['interface1 : eth1', 'interface2 : eth2'],
group='SDNVE')
def setUp(self):
super(TestSdnveNeutronAgent, self).setUp()
- self.addCleanup(cfg.CONF.reset)
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
import mock
-from oslo.config import cfg
from neutron.openstack.common import uuidutils
from neutron.plugins.ibm.common import constants
def setUp(self):
super(TestSdnveApi, self).setUp()
- self.addCleanup(cfg.CONF.reset)
class MockKeystoneClient(object):
def __init__(self, **kwargs):
def setUp(self):
super(TestLinuxBridge, self).setUp()
- self.addCleanup(cfg.CONF.reset)
interface_mappings = {'physnet1': 'eth1'}
root_helper = cfg.CONF.AGENT.root_helper
class TestLinuxBridgeRpcCallbacks(base.BaseTestCase):
def setUp(self):
cfg.CONF.set_override('local_ip', LOCAL_IP, 'VXLAN')
- self.addCleanup(cfg.CONF.reset)
super(TestLinuxBridgeRpcCallbacks, self).setUp()
self.u_execute_p = mock.patch('neutron.agent.linux.utils.execute')
for opt, val in ml2_opts.items():
ml2_config.cfg.CONF.set_override(opt, val, 'ml2')
- self.addCleanup(ml2_config.cfg.CONF.reset)
-
def mocked_brocade_init(self):
self._driver = mock.MagicMock()
}
for opt, val in ml2_opts.items():
ml2_config.cfg.CONF.set_override(opt, val, 'ml2')
- self.addCleanup(ml2_config.cfg.CONF.reset)
# Configure the ML2 VLAN parameters
phys_vrange = ':'.join([PHYS_NET, str(VLAN_START), str(VLAN_END)])
vlan_config.cfg.CONF.set_override('network_vlan_ranges',
[phys_vrange],
'ml2_type_vlan')
- self.addCleanup(vlan_config.cfg.CONF.reset)
# Configure the Cisco Nexus mechanism driver
nexus_config = {
}
for opt, val in ml2_opts.items():
ml2_config.cfg.CONF.set_override(opt, val, 'ml2')
- self.addCleanup(ml2_config.cfg.CONF.reset)
# Configure the ML2 VLAN parameters
phys_vrange = ':'.join([PHYS_NET, str(VLAN_START), str(VLAN_END)])
vlan_config.cfg.CONF.set_override('network_vlan_ranges',
[phys_vrange],
'ml2_type_vlan')
- self.addCleanup(vlan_config.cfg.CONF.reset)
super(TestBigSwitchMechDriverBase,
self).setUp(ML2_PLUGIN)
'l2population'],
'ml2')
super(TestL2PopulationRpcTestCase, self).setUp(PLUGIN_NAME)
- self.addCleanup(config.cfg.CONF.reset)
self.adminContext = context.get_admin_context()
config.cfg.CONF.set_override('mechanism_drivers',
['logger', 'ncs'],
'ml2')
- self.addCleanup(config.cfg.CONF.reset)
super(NCSTestCase, self).setUp(PLUGIN_NAME)
self.port_create_status = 'DOWN'
mechanism_ncs.NCSMechanismDriver.sendjson = self.check_sendjson
self.phys_vrange = ':'.join([self.physnet, self.vlan_range])
config.cfg.CONF.set_override('network_vlan_ranges', [self.phys_vrange],
group='ml2_type_vlan')
- self.addCleanup(config.cfg.CONF.reset)
super(Ml2PluginV2TestCase, self).setUp(PLUGIN_NAME,
service_plugins=service_plugins)
self.port_create_status = 'DOWN'
config.cfg.CONF.set_override('mechanism_drivers',
['logger', 'test'],
'ml2')
- self.addCleanup(config.cfg.CONF.reset)
super(PortBindingTestCase, self).setUp(PLUGIN_NAME)
self.port_create_status = 'DOWN'
self.plugin = manager.NeutronManager.get_plugin()
self.driver.vxlan_vni_ranges = TUNNEL_RANGES
self.driver._sync_vxlan_allocations()
self.session = db.get_session()
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(db.clear_db)
def test_vxlan_tunnel_type(self):
def setUp(self):
super(TestNecAgentBase, self).setUp()
- self.addCleanup(cfg.CONF.reset)
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
cfg.CONF.set_override('host', 'dummy-host')
def setUp(self):
super(TestOneConvergenceAgentBase, self).setUp()
- self.addCleanup(cfg.CONF.reset)
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
with contextlib.nested(
self.assertTrue(ovs_neutron_agent.create_agent_config_map(cfg.CONF))
def test_create_agent_config_map_fails_for_invalid_tunnel_config(self):
- self.addCleanup(cfg.CONF.reset)
# 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],
ovs_neutron_agent.create_agent_config_map(cfg.CONF)
def test_create_agent_config_map_enable_tunneling(self):
- self.addCleanup(cfg.CONF.reset)
# 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')
self.assertEqual(cfgmap['tunnel_types'], [p_const.TYPE_GRE])
def test_create_agent_config_map_fails_no_local_ip(self):
- self.addCleanup(cfg.CONF.reset)
# 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):
ovs_neutron_agent.create_agent_config_map(cfg.CONF)
def test_create_agent_config_map_fails_for_invalid_tunnel_type(self):
- self.addCleanup(cfg.CONF.reset)
cfg.CONF.set_override('tunnel_types', ['foobar'], group='AGENT')
with testtools.ExpectedException(ValueError):
ovs_neutron_agent.create_agent_config_map(cfg.CONF)
def test_create_agent_config_map_multiple_tunnel_types(self):
- self.addCleanup(cfg.CONF.reset)
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')
def setUp(self):
super(TestOvsNeutronAgent, self).setUp()
- self.addCleanup(cfg.CONF.reset)
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
def setUp(self):
super(AncillaryBridgesTest, self).setUp()
- self.addCleanup(cfg.CONF.reset)
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
class TestLbaasService(base.BaseTestCase):
- def setUp(self):
- super(TestLbaasService, self).setUp()
- self.addCleanup(cfg.CONF.reset)
-
def test_start(self):
with mock.patch.object(
agent.rpc_service.Service, 'start'
self.saved_attr_map[resource] = attrs.copy()
ext_mgr = AgentTestExtensionManager()
self.addCleanup(self.restore_resource_attribute_map)
- self.addCleanup(cfg.CONF.reset)
super(AgentDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def restore_resource_attribute_map(self):
# under the License.
import mock
-from oslo.config import cfg
from neutron.agent import netns_cleanup_util as util
from neutron.tests import base
class TestNetnsCleanup(base.BaseTestCase):
- def setUp(self):
- super(TestNetnsCleanup, self).setUp()
- self.addCleanup(cfg.CONF.reset)
def test_kill_dhcp(self, dhcp_active=True):
conf = mock.Mock()
import contextlib
import itertools
import mock
-from oslo.config import cfg
from neutron.agent.linux import ip_lib
from neutron.agent.linux import ovs_lib
class TestOVSCleanup(base.BaseTestCase):
- def setUp(self):
- super(TestOVSCleanup, self).setUp()
- self.addCleanup(cfg.CONF.reset)
def test_setup_conf(self):
conf = util.setup_conf()
instance._NeutronPluginBaseV2__native_pagination_support = True
instance._NeutronPluginBaseV2__native_sorting_support = True
self.addCleanup(self._plugin_patcher.stop)
- self.addCleanup(cfg.CONF.reset)
api = router.APIRouter()
self.api = webtest.TestApp(api)
self._plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = self._plugin_patcher.start()
self.addCleanup(self._plugin_patcher.stop)
- self.addCleanup(cfg.CONF.reset)
router.SUB_RESOURCES['dummy'] = {
'collection_name': 'dummies',
self._plugin_patcher.stop()
self.api = None
self.plugin = None
- cfg.CONF.reset()
# Restore the global RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
self.do_alembic_cmd_p = mock.patch.object(cli, 'do_alembic_command')
self.do_alembic_cmd = self.do_alembic_cmd_p.start()
self.addCleanup(self.do_alembic_cmd_p.stop)
- self.addCleanup(cli.CONF.reset)
def _main_test_helper(self, argv, func_name, exp_args=(), exp_kwargs={}):
with mock.patch.object(sys, 'argv', argv):
# NOTE(jkoelker) for a 'pluggable' framework, Neutron sure
# doesn't like when the plugin changes ;)
db.clear_db()
- cfg.CONF.reset()
# Restore the original attribute map
attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk
super(NeutronDbPluginV2TestCase, self).tearDown()
self.plugin_p.stop()
self.mock_makedirs_p.stop()
self.mock_init_p.stop()
- cfg.CONF.reset()
super(TestDhcpAgentEventHandler, self).tearDown()
def _enable_dhcp_helper(self, isolated_metadata=False):
self.dvr_cls_p.stop()
self.device_exists_p.stop()
self.iproute_cls_p.stop()
- cfg.CONF.reset()
super(TestDeviceManager, self).tearDown()
def _test_setup_helper(self, device_exists, reuse_existing=False,
attributes.RESOURCE_ATTRIBUTE_MAP.update(
extattr.EXTENDED_ATTRIBUTES_2_0)
self.agentscheduler_dbMinxin = manager.NeutronManager.get_plugin()
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(self.restore_attribute_map)
quota.QUOTAS._driver = None
ext_mgr = ProviderExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.addCleanup(self._plugin_patcher.stop)
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(self._restore_attribute_map)
self.api = webtest.TestApp(router.APIRouter())
# If test_config specifies some config-file, use it, as well
config.parse(args=args)
self.setup_coreplugin()
- self.addCleanup(cfg.CONF.reset)
self.useFixture(
fixtures.MonkeyPatch('neutron.manager.NeutronManager._instance'))
self.api = None
self.plugin = None
db.clear_db()
- cfg.CONF.reset()
# Restore the global RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
self.setup_coreplugin(plugin)
cfg.CONF.set_override('service_plugins', [])
cfg.CONF.set_override('quota_router', -1, group='QUOTAS')
- self.addCleanup(cfg.CONF.reset)
# Ensure existing ExtensionManager is not used
def tearDown(self):
self.api = None
db.clear_db()
- cfg.CONF.reset()
super(RouterServiceInsertionTestCase, self).tearDown()
def _setup_core_resources(self):
cfg.CONF.set_override('service_plugins',
["%s.%s" % (dp.__name__,
dp.DummyServicePlugin.__name__)])
- self.addCleanup(cfg.CONF.reset)
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
ext_mgr = TestServiceTypeExtensionManager()
cfg.CONF.set_override('metadata_mode', None, 'NSX')
self.addCleanup(self.fc.reset_all)
self.addCleanup(self.restore_resource_attribute_map)
- self.addCleanup(cfg.CONF.reset)
super(MacLearningDBTestCase, self).setUp(plugin=PLUGIN_NAME,
ext_mgr=ext_mgr)
# Update the plugin and extensions path
self.setup_coreplugin(plugin)
- self.addCleanup(cfg.CONF.reset)
_plugin_patcher = mock.patch(plugin, autospec=True)
self.plugin = _plugin_patcher.start()
self.addCleanup(self.fc.reset_all)
self.addCleanup(patch_sync.stop)
self.addCleanup(self.mock_nsx_api.stop)
- self.addCleanup(cfg.CONF.reset)
def _test_gateway_subnet_notification(self, gateway='10.0.0.1'):
cfg.CONF.set_override('metadata_mode', 'dhcp_host_route', 'NSX')
self.mock_lsn_api = self.mock_lsn_api_p.start()
nsx.register_dhcp_opts(cfg)
nsx.register_metadata_opts(cfg)
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(self.mock_lsn_api_p.stop)
def test_lsn_get(self):
self.context = context.get_admin_context()
self.mock_lsn_api_p = mock.patch.object(lsn_man, 'lsn_api')
self.mock_lsn_api = self.mock_lsn_api_p.start()
- self.addCleanup(cfg.CONF.reset)
self.addCleanup(self.mock_lsn_api_p.stop)
self.addCleanup(db.clear_db)
self.addCleanup(mock_nm_get_plugin.stop)
self.addCleanup(mock_nm_get_service_plugins.stop)
- def tearDown(self):
- cfg.CONF.reset()
- super(SyncTestCase, self).tearDown()
-
@contextlib.contextmanager
def _populate_data(self, ctx, net_size=2, port_size=2, router_size=2):
Code Review / openstack-build / neutron-build.git / commitdiff