{'key': k, 'exp': v, 'act': actual_superset[k]})
-class ProcessMonitorFixture(fixtures.Fixture):
+class ProcessMonitorFixture(tools.SafeFixture):
"""Test fixture to capture and cleanup any spawn process monitor."""
def setUp(self):
super(ProcessMonitorFixture, self).setUp()
cfg.CONF.set_override("notification_driver", notification_driver)
-class PluginFixture(fixtures.Fixture):
+class PluginFixture(tools.SafeFixture):
def __init__(self, core_plugin=None):
+ super(PluginFixture, self).__init__()
self.core_plugin = core_plugin
def setUp(self):
# License for the specific language governing permissions and limitations
# 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(fixtures.Fixture):
+class FakeMachine(tools.SafeFixture):
"""Create a fake machine.
:ivar bridge: bridge on which the fake machine is bound
net_helpers.assert_no_ping(self.namespace, dst_ip)
-class PeerMachines(fixtures.Fixture):
+class PeerMachines(tools.SafeFixture):
"""Create 'amount' peered machines on an ip_cidr.
:ivar bridge: bridge on which peer machines are bound
import abc
-import fixtures
import netaddr
import six
{'ns': src_namespace, 'destination': dst_ip})
-class NamespaceFixture(fixtures.Fixture):
+class NamespaceFixture(tools.SafeFixture):
"""Create a namespace.
:ivar ip_wrapper: created namespace
self.ip_wrapper.netns.delete(self.name)
-class VethFixture(fixtures.Fixture):
+class VethFixture(tools.SafeFixture):
"""Create a veth.
:ivar ports: created veth ports
@six.add_metaclass(abc.ABCMeta)
-class PortFixture(fixtures.Fixture):
+class PortFixture(tools.SafeFixture):
"""Create a port.
:ivar port: created port
"""
def __init__(self, bridge=None, namespace=None):
+ super(PortFixture, self).__init__()
self.bridge = bridge
self.namespace = namespace
tools.fail('Unexpected bridge type: %s' % type(bridge))
-class OVSBridgeFixture(fixtures.Fixture):
+class OVSBridgeFixture(tools.SafeFixture):
"""Create an OVS bridge.
:ivar prefix: bridge name prefix
"""
def __init__(self, prefix=BR_PREFIX):
+ super(OVSBridgeFixture, self).__init__()
self.prefix = prefix
def setUp(self):
return name
-class LinuxBridgeFixture(fixtures.Fixture):
+class LinuxBridgeFixture(tools.SafeFixture):
"""Create a linux bridge.
:ivar bridge: created bridge
len(self.ports))
-class VethBridgeFixture(fixtures.Fixture):
+class VethBridgeFixture(tools.SafeFixture):
"""Simulate a bridge with a veth.
:ivar bridge: created bridge
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.functional.agent.linux import helpers
+from neutron.tests import tools
class ConfigDict(base.AttributeDict):
self.convert_to_attr_dict(value)
-class ConfigFileFixture(fixtures.Fixture):
+class ConfigFileFixture(tools.SafeFixture):
"""A fixture that knows how to translate configurations to files.
:param base_filename: the filename to use on disk.
return config_parser
-class ConfigFixture(fixtures.Fixture):
+class ConfigFixture(tools.SafeFixture):
"""A fixture that holds an actual Neutron configuration.
Note that 'self.config' is intended to only be updated once, during
is initializing a new instance of the class.
"""
def __init__(self, temp_dir, base_filename):
+ super(ConfigFixture, self).__init__()
self.config = ConfigDict()
self.temp_dir = temp_dir
self.base_filename = base_filename
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(fixtures.Fixture):
+class ProcessFixture(tools.SafeFixture):
def __init__(self, test_name, process_name, exec_name, config_filenames):
+ super(ProcessFixture, self).__init__()
self.test_name = test_name
self.process_name = process_name
self.exec_name = exec_name
super(ProcessFixture, self).cleanUp(*args, **kwargs)
-class RabbitmqEnvironmentFixture(fixtures.Fixture):
+class RabbitmqEnvironmentFixture(tools.SafeFixture):
def setUp(self):
super(RabbitmqEnvironmentFixture, self).setUp()
utils.execute(cmd, run_as_root=True)
-class FullstackFixture(fixtures.Fixture):
+class FullstackFixture(tools.SafeFixture):
def __init__(self):
+ super(FullstackFixture, self).__init__()
self.test_name = None
def setUp(self):
return False
-class NeutronServerFixture(fixtures.Fixture):
+class NeutronServerFixture(tools.SafeFixture):
NEUTRON_SERVER = "neutron-server"
def __init__(self, test_name, temp_dir, rabbitmq_environment):
+ super(NeutronServerFixture, self).__init__()
self.test_name = test_name
self.temp_dir = temp_dir
self.rabbitmq_environment = rabbitmq_environment
return client.Client(auth_strategy="noauth", endpoint_url=url)
-class OVSAgentFixture(fixtures.Fixture):
+class OVSAgentFixture(tools.SafeFixture):
NEUTRON_OVS_AGENT = "neutron-openvswitch-agent"
def __init__(self, test_name, neutron_cfg_fixture, ml2_cfg_fixture):
+ super(OVSAgentFixture, self).__init__()
self.test_name = test_name
self.neutron_cfg_fixture = neutron_cfg_fixture
self.plugin_cfg_fixture = ml2_cfg_fixture
return self.plugin_config.ovs.bridge_mappings.split(':')[1]
-class L3AgentFixture(fixtures.Fixture):
+class L3AgentFixture(tools.SafeFixture):
NEUTRON_L3_AGENT = "neutron-l3-agent"
def __init__(self, test_name, temp_dir,
neutron_cfg_fixture, integration_bridge_name):
+ super(L3AgentFixture, self).__init__()
self.test_name = test_name
self.temp_dir = temp_dir
self.neutron_cfg_fixture = neutron_cfg_fixture
import shlex
import subprocess
-import fixtures
-
from neutron.agent.common import config
from neutron.agent.linux import ip_lib
from neutron.agent.linux import utils
+from neutron.tests import tools
CHILD_PROCESS_TIMEOUT = os.environ.get('OS_TEST_CHILD_PROCESS_TIMEOUT', 20)
CHILD_PROCESS_SLEEP = os.environ.get('OS_TEST_CHILD_PROCESS_SLEEP', 0.5)
r'^.*\s+\d+\s+.*:(?P<port>\d+)\s+[0-9:].*')
-class RecursivePermDirFixture(fixtures.Fixture):
+class RecursivePermDirFixture(tools.SafeFixture):
"""Ensure at least perms permissions on directory and ancestors."""
def __init__(self, directory, perms):
--- /dev/null
+# Copyright (c) 2015 Thales Services SAS
+# All Rights Reserved.
+#
+# 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 fixtures
+import testscenarios
+
+from neutron.tests import base
+from neutron.tests import tools
+
+
+class NoErrorFixture(tools.SafeFixture):
+
+ def __init__(self):
+ super(NoErrorFixture, self).__init__()
+ self.cleaned = False
+ self.called = False
+
+ def setUp(self):
+ super(NoErrorFixture, self).setUp()
+ self.called = True
+
+ def cleanUp(self):
+ self.cleaned = True
+ super(NoErrorFixture, self).cleanUp()
+
+
+class ErrorAfterFixtureSetup(NoErrorFixture):
+
+ def setUp(self):
+ super(tools.SafeFixture, self).setUp()
+ raise ValueError
+
+
+class ErrorBeforeFixtureSetup(NoErrorFixture):
+
+ def setUp(self):
+ raise ValueError
+
+
+class TestSafeFixture(testscenarios.WithScenarios, base.BaseTestCase):
+ scenarios = [
+ ('testtools useFixture', dict(fixtures=False)),
+ ('fixtures useFixture', dict(fixtures=True)),
+ ]
+
+ def setUp(self):
+ super(TestSafeFixture, self).setUp()
+ if self.fixtures:
+ self.parent = self.useFixture(fixtures.Fixture())
+ else:
+ self.parent = self
+
+ def test_no_error(self):
+ fixture = NoErrorFixture()
+ self.parent.useFixture(fixture)
+ self.assertTrue(fixture.called)
+ self.assertFalse(fixture.cleaned)
+
+ def test_error_after_root_setup(self):
+ fixture = ErrorAfterFixtureSetup()
+ self.assertRaises(ValueError, self.parent.useFixture, fixture)
+ self.assertTrue(fixture.cleaned)
+
+ def test_error_before_root_setup(self):
+ fixture = ErrorBeforeFixtureSetup()
+ # NOTE(cbrandily); testtools.useFixture crashs badly if Fixture.setUp
+ # is not called or fails.
+ self.assertRaises(AttributeError, self.parent.useFixture, fixture)
+ self.assertFalse(fixture.cleaned)
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(fixtures.Fixture):
+class AbstractClientFixture(tools.SafeFixture):
"""
Base class for a client that can interact the neutron api in some
manner.
"""Targets the Neutron API via the plugin API"""
def __init__(self, plugin_conf):
+ super(PluginClientFixture, self).__init__()
self.plugin_conf = plugin_conf
def setUp(self):
import warnings
import fixtures
+from oslo_utils import excutils
import six
from neutron.api.v2 import attributes
-class AttributeMapMemento(fixtures.Fixture):
+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):
"""Create a copy of the resource attribute map so it can be restored during
test cleanup.
attributes.RESOURCE_ATTRIBUTE_MAP = self.contents_backup
-class WarningsFixture(fixtures.Fixture):
+class WarningsFixture(SafeFixture):
"""Filters out warnings during test runs."""
warning_types = (
import uuid
import webob
-import fixtures
from oslo_db import exception as db_exc
from sqlalchemy.orm import exc as sqla_exc
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(fixtures.Fixture):
+class PluginConfFixture(tools.SafeFixture):
"""Plugin configuration shared across the unit and functional tests."""
def __init__(self, plugin_name, parent_setup=None):
+ super(PluginConfFixture, self).__init__()
self.plugin_name = plugin_name
self.parent_setup = 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(fixtures.Fixture):
+class SqlFixture(tools.SafeFixture):
# flag to indicate that the models have been loaded
_TABLES_ESTABLISHED = False
Code Review / openstack-build / neutron-build.git / commitdiff