# License for the specific language governing permissions and limitations
# under the License.
-import copy
import os.path
from neutron import context
from neutron.api.v2 import attributes
from neutron.tests import base
+from neutron.tests import tools
TEST_PATH = os.path.dirname(os.path.abspath(__file__))
def setUp(self):
super(APIPolicyTestCase, self).setUp()
-
- self.ATTRIBUTE_MAP_COPY = copy.copy(attributes.RESOURCE_ATTRIBUTE_MAP)
+ self.useFixture(tools.AttributeMapMemento())
self.extension_path = os.path.abspath(os.path.join(
TEST_PATH, "../../../extensions"))
policy.reset()
True)
def tearDown(self):
- if self.ATTRIBUTE_MAP_COPY:
- attributes.RESOURCE_ATTRIBUTE_MAP = self.ATTRIBUTE_MAP_COPY
+ policy.reset()
super(APIPolicyTestCase, self).tearDown()
# License for the specific language governing permissions and limitations
# under the License.
+import fixtures
+
+from neutron.api.v2 import attributes
+
+
+class AttributeMapMemento(fixtures.Fixture):
+ """Create a copy of the resource attribute map so it can be restored during
+ test cleanup.
+
+ There are a few reasons why this is not included in a class derived
+ from BaseTestCase:
+
+ - Test cases may need more control about when the backup is
+ made, especially if they are not direct descendants of
+ BaseTestCase.
+
+ - Inheritance is a bit of overkill for this facility and it's a
+ stretch to rationalize the "is a" criteria.
+ """
+ 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 resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
+ self.contents_backup[resource] = attrs.copy()
+ self.addCleanup(self.restore)
+
+ def restore(self):
+ attributes.RESOURCE_ATTRIBUTE_MAP = self.contents_backup
+
"""setup_mock_calls and verify_mock_calls are convenient methods
to setup a sequence of mock calls.
from neutron import quota
from neutron.tests import base
from neutron.tests import fake_notifier
+from neutron.tests import tools
from neutron.tests.unit import testlib_api
plugin = 'neutron.tests.unit.api.v2.test_base.TestSubresourcePlugin'
extensions.PluginAwareExtensionManager._instance = None
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
self.config_parse()
self.setup_coreplugin(plugin)
def tearDown(self):
router.SUB_RESOURCES = {}
- # Restore the global RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
super(SubresourceTest, self).tearDown()
def test_index_sub_resource(self):
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
# Create the default configurations
self.config_parse()
super(ExtensionTestCase, self).tearDown()
self.api = None
self.plugin = None
- # Restore the global RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
def test_extended_create(self):
net_id = _uuid()
from neutron.db import models_v2
from neutron import manager
from neutron.tests import base
+from neutron.tests import tools
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit import testlib_api
extensions.PluginAwareExtensionManager._instance = None
# Save the attributes map in case the plugin will alter it
# loading extensions
- # Note(salvatore-orlando): shallow copy is not good enough in
- # this case, but copy.deepcopy does not seem to work, since it
- # causes test failures
- self._attribute_map_bk = {}
- for item in attributes.RESOURCE_ATTRIBUTE_MAP:
- self._attribute_map_bk[item] = (attributes.
- RESOURCE_ATTRIBUTE_MAP[item].
- copy())
+ self.useFixture(tools.AttributeMapMemento())
self._tenant_id = 'test-tenant'
if not plugin:
self._skip_native_pagination = None
self._skip_native_sortin = None
self.ext_api = None
- # Restore the original attribute map
- attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk
super(NeutronDbPluginV2TestCase, self).tearDown()
def setup_config(self):
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron import quota
+from neutron.tests import tools
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit import testlib_api
class ExtensionTestCase(testlib_api.WebTestCase):
- def _resotre_attr_map(self):
- attributes.RESOURCE_ATTRIBUTE_MAP = self._saved_attr_map
def _setUpExtension(self, plugin, service_type,
resource_attribute_map, extension_class,
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self._saved_attr_map = attributes.RESOURCE_ATTRIBUTE_MAP.copy()
- # Restore the global RESOURCE_ATTRIBUTE_MAP
- self.addCleanup(self._resotre_attr_map)
+ self.useFixture(tools.AttributeMapMemento())
# Create the default configurations
self.config_parse()
from neutron.db import db_base_plugin_v2
from neutron.extensions import agent
from neutron.openstack.common import uuidutils
+from neutron.tests import tools
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2
plugin = 'neutron.tests.unit.extensions.test_agent.TestAgentPlugin'
# for these tests we need to enable overlapping ips
cfg.CONF.set_default('allow_overlapping_ips', True)
- # Save the original RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
ext_mgr = AgentTestExtensionManager()
- self.addCleanup(self.restore_resource_attribute_map)
super(AgentDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
self.adminContext = context.get_admin_context()
- def restore_resource_attribute_map(self):
- # Restore the originak RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
def test_create_agent(self):
data = {'agent': {}}
_req = self.new_create_request('agents', data, self.fmt)
import webtest
from neutron.api import extensions
-from neutron.api.v2 import attributes
from neutron.api.v2 import router
from neutron import context
from neutron.extensions import providernet as pnet
from neutron import manager
from neutron.openstack.common import uuidutils
from neutron import quota
+from neutron.tests import tools
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit import testlib_api
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
# Update the plugin and extensions path
self.setup_coreplugin(plugin)
ext_mgr = ProviderExtensionManager()
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.addCleanup(self._plugin_patcher.stop)
- self.addCleanup(self._restore_attribute_map)
self.api = webtest.TestApp(router.APIRouter())
quota.QUOTAS._driver = None
cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
group='QUOTAS')
- def _restore_attribute_map(self):
- # Restore the global RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
def _prepare_net_data(self):
return {'name': 'net1',
pnet.NETWORK_TYPE: 'sometype',
import webtest
from neutron.api import extensions
-from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import constants
from neutron.common import exceptions
from neutron.db import quota_db
from neutron import quota
from neutron.tests import base
+from neutron.tests import tools
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit import testlib_api
# Ensure existing ExtensionManager is not used
extensions.PluginAwareExtensionManager._instance = None
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
# Create the default configurations
self.config_parse()
def tearDown(self):
self.api = None
self.plugin = None
- # Restore the global RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
super(QuotaExtensionTestCase, self).tearDown()
def _test_quota_default_values(self, expected_values):
from neutron.plugins.cisco.extensions import policy_profile
from neutron.plugins.cisco.n1kv import n1kv_client
from neutron.plugins.cisco.n1kv import n1kv_neutron_plugin
+from neutron.tests import tools
from neutron.tests.unit import _test_extension_portbindings as test_bindings
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
neutron_extensions.append_api_extensions_path(extensions.__path__)
- # Save the original RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.items():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
# Update the RESOURCE_ATTRIBUTE_MAP with n1kv specific extended attrs.
attributes.RESOURCE_ATTRIBUTE_MAP["networks"].update(
n1kv.EXTENDED_ATTRIBUTES_2_0["networks"])
attributes.RESOURCE_ATTRIBUTE_MAP["ports"].update(
n1kv.EXTENDED_ATTRIBUTES_2_0["ports"])
- self.addCleanup(self.restore_resource_attribute_map)
super(N1kvPluginTestCase, self).setUp(self._plugin_name,
ext_mgr=ext_mgr)
# Create some of the database entries that we require.
self._make_test_profile()
self._make_test_policy_profile()
- def restore_resource_attribute_map(self):
- # Restore the original RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
class TestN1kvNetworkProfiles(N1kvPluginTestCase):
def _prepare_net_profile_data(self,
import math
import mock
-from neutron.api.v2 import attributes
from neutron.common import constants as const
from neutron.extensions import securitygroup as ext_sg
from neutron import manager
+from neutron.tests import tools
from neutron.tests.unit.agent import test_securitygroups_rpc as test_sg_rpc
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.extensions import test_securitygroup as test_sg
notifier_cls = notifier_p.start()
self.notifier = mock.Mock()
notifier_cls.return_value = self.notifier
- self._attribute_map_bk_ = {}
- for item in attributes.RESOURCE_ATTRIBUTE_MAP:
- self._attribute_map_bk_[item] = (attributes.
- RESOURCE_ATTRIBUTE_MAP[item].
- copy())
+ self.useFixture(tools.AttributeMapMemento())
super(Ml2SecurityGroupsTestCase, self).setUp(PLUGIN_NAME)
def tearDown(self):
super(Ml2SecurityGroupsTestCase, self).tearDown()
- attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
class TestMl2SecurityGroups(Ml2SecurityGroupsTestCase,
import mock
-from neutron.api.v2 import attributes
from neutron.extensions import securitygroup as ext_sg
from neutron import manager
from neutron.plugins.oneconvergence import plugin as nvsd_plugin
+from neutron.tests import tools
from neutron.tests.unit.agent import test_securitygroups_rpc as test_sg_rpc
from neutron.tests.unit.extensions import test_securitygroup as test_sg
notifier_cls = mock.patch(AGENTNOTIFIER).start()
self.notifier = mock.Mock()
notifier_cls.return_value = self.notifier
- self._attribute_map_bk_ = {}
- for item in attributes.RESOURCE_ATTRIBUTE_MAP:
- self._attribute_map_bk_[item] = (attributes.
- RESOURCE_ATTRIBUTE_MAP[item].
- copy())
+ self.useFixture(tools.AttributeMapMemento())
with mock.patch.object(nvsd_plugin.OneConvergencePluginV2,
'oneconvergence_init',
new=mocked_oneconvergence_init):
def tearDown(self):
super(OneConvergenceSecurityGroupsTestCase, self).tearDown()
- attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk_
class TestOneConvergenceSGServerRpcCallBack(
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants as service_constants
from neutron.tests import fake_notifier
+from neutron.tests import tools
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
from neutron.tests.unit.extensions import test_agent
'TestL3NatAgentSchedulingServicePlugin')
def setUp(self):
- # Save the global RESOURCE_ATTRIBUTE_MAP before loading plugin
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
if self.l3_plugin:
service_plugins = {'l3_plugin_name': self.l3_plugin}
else:
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
- self.addCleanup(self.restore_attribute_map)
self.l3agentscheduler_dbMinxin = (
manager.NeutronManager.get_service_plugins().get(
service_constants.L3_ROUTER_NAT))
'neutron.extensions.dhcpagentscheduler.notify')
self.patched_dhcp_notify = self.dhcp_notify_p.start()
- def restore_attribute_map(self):
- # Restore the original RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
class OvsAgentSchedulerTestCase(OvsAgentSchedulerTestCaseBase):
plugin_str = 'neutron.plugins.ml2.plugin.Ml2Plugin'
def setUp(self):
- # Save the global RESOURCE_ATTRIBUTE_MAP before loading plugin
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+ self.useFixture(tools.AttributeMapMemento())
super(OvsDhcpAgentNotifierTestCase, self).setUp(self.plugin_str)
self.dhcp_notifier = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
self.dhcp_notifier_cast = mock.patch(
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
- self.addCleanup(self.restore_attribute_map)
fake_notifier.reset()
- def restore_attribute_map(self):
- # Restore the original RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
def test_network_add_to_dhcp_agent_notification(self):
with self.network() as net1:
network_id = net1['network']['id']
self.dhcp_notifier = mock.Mock(name='dhcp_notifier')
self.dhcp_notifier_cls = self.dhcp_notifier_cls_p.start()
self.dhcp_notifier_cls.return_value = self.dhcp_notifier
- # Save the global RESOURCE_ATTRIBUTE_MAP
- self.saved_attr_map = {}
- for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
- self.saved_attr_map[resource] = attrs.copy()
+
+ self.useFixture(tools.AttributeMapMemento())
+
if self.l3_plugin:
service_plugins = {'l3_plugin_name': self.l3_plugin}
else:
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
- self.addCleanup(self.restore_attribute_map)
fake_notifier.reset()
- def restore_attribute_map(self):
- # Restore the original RESOURCE_ATTRIBUTE_MAP
- attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
-
def test_router_add_to_l3_agent_notification(self):
l3_plugin = (manager.NeutronManager.get_service_plugins()
[service_constants.L3_ROUTER_NAT])