self.callbacks_p = mock.patch(CALLBACKS)
self.spawn_p = mock.patch(SPAWN)
self.watch_p = mock.patch(CWATCH)
- self.addCleanup(mock.patch.stopall)
self.addCleanup(db.clear_db)
self.callbacks_p.start()
self.plugin_notifier_p.start()
def setUp(self):
super(BaseAgentTestCase, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.mod_agent = importutils.import_module(AGENTMOD)
# Used for SVI placement when round-robin placement is disabled.
mock.patch.object(cisco_config, 'first_device_ip',
new=NEXUS_IP_ADDR).start()
- self.addCleanup(mock.patch.stopall)
def _get_plugin_ref(self):
return getattr(NeutronManager.get_plugin(),
# according to the preceding switch config.
mock.patch.object(cisco_config, 'first_device_ip',
new=NEXUS_IP_ADDRESS).start()
- self.addCleanup(mock.patch.stopall)
def test_create_delete_networks(self):
"""Tests creation of two new Virtual Networks."""
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(mock.patch.stopall)
self.addCleanup(cfg.CONF.reset)
self._subnet_id = _subnet_id
self._mock_windows_version = mock.patch.object(utilsfactory,
'get_hypervutils')
self._mock_windows_version.start()
- self.addCleanup(mock.patch.stopall)
self._driver = sg_driver.HyperVSecurityGroupsDriver()
self._driver._utils = mock.MagicMock()
RESOURCE_ATTRIBUTE_MAP[item].
copy())
super(LinuxBridgeSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
- self.addCleanup(mock.patch.stopall)
def tearDown(self):
attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
driver
"""
- self.addCleanup(mock.patch.stopall)
# Configure the ML2 mechanism drivers and network types
ml2_opts = {
self._cisco_mech_driver = (mech_cisco_nexus.
CiscoNexusMechanismDriver())
- self.addCleanup(mock.patch.stopall)
self.addCleanup(db.clear_db)
def _create_delete_port(self, port_config):
uptime_patch = mock.patch(uptime, return_value=190)
uptime_patch.start()
- self.addCleanup(mock.patch.stopall)
self.addCleanup(db_api.clear_db)
def tearDown(self):
def setUp(self, plugin=None):
test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)
- self.addCleanup(mock.patch.stopall)
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
def setUp(self):
super(TestEswitchAgent, self).setUp()
- self.addCleanup(mock.patch.stopall)
# Avoid rpc initialization for unit tests
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
RESOURCE_ATTRIBUTE_MAP[item].
copy())
super(MlnxSecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
- self.addCleanup(mock.patch.stopall)
def tearDown(self):
attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
plugin_str = test_nec_plugin.PLUGIN_NAME
def setUp(self):
- # OvsDhcpAgentNotifierTestCase uses stop() for each mock.
- self.setup_nec_plugin_base(use_stop_each=True)
+ self.setup_nec_plugin_base()
super(NecDhcpAgentNotifierTestCase, self).setUp()
l3_plugin = None
def setUp(self):
- # OvsDhcpAgentNotifierTestCase uses stop() for each mock.
- self.setup_nec_plugin_base(use_stop_each=True)
+ self.setup_nec_plugin_base()
super(NecL3AgentNotifierTestCase, self).setUp()
def setUp(self):
super(TestNecAgentBase, self).setUp()
self.addCleanup(cfg.CONF.reset)
- self.addCleanup(mock.patch.stopall)
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
cfg.CONF.set_override('host', 'dummy-host')
os.remove(self.nec_ini_file)
self.nec_ini_file = None
- def patch_remote_calls(self, use_stop=False):
+ def patch_remote_calls(self):
self.plugin_notifier_p = mock.patch(NOTIFIER)
self.ofc_manager_p = mock.patch(OFC_MANAGER)
self.plugin_notifier_p.start()
self.ofc_manager_p.start()
- # When using mock.patch.stopall, we need to ensure
- # stop is not used anywhere in a single test.
- # In Neutron several tests use stop for each patched object,
- # so we need to take care of both cases.
- if use_stop:
- self.addCleanup(self.plugin_notifier_p.stop)
- self.addCleanup(self.ofc_manager_p.stop)
-
- def setup_nec_plugin_base(self, use_stop_all=True,
- use_stop_each=False):
- # If use_stop_each is set, use_stop_all cannot be set.
- if use_stop_all and not use_stop_each:
- self.addCleanup(mock.patch.stopall)
+
+ def setup_nec_plugin_base(self):
self._set_nec_ini()
self.addCleanup(self._clean_nec_ini)
- # kevinbenton: stopping patches is now handled in base test class
- self.patch_remote_calls(use_stop=False)
+ self.patch_remote_calls()
class NecPluginV2TestCase(NecPluginV2TestCaseBase,
self.callback_nec.update_ports(self.context, **kwargs)
def setUp(self, plugin=None, ext_mgr=None):
- self.addCleanup(mock.patch.stopall)
self._set_nec_ini()
self.addCleanup(self._clean_nec_ini)
# NOTE: enable_autocheck() is a feature of StubOFCDriver
self.ofc.driver.enable_autocheck()
self.ctx = context.get_admin_context()
- self.addCleanup(mock.patch.stopall)
def get_random_params(self):
"""create random parameters for portinfo test."""
self.driver = drivers.get_driver(self.driver)(TestConfig)
self.do_request = mock.patch.object(ofc.OFCClient,
'do_request').start()
- self.addCleanup(mock.patch.stopall)
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test."""
_plugin_name = test_nec_plugin.PLUGIN_NAME
def setUp(self):
- self.addCleanup(mock.patch.stopall)
mock.patch(test_nec_plugin.OFC_MANAGER).start()
super(NecRouterL3AgentTestCase, self).setUp(self._plugin_name)
def setUp(self, plugin=None):
test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)
- self.addCleanup(mock.patch.stopall)
mock.patch(NOTIFIER).start()
mock.patch(OFC_MANAGER).start()
self._attribute_map_bk_ = {}
self.driver = drivers.get_driver(self.driver_name)(TestConfig)
self.do_request = mock.patch.object(ofc_client.OFCClient,
'do_request').start()
- self.addCleanup(mock.patch.stopall)
def get_ofc_item_random_params(self):
"""create random parameters for ofc_item test."""
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()
def setUp(self):
super(TestOneConvergenceAgentBase, self).setUp()
self.addCleanup(cfg.CONF.reset)
- self.addCleanup(mock.patch.stopall)
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
with contextlib.nested(
self.nvsdlib = mock.Mock()
self.nvsdlib.create_network.side_effect = side_effect
- self.addCleanup(mock.patch.stopall)
-
with mock.patch.object(nvsd_plugin.OneConvergencePluginV2,
'oneconvergence_init',
new=mocked_oneconvergence_init):
self.nvsdlib = mock.Mock()
self.nvsdlib.create_network.side_effect = side_effect
- self.addCleanup(mock.patch.stopall)
ext_mgr = test_l3_plugin.L3TestExtensionManager()
with mock.patch.object(nvsd_plugin.OneConvergencePluginV2,
self.nvsdlib = mock.Mock()
self.nvsdlib.create_network.side_effect = side_effect
- self.addCleanup(mock.patch.stopall)
test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)
- self.addCleanup(mock.patch.stopall)
notifier_cls = mock.patch(AGENTNOTIFIER).start()
self.notifier = mock.Mock()
notifier_cls.return_value = self.notifier
self.br = ovs_lib.OVSBridge(self.BR_NAME, self.root_helper)
self.execute = mock.patch.object(
utils, "execute", spec=utils.execute).start()
- self.addCleanup(mock.patch.stopall)
def test_vifport(self):
"""Create and stringify vif port, confirm no exceptions."""
def setUp(self):
super(TestOvsNeutronAgent, self).setUp()
self.addCleanup(cfg.CONF.reset)
- self.addCleanup(mock.patch.stopall)
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)
- self.addCleanup(mock.patch.stopall)
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
def setUp(self, plugin=None):
test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)
- self.addCleanup(mock.patch.stopall)
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
cfg.CONF.set_override('report_interval', 0, 'AGENT')
- self.addCleanup(mock.patch.stopall)
self.INT_BRIDGE = 'integration_bridge'
self.TUN_BRIDGE = 'tunnel_bridge'
def setUp(self):
super(RyuAgentTestCase, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.fake_ryu = fake_ryu.patch_fake_ryu_client().start()
self.mod_agent = importutils.import_module(self._AGENT_NAME)
def setUp(self, plugin=None):
test_sg_rpc.set_firewall_driver(test_sg_rpc.FIREWALL_HYBRID_DRIVER)
- self.addCleanup(mock.patch.stopall)
self.fake_ryu = fake_ryu.patch_fake_ryu_client().start()
notifier_p = mock.patch(NOTIFIER)
notifier_cls = notifier_p.start()
class TestFwaasL3AgentRpcCallback(base.BaseTestCase):
def setUp(self):
super(TestFwaasL3AgentRpcCallback, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.conf = cfg.ConfigOpts()
self.conf.register_opts(base_config.core_opts)
class TestFWaaSAgentApi(base.BaseTestCase):
def setUp(self):
super(TestFWaaSAgentApi, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.api = api.FWaaSPluginApiMixin(
'topic',
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
self.looping_call_p.start()
- self.addCleanup(mock.patch.stopall)
-
def _create_router(self):
router = varmour_router.vArmourL3NATAgent(HOSTNAME, self.conf)
router.rest.server = FAKE_DIRECTOR
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
self.looping_call_p.start()
- self.addCleanup(mock.patch.stopall)
-
def _create_router(self):
router = varmour_router.vArmourL3NATAgent(HOSTNAME, self.conf)
router.rest.server = FAKE_DIRECTOR
class TestFirewallAgentApi(base.BaseTestCase):
def setUp(self):
super(TestFirewallAgentApi, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.api = fwaas_plugin.FirewallAgentApi('topic', 'host')
self.mock_fanoutcast = mock.patch.object(self.api,
class TestManager(base.BaseTestCase):
def setUp(self):
super(TestManager, self).setUp()
- self.addCleanup(mock.patch.stopall)
mock_conf = mock.Mock()
mock_conf.device_driver = ['devdriver']
class TestApiCache(base.BaseTestCase):
def setUp(self):
super(TestApiCache, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.api = api.LbaasAgentApi('topic', mock.sentinel.context, 'host')
self.make_msg = mock.patch.object(self.api, 'make_msg').start()
class TestHaproxyNSDriver(base.BaseTestCase):
def setUp(self):
super(TestHaproxyNSDriver, self).setUp()
- self.addCleanup(mock.patch.stopall)
conf = mock.Mock()
conf.haproxy.loadbalancer_state_path = '/the/path'
self.request_method_mock = mock.Mock()
requests.request = self.request_method_mock
self.testclient = self._get_nsclient()
- self.addCleanup(mock.patch.stopall)
def test_instantiate_nsclient_with_empty_uri(self):
"""Asserts that a call with empty URI will raise an exception."""
netscaler_driver.NetScalerPluginDriver(self.plugin_instance))
self.driver = self.plugin_instance.drivers[LBAAS_PROVIDER_NAME]
self.context = context.get_admin_context()
- self.addCleanup(mock.patch.stopall)
def test_create_vip(self):
with contextlib.nested(
radware_driver.rest_client.call = self.rest_call_mock
self.addCleanup(radware_driver.completion_handler.join)
- self.addCleanup(mock.patch.stopall)
def test_verify_workflow_templates(self):
"""Test the rest call failure handling by Exception raising."""
self.mock_importer = mock.patch.object(
agent_driver_base, 'importutils').start()
- self.addCleanup(mock.patch.stopall)
# needed to reload provider configuration
st_db.ServiceTypeManager._instance = None
'.LbaasAgentSchedulerDbMixin.get_lbaas_agents')
get_lbaas_agents_patcher.start()
- self.addCleanup(mock.patch.stopall)
-
def test_get_ready_devices(self):
with self.vip() as vip:
with mock.patch('neutron.services.loadbalancer.agent_scheduler'
class TestLoadBalancerAgentApi(base.BaseTestCase):
def setUp(self):
super(TestLoadBalancerAgentApi, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.api = agent_driver_base.LoadBalancerAgentApi('topic')
self.mock_cast = mock.patch.object(self.api, 'cast').start()
self.plugin_instance
))
- self.addCleanup(mock.patch.stopall)
-
def test_create_vip(self):
with self.subnet() as subnet:
with self.pool(subnet=subnet) as pool:
self.agent = metering_agent.MeteringAgent('my agent', cfg.CONF)
self.driver = self.agent.metering_driver
- self.addCleanup(mock.patch.stopall)
-
def tearDown(self):
test_notifier.NOTIFICATIONS = []
super(TestMeteringOperations, self).tearDown()
self.driver = mock.Mock()
self.agent.metering_driver = self.driver
- self.addCleanup(mock.patch.stopall)
-
def test_add_metering_label_with_bad_driver_impl(self):
del self.driver.add_metering_label
super(TestMeteringPlugin, self).setUp(plugin=plugin, ext_mgr=ext_mgr,
service_plugins=service_plugins)
- self.addCleanup(mock.patch.stopall)
-
self.uuid = '654f6b9d-0f36-4ae5-bd1b-01616794ca60'
uuid = 'neutron.openstack.common.uuidutils.generate_uuid'
self.uuid = '654f6b9d-0f36-4ae5-bd1b-01616794ca60'
- self.addCleanup(mock.patch.stopall)
-
uuid = 'neutron.openstack.common.uuidutils.generate_uuid'
self.uuid_patch = mock.patch(uuid, return_value=self.uuid)
self.mock_uuid = self.uuid_patch.start()
class TestCiscoCsrIPSecConnection(base.BaseTestCase):
def setUp(self):
super(TestCiscoCsrIPSecConnection, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.conn_info = {
u'id': '123',
u'status': constants.PENDING_CREATE,
def setUp(self):
super(TestCiscoCsrIPsecConnectionCreateTransforms, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.conn_info = {
u'id': '123',
u'status': constants.PENDING_CREATE,
def setUp(self):
super(TestCiscoCsrIPsecDeviceDriverSyncStatuses, self).setUp()
- self.addCleanup(mock.patch.stopall)
for klass in ['neutron.openstack.common.rpc.create_connection',
'neutron.context.get_admin_context_without_session',
'neutron.openstack.common.'
class TestCiscoCsrIPsecDeviceDriverConfigLoading(base.BaseTestCase):
- def setUp(self):
- super(TestCiscoCsrIPsecDeviceDriverConfigLoading, self).setUp()
- self.addCleanup(mock.patch.stopall)
-
def create_tempfile(self, contents):
(fd, path) = tempfile.mkstemp(prefix='test', suffix='.conf')
try:
class TestIPsecDeviceDriver(base.BaseTestCase):
def setUp(self, driver=ipsec_driver.OpenSwanDriver):
super(TestIPsecDeviceDriver, self).setUp()
- self.addCleanup(mock.patch.stopall)
for klass in [
'os.makedirs',
def setUp(self):
super(TestCiscoIPsecDriverValidation, self).setUp()
- self.addCleanup(mock.patch.stopall)
mock.patch('neutron.openstack.common.rpc.create_connection').start()
self.service_plugin = mock.Mock()
self.driver = ipsec_driver.CiscoCsrIPsecVPNDriver(self.service_plugin)
def setUp(self):
super(TestCiscoIPsecDriverMapping, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.context = mock.patch.object(n_ctx, 'Context').start()
self.session = self.context.session
self.query_mock = self.session.query.return_value.order_by
def setUp(self):
super(TestCiscoIPsecDriver, self).setUp()
- self.addCleanup(mock.patch.stopall)
dbapi.configure_db()
self.addCleanup(dbapi.clear_db)
mock.patch('neutron.openstack.common.rpc.create_connection').start()
class TestIPsecDriver(base.BaseTestCase):
def setUp(self):
super(TestIPsecDriver, self).setUp()
- self.addCleanup(mock.patch.stopall)
mock.patch('neutron.openstack.common.rpc.create_connection').start()
l3_agent = mock.Mock()
class TestVPNAgent(base.BaseTestCase):
def setUp(self):
super(TestVPNAgent, self).setUp()
- self.addCleanup(mock.patch.stopall)
self.conf = cfg.CONF
self.conf.register_opts(base_config.core_opts)
self.conf.register_opts(l3_agent.L3NATAgent.OPTS)
test_agent_scheduler.AgentSchedulerTestMixIn,
test_agent_ext_plugin.AgentDBTestMixIn):
def setUp(self):
- self.addCleanup(mock.patch.stopall)
self.adminContext = context.get_admin_context()
driver_cls_p = mock.patch(
'neutron.services.vpn.'
supported_extension_aliases=None,
use_quota=False,
):
- self.addCleanup(mock.patch.stopall)
self._resource_prefix = resource_prefix
self._plural_mappings = plural_mappings or {}
config.register_root_helper(cfg.CONF)
cfg.CONF.set_override('use_namespaces', True)
- self.addCleanup(mock.patch.stopall)
device_exists_p = mock.patch(
'neutron.agent.linux.ip_lib.device_exists', return_value=False)
device_exists_p.start()
self.driver_cls.return_value = self.driver
self.mock_makedirs_p = mock.patch("os.makedirs")
self.mock_makedirs = self.mock_makedirs_p.start()
- self.addCleanup(mock.patch.stopall)
def test_dhcp_agent_manager(self):
state_rpc_str = 'neutron.agent.rpc.PluginReportStateAPI'
self.iptables = (iptables_manager.
IptablesManager(root_helper=self.root_helper))
self.execute = mock.patch.object(self.iptables, "execute").start()
- self.addCleanup(mock.patch.stopall)
def test_binary_name(self):
self.assertEqual(iptables_manager.binary_name,
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
self.looping_call_p.start()
- self.addCleanup(mock.patch.stopall)
-
def test_router_info_create(self):
id = _uuid()
ri = l3_agent.RouterInfo(id, self.conf.root_helper,
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
looping_call_p.start()
self.agent = l3_agent.L3NATAgent(HOSTNAME)
- self.addCleanup(mock.patch.stopall)
def test_spawn_metadata_proxy(self):
router_id = _uuid()
self.execute_p = mock.patch('neutron.agent.linux.utils.execute')
self.safe = self.replace_p.start()
self.execute = self.execute_p.start()
- self.addCleanup(mock.patch.stopall)
class TestDhcpBase(TestBase):
config.register_root_helper(self.conf)
self.device_exists_p = mock.patch.object(ip_lib, 'device_exists')
self.device_exists = self.device_exists_p.start()
- self.addCleanup(mock.patch.stopall)
self.driver = interface.MidonetInterfaceDriver(self.conf)
self.network_id = uuidutils.generate_uuid()
self.port_id = uuidutils.generate_uuid()
looping_call_p = mock.patch(
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
self.looping_mock = looping_call_p.start()
- self.addCleanup(mock.patch.stopall)
self.cfg.CONF.metadata_proxy_socket = '/the/path'
self.cfg.CONF.metadata_workers = 0
self.cfg.CONF.metadata_backlog = 128
super(SecurityGroupAgentRpcTestCase, self).setUp()
self.agent = sg_rpc.SecurityGroupAgentRpcMixin()
self.agent.context = None
- self.addCleanup(mock.patch.stopall)
mock.patch('neutron.agent.linux.iptables_manager').start()
self.agent.root_helper = 'sudo'
self.agent.init_firewall(defer_refresh_firewall=defer_refresh_firewall)
'firewall_driver',
self.FIREWALL_DRIVER,
group='SECURITYGROUP')
- self.addCleanup(mock.patch.stopall)
self.agent = sg_rpc.SecurityGroupAgentRpcMixin()
self.agent.context = None
delete_gw_dev_patcher.start()
self.mock_get_gw_dev_status = get_gw_dev_status_patcher.start()
- self.addCleanup(mock.patch.stopall)
super(TestNetworkGateway,
self).setUp(plugin=plugin, ext_mgr=ext_mgr)
self.port_create_status = constants.PORT_STATUS_DOWN
cfg.CONF.set_override('metadata_mode', None, 'NSX')
self.addCleanup(self.fc.reset_all)
- self.addCleanup(mock.patch.stopall)
class TestBasicGet(test_plugin.TestBasicGet, NsxPluginV2TestCase):
class NsxUtilsTestCase(base.BaseTestCase):
- def setUp(self):
- self.addCleanup(mock.patch.stopall)
- super(NsxUtilsTestCase, self).setUp()
-
def _mock_port_mapping_db_calls(self, ret_value):
# Mock relevant db calls
# This will allow for avoiding setting up the plugin
return_value=ret_value).start()
mock.patch(nsx_method('add_neutron_nsx_security_group_mapping',
module_name='dbexts.db')).start()
- self.addCleanup(mock.patch.stopall)
def _verify_get_nsx_sec_profile_id(self, exp_sec_prof_uuid):
# The nsxlib and db calls are mocked, therefore the cluster
self.fc2.set_fake_nsx_api(self.fc)
self.addCleanup(self.fc2.reset_all)
- self.addCleanup(mock.patch.stopall)
def tearDown(self):
plugin = NeutronManager.get_plugin()