{'key': k, 'exp': v, 'act': actual_superset[k]})
-class ProcessMonitorFixture(tools.SafeFixture):
+class ProcessMonitorFixture(fixtures.Fixture):
"""Test fixture to capture and cleanup any spawn process monitor."""
- def setUp(self):
- super(ProcessMonitorFixture, self).setUp()
+
+ def _setUp(self):
self.old_callable = (
external_process.ProcessMonitor._spawn_checking_thread)
p = mock.patch("neutron.agent.linux.external_process.ProcessMonitor."
cfg.CONF.set_override("notification_driver", notification_driver)
-class PluginFixture(tools.SafeFixture):
+class PluginFixture(fixtures.Fixture):
def __init__(self, core_plugin=None):
super(PluginFixture, self).__init__()
self.core_plugin = core_plugin
- def setUp(self):
- super(PluginFixture, self).setUp()
+ def _setUp(self):
self.dhcp_periodic_p = mock.patch(
'neutron.db.agentschedulers_db.DhcpAgentSchedulerDbMixin.'
'start_periodic_dhcp_agent_status_check')
# under the License.
#
+import fixtures
+
from neutron.agent.linux import ip_lib
from neutron.tests.common import net_helpers
-from neutron.tests import tools
-class FakeMachine(tools.SafeFixture):
+class FakeMachine(fixtures.Fixture):
"""Create a fake machine.
:ivar bridge: bridge on which the fake machine is bound
self.ip = self.ip_cidr.partition('/')[0]
self.gateway_ip = gateway_ip
- def setUp(self):
- super(FakeMachine, self).setUp()
+ def _setUp(self):
ns_fixture = self.useFixture(
net_helpers.NamespaceFixture())
self.namespace = ns_fixture.name
net_helpers.assert_no_ping(self.namespace, dst_ip)
-class PeerMachines(tools.SafeFixture):
+class PeerMachines(fixtures.Fixture):
"""Create 'amount' peered machines on an ip_cidr.
:ivar bridge: bridge on which peer machines are bound
self.ip_cidr = ip_cidr or self.CIDR
self.gateway_ip = gateway_ip
- def setUp(self):
- super(PeerMachines, self).setUp()
+ def _setUp(self):
self.machines = []
for index in range(self.AMOUNT):
import signal
import subprocess
+import fixtures
import netaddr
from oslo_utils import uuidutils
import six
setattr(self, proc_attr, None)
-class NamespaceFixture(tools.SafeFixture):
+class NamespaceFixture(fixtures.Fixture):
"""Create a namespace.
:ivar ip_wrapper: created namespace
super(NamespaceFixture, self).__init__()
self.prefix = prefix
- def setUp(self):
- super(NamespaceFixture, self).setUp()
+ def _setUp(self):
ip = ip_lib.IPWrapper()
self.name = self.prefix + uuidutils.generate_uuid()
- self.ip_wrapper = ip.ensure_namespace(self.name)
self.addCleanup(self.destroy)
+ self.ip_wrapper = ip.ensure_namespace(self.name)
def destroy(self):
if self.ip_wrapper.netns.exists(self.name):
self.ip_wrapper.netns.delete(self.name)
-class VethFixture(tools.SafeFixture):
+class VethFixture(fixtures.Fixture):
"""Create a veth.
:ivar ports: created veth ports
:type ports: IPDevice 2-uplet
"""
- def setUp(self):
- super(VethFixture, self).setUp()
+ def _setUp(self):
ip_wrapper = ip_lib.IPWrapper()
self.ports = common_base.create_resource(
@six.add_metaclass(abc.ABCMeta)
-class PortFixture(tools.SafeFixture):
+class PortFixture(fixtures.Fixture):
"""Create a port.
:ivar port: created port
pass
@abc.abstractmethod
- def setUp(self):
- super(PortFixture, self).setUp()
+ def _setUp(self):
+ super(PortFixture, self)._setUp()
if not self.bridge:
self.bridge = self.useFixture(self._create_bridge_fixture()).bridge
tools.fail('Unexpected bridge type: %s' % type(bridge))
-class OVSBridgeFixture(tools.SafeFixture):
+class OVSBridgeFixture(fixtures.Fixture):
"""Create an OVS bridge.
:ivar prefix: bridge name prefix
super(OVSBridgeFixture, self).__init__()
self.prefix = prefix
- def setUp(self):
- super(OVSBridgeFixture, self).setUp()
+ def _setUp(self):
ovs = ovs_lib.BaseOVS()
self.bridge = common_base.create_resource(self.prefix, ovs.add_bridge)
self.addCleanup(self.bridge.destroy)
def _create_bridge_fixture(self):
return OVSBridgeFixture()
- def setUp(self):
- super(OVSPortFixture, self).setUp()
+ def _setUp(self):
+ super(OVSPortFixture, self)._setUp()
port_name = common_base.create_resource(PORT_PREFIX, self.create_port)
self.addCleanup(self.bridge.delete_port, port_name)
return name
-class LinuxBridgeFixture(tools.SafeFixture):
+class LinuxBridgeFixture(fixtures.Fixture):
"""Create a linux bridge.
:ivar bridge: created bridge
:type namespace: str
"""
- def setUp(self):
- super(LinuxBridgeFixture, self).setUp()
-
+ def _setUp(self):
self.namespace = self.useFixture(NamespaceFixture()).name
self.bridge = common_base.create_resource(
BR_PREFIX,
def _create_bridge_fixture(self):
return LinuxBridgeFixture()
- def setUp(self):
- super(LinuxBridgePortFixture, self).setUp()
+ def _setUp(self):
+ super(LinuxBridgePortFixture, self)._setUp()
self.port, self.br_port = self.useFixture(VethFixture()).ports
# bridge side
len(self.ports))
-class VethBridgeFixture(tools.SafeFixture):
+class VethBridgeFixture(fixtures.Fixture):
"""Simulate a bridge with a veth.
:ivar bridge: created bridge
:type bridge: FakeBridge
"""
- def setUp(self):
- super(VethBridgeFixture, self).setUp()
+ def _setUp(self):
ports = self.useFixture(VethFixture()).ports
self.bridge = VethBridge(ports)
def _create_bridge_fixture(self):
return VethBridgeFixture()
- def setUp(self):
- super(VethPortFixture, self).setUp()
+ def _setUp(self):
+ super(VethPortFixture, self)._setUp()
self.port = self.bridge.allocate_port()
ns_ip_wrapper = ip_lib.IPWrapper(self.namespace)
import os.path
import tempfile
+import fixtures
import six
from neutron.common import constants
from neutron.tests import base
from neutron.tests.common import helpers as c_helpers
from neutron.tests.common import net_helpers
-from neutron.tests import tools
class ConfigDict(base.AttributeDict):
self.convert_to_attr_dict(value)
-class ConfigFileFixture(tools.SafeFixture):
+class ConfigFileFixture(fixtures.Fixture):
"""A fixture that knows how to translate configurations to files.
:param base_filename: the filename to use on disk.
self.config = config
self.temp_dir = temp_dir
- def setUp(self):
- super(ConfigFileFixture, self).setUp()
+ def _setUp(self):
config_parser = self.dict_to_config_parser(self.config)
# Need to randomly generate a unique folder to put the file in
self.filename = os.path.join(self.temp_dir, self.base_filename)
return config_parser
-class ConfigFixture(tools.SafeFixture):
+class ConfigFixture(fixtures.Fixture):
"""A fixture that holds an actual Neutron configuration.
Note that 'self.config' is intended to only be updated once, during
self.temp_dir = temp_dir
self.base_filename = base_filename
- def setUp(self):
- super(ConfigFixture, self).setUp()
+ def _setUp(self):
cfg_fixture = ConfigFileFixture(
self.base_filename, self.config, self.temp_dir)
self.useFixture(cfg_fixture)
from neutron.tests import base
from neutron.tests.common import net_helpers
from neutron.tests.fullstack import config_fixtures
-from neutron.tests import tools
LOG = logging.getLogger(__name__)
DEFAULT_LOG_DIR = '/tmp/fullstack-logs/'
-class ProcessFixture(tools.SafeFixture):
+class ProcessFixture(fixtures.Fixture):
def __init__(self, test_name, process_name, exec_name, config_filenames):
super(ProcessFixture, self).__init__()
self.test_name = test_name
self.config_filenames = config_filenames
self.process = None
- def setUp(self):
- super(ProcessFixture, self).setUp()
+ def _setUp(self):
+ self.addCleanup(self.stop)
self.start()
def start(self):
def stop(self):
self.process.stop(block=True)
- def cleanUp(self, *args, **kwargs):
- self.stop()
- super(ProcessFixture, self).cleanUp(*args, **kwargs)
+class RabbitmqEnvironmentFixture(fixtures.Fixture):
-class RabbitmqEnvironmentFixture(tools.SafeFixture):
- def setUp(self):
- super(RabbitmqEnvironmentFixture, self).setUp()
-
+ def _setUp(self):
self.user = base.get_rand_name(prefix='user')
self.password = base.get_rand_name(prefix='pass')
self.vhost = base.get_rand_name(prefix='vhost')
utils.execute(cmd, run_as_root=True)
-class FullstackFixture(tools.SafeFixture):
+class FullstackFixture(fixtures.Fixture):
def __init__(self):
super(FullstackFixture, self).__init__()
self.test_name = None
- def setUp(self):
- super(FullstackFixture, self).setUp()
-
+ def _setUp(self):
self.temp_dir = self.useFixture(fixtures.TempDir()).path
rabbitmq_environment = self.useFixture(RabbitmqEnvironmentFixture())
return False
-class NeutronServerFixture(tools.SafeFixture):
+class NeutronServerFixture(fixtures.Fixture):
NEUTRON_SERVER = "neutron-server"
self.temp_dir = temp_dir
self.rabbitmq_environment = rabbitmq_environment
- def setUp(self):
- super(NeutronServerFixture, self).setUp()
-
+ def _setUp(self):
self.neutron_cfg_fixture = config_fixtures.NeutronConfigFixture(
self.temp_dir, cfg.CONF.database.connection,
self.rabbitmq_environment)
return client.Client(auth_strategy="noauth", endpoint_url=url)
-class OVSAgentFixture(tools.SafeFixture):
+class OVSAgentFixture(fixtures.Fixture):
NEUTRON_OVS_AGENT = "neutron-openvswitch-agent"
self.neutron_config = self.neutron_cfg_fixture.config
self.plugin_config = self.plugin_cfg_fixture.config
- def setUp(self):
- super(OVSAgentFixture, self).setUp()
-
+ def _setUp(self):
self.useFixture(net_helpers.OVSBridgeFixture(self._get_br_int_name()))
self.useFixture(net_helpers.OVSBridgeFixture(self._get_br_phys_name()))
return self.plugin_config.ovs.bridge_mappings.split(':')[1]
-class L3AgentFixture(tools.SafeFixture):
+class L3AgentFixture(fixtures.Fixture):
NEUTRON_L3_AGENT = "neutron-l3-agent"
self.neutron_config = self.neutron_cfg_fixture.config
self.integration_bridge_name = integration_bridge_name
- def setUp(self):
- super(L3AgentFixture, self).setUp()
-
+ def _setUp(self):
self.plugin_cfg_fixture = config_fixtures.L3ConfigFixture(
self.temp_dir, self.integration_bridge_name)
self.useFixture(self.plugin_cfg_fixture)
class SingleNodeEnvironment(f_fixtures.FullstackFixture):
- def setUp(self):
- super(SingleNodeEnvironment, self).setUp()
+ def _setUp(self):
+ super(SingleNodeEnvironment, self)._setUp()
neutron_config = self.neutron_server.neutron_cfg_fixture
ml2_config = self.neutron_server.plugin_cfg_fixture
# under the License.
import os
-from neutron.tests import tools
+import fixtures
-class RecursivePermDirFixture(tools.SafeFixture):
+class RecursivePermDirFixture(fixtures.Fixture):
"""Ensure at least perms permissions on directory and ancestors."""
def __init__(self, directory, perms):
self.directory = directory
self.least_perms = perms
- def setUp(self):
- super(RecursivePermDirFixture, self).setUp()
+ def _setUp(self):
previous_directory = None
current_directory = self.directory
while previous_directory != current_directory:
import abc
+import fixtures
import six
from neutron.common import exceptions as q_exc
from neutron import context
from neutron import manager
from neutron.tests import base
-from neutron.tests import tools
from neutron.tests.unit import testlib_api
@six.add_metaclass(abc.ABCMeta)
-class AbstractClientFixture(tools.SafeFixture):
+class AbstractClientFixture(fixtures.Fixture):
"""
Base class for a client that can interact the neutron api in some
manner.
super(PluginClientFixture, self).__init__()
self.plugin_conf = plugin_conf
- def setUp(self):
- super(PluginClientFixture, self).setUp()
+ def _setUp(self):
+ super(PluginClientFixture, self)._setUp()
self.useFixture(testlib_api.SqlFixture())
self.useFixture(self.plugin_conf)
self.useFixture(base.PluginFixture(self.plugin_conf.plugin_name))
import warnings
import fixtures
-from oslo_utils import excutils
import six
from neutron.api.v2 import attributes
-class SafeFixture(fixtures.Fixture):
- """Base Fixture ensuring cleanups are done even if setUp fails.
-
- Required until testtools/fixtures bugs #1456353 #1456370 are solved.
- """
-
- def __init__(self):
- unsafe_setup = self.setUp
- self.setUp = lambda: self.safe_setUp(unsafe_setup)
- self.initialized = True
-
- def setUp(self):
- assert getattr(self, 'initialized', True)
- super(SafeFixture, self).setUp()
-
- def safe_setUp(self, unsafe_setup):
- """Ensure cleanup is done even if setUp fails."""
- try:
- unsafe_setup()
- except Exception:
- with excutils.save_and_reraise_exception():
- self.safe_cleanUp()
-
- def safe_cleanUp(self):
- """Perform cleanUp if required.
-
- Fixture.addCleanup/cleanUp can be called only after Fixture.setUp
- successful call. It implies we cannot and don't need to call cleanUp
- if Fixture.setUp fails or is not called.
-
- This method assumes Fixture.setUp was called successfully if
- self._detail_sources is defined (Fixture.setUp last action).
- """
- root_setup_succeed = hasattr(self, '_detail_sources')
-
- if root_setup_succeed:
- self.cleanUp()
-
-
-class AttributeMapMemento(SafeFixture):
+class AttributeMapMemento(fixtures.Fixture):
"""Create a copy of the resource attribute map so it can be restored during
test cleanup.
- Inheritance is a bit of overkill for this facility and it's a
stretch to rationalize the "is a" criteria.
"""
- def setUp(self):
+
+ def _setUp(self):
# Shallow copy is not a proper choice for keeping a backup copy as
# the RESOURCE_ATTRIBUTE_MAP map is modified in place through the
# 0th level keys. Ideally deepcopy() would be used but this seems
# to result in test failures. A compromise is to copy one level
# deeper than a shallow copy.
- super(AttributeMapMemento, self).setUp()
self.contents_backup = {}
for res, attrs in six.iteritems(attributes.RESOURCE_ATTRIBUTE_MAP):
self.contents_backup[res] = attrs.copy()
attributes.RESOURCE_ATTRIBUTE_MAP = self.contents_backup
-class WarningsFixture(SafeFixture):
+class WarningsFixture(fixtures.Fixture):
"""Filters out warnings during test runs."""
warning_types = (
DeprecationWarning, PendingDeprecationWarning, ImportWarning
)
- def setUp(self):
- super(WarningsFixture, self).setUp()
+ def _setUp(self):
+ self.addCleanup(warnings.resetwarnings)
for wtype in self.warning_types:
warnings.filterwarnings(
"always", category=wtype, module='^neutron\\.')
- self.addCleanup(warnings.resetwarnings)
"""setup_mock_calls and verify_mock_calls are convenient methods
# under the License.
import functools
+
+import fixtures
import mock
import six
import testtools
from neutron.plugins.ml2 import models
from neutron.plugins.ml2 import plugin as ml2_plugin
from neutron.tests import base
-from neutron.tests import tools
from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit.agent import test_securitygroups_rpc as test_sg_rpc
from neutron.tests.unit.db import test_allowedaddresspairs_db as test_pair
# TODO(marun) - Move to somewhere common for reuse
-class PluginConfFixture(tools.SafeFixture):
+class PluginConfFixture(fixtures.Fixture):
"""Plugin configuration shared across the unit and functional tests."""
def __init__(self, plugin_name, parent_setup=None):
self.plugin_name = plugin_name
self.parent_setup = parent_setup
- def setUp(self):
- super(PluginConfFixture, self).setUp()
+ def _setUp(self):
if self.parent_setup:
self.parent_setup()
# License for the specific language governing permissions and limitations
# under the License.
+import fixtures
import six
import testtools
from neutron.db.migration.models import head # noqa
from neutron.db import model_base
from neutron.tests import base
-from neutron.tests import tools
from neutron import wsgi
return req
-class SqlFixture(tools.SafeFixture):
+class SqlFixture(fixtures.Fixture):
# flag to indicate that the models have been loaded
_TABLES_ESTABLISHED = False
- def setUp(self):
- super(SqlFixture, self).setUp()
+ def _setUp(self):
# Register all data models
engine = db_api.get_engine()
if not SqlFixture._TABLES_ESTABLISHED:
cliff>=1.13.0 # Apache-2.0
coverage>=3.6
-fixtures>=0.3.14
+fixtures>=1.3.1
mock>=1.0
python-subunit>=0.0.18
requests-mock>=0.6.0 # Apache-2.0
Code Review / openstack-build / neutron-build.git / commitdiff