Merge "cleanup unit test usage of api extension maps"

This commit is contained in:
Zuul 2017-10-26 02:41:46 +00:00 committed by Gerrit Code Review
commit fe5f3dc9f2
20 changed files with 61 additions and 129 deletions

View File

@ -288,6 +288,7 @@ class BaseTestCase(DietTestCase):
super(BaseTestCase, self).setUp()
self.useFixture(lockutils.ExternalLockFixture())
self.useFixture(tools.AttributeMapMemento())
cfg.CONF.set_override('state_path', self.get_default_temp_dir().path)

View File

@ -32,7 +32,6 @@ import webtest
import neutron
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import exceptions
from neutron.plugins.common import constants
@ -968,27 +967,13 @@ class ExtensionExtendedAttributeTestCase(base.BaseTestCase):
self._api = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
self._tenant_id = "8c70909f-b081-452d-872b-df48e6c355d1"
# Save the global RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {}
for res, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.items():
self.saved_attr_map[res] = attrs.copy()
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
extattr.EXTENDED_ATTRIBUTES_2_0)
self.agentscheduler_dbMinxin = directory.get_plugin()
self.addCleanup(self.restore_attribute_map)
quota.QUOTAS._driver = None
cfg.CONF.set_override('quota_driver', 'neutron.quota.ConfDriver',
group='QUOTAS')
def restore_attribute_map(self):
# Restore the original RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
def _do_request(self, method, path, data=None, params=None, action=None):
content_type = 'application/json'
body = None

View File

@ -30,18 +30,15 @@ from neutron.api import extensions
from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
from neutron.api.rpc.handlers import dhcp_rpc
from neutron.api.rpc.handlers import l3_rpc
from neutron.api.v2 import attributes
from neutron.db import agents_db
from neutron.db import agentschedulers_db
from neutron.db.models import agent as agent_model
from neutron.extensions import agent
from neutron.extensions import dhcpagentscheduler
from neutron.extensions import l3agentscheduler
from neutron.objects import agent as ag_obj
from neutron.objects import l3agent as rb_obj
from neutron.tests.common import helpers
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
@ -231,7 +228,6 @@ class OvsAgentSchedulerTestCaseBase(test_l3.L3NatTestCaseMixin,
'TestL3NatAgentSchedulingServicePlugin')
def setUp(self):
self.useFixture(tools.AttributeMapMemento())
if self.l3_plugin:
service_plugins = {
'l3_plugin_name': self.l3_plugin,
@ -253,12 +249,6 @@ class OvsAgentSchedulerTestCaseBase(test_l3.L3NatTestCaseMixin,
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
self.adminContext = context.get_admin_context()
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
self.l3plugin = directory.get_plugin(plugin_constants.L3)
self.l3_notify_p = mock.patch(
'neutron.extensions.l3agentscheduler.notify')
@ -1293,7 +1283,6 @@ class OvsDhcpAgentNotifierTestCase(test_agent.AgentDBTestMixIn,
AgentSchedulerTestMixIn,
test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
self.useFixture(tools.AttributeMapMemento())
super(OvsDhcpAgentNotifierTestCase, self).setUp('ml2')
mock.patch.object(
self.plugin, 'filter_hosts_with_network_access',
@ -1306,12 +1295,6 @@ class OvsDhcpAgentNotifierTestCase(test_agent.AgentDBTestMixIn,
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
self.adminContext = context.get_admin_context()
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
fake_notifier.reset()
def test_network_add_to_dhcp_agent_notification(self):
@ -1463,8 +1446,6 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin,
self.dhcp_notifier_cls = self.dhcp_notifier_cls_p.start()
self.dhcp_notifier_cls.return_value = self.dhcp_notifier
self.useFixture(tools.AttributeMapMemento())
if self.l3_plugin:
service_plugins = {
'l3_plugin_name': self.l3_plugin,
@ -1478,12 +1459,6 @@ class OvsL3AgentNotifierTestCase(test_l3.L3NatTestCaseMixin,
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
self.adminContext = context.get_admin_context()
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
fake_notifier.reset()
def test_router_add_to_l3_agent_notification(self):

View File

@ -6652,6 +6652,7 @@ class DbOperationBoundMixin(object):
def setUp(self, *args, **kwargs):
super(DbOperationBoundMixin, self).setUp(*args, **kwargs)
self.useFixture(tools.AttributeMapMemento())
self._recorded_statements = []
def _event_incrementer(conn, clauseelement, *args, **kwargs):

View File

@ -23,7 +23,6 @@ from webob import exc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron import manager
from neutron import quota
from neutron.tests import tools
@ -38,7 +37,7 @@ CORE_PLUGIN = 'neutron.db.db_base_plugin_v2.NeutronDbPluginV2'
class ExtensionTestCase(testlib_api.WebTestCase):
def _setUpExtension(self, plugin, service_type,
resource_attribute_map, extension_class,
extension_class,
resource_prefix, plural_mappings=None,
translate_resource_name=False,
allow_pagination=False, allow_sorting=False,
@ -90,12 +89,6 @@ class ExtensionTestCase(testlib_api.WebTestCase):
class ExtensionTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
resource_attribute_map)
return extension_class.get_resources()
def get_actions(self):

View File

@ -24,7 +24,6 @@ from neutron_lib import constants
from neutron_lib import context
import webob.exc
from neutron.api.v2 import attributes as attr
from neutron.db import address_scope_db
from neutron.db import db_base_plugin_v2
from neutron.extensions import address_scope as ext_address_scope
@ -37,12 +36,6 @@ DB_PLUGIN_KLASS = ('neutron.tests.unit.extensions.test_address_scope.'
class AddressScopeTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attr.RESOURCE_ATTRIBUTE_MAP.update(
apidef.RESOURCE_ATTRIBUTE_MAP)
return ext_address_scope.Address_scope.get_resources()
def get_actions(self):

View File

@ -23,12 +23,10 @@ from oslo_config import cfg
from oslo_utils import uuidutils
from webob import exc
from neutron.api.v2 import attributes
from neutron.db import agents_db
from neutron.db import db_base_plugin_v2
from neutron.extensions import agent
from neutron.tests.common import helpers
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
@ -46,12 +44,6 @@ LBAAS_HOSTB = 'hostb'
class AgentTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
agent.RESOURCE_ATTRIBUTE_MAP)
return agent.Agent.get_resources()
def get_actions(self):
@ -134,7 +126,6 @@ class AgentDBTestCase(AgentDBTestMixIn,
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)
self.useFixture(tools.AttributeMapMemento())
ext_mgr = AgentTestExtensionManager()
super(AgentDBTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
self.adminContext = context.get_admin_context()

View File

@ -11,6 +11,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
from neutron_lib import context
from neutron.db import agents_db
@ -24,8 +26,7 @@ from neutron.tests.unit.db import test_db_base_plugin_v2
class AZExtensionManager(object):
def get_resources(self):
agent.RESOURCE_ATTRIBUTE_MAP['agents'].update(
az_ext.EXTENDED_ATTRIBUTES_2_0['agents'])
agent.Agent().update_attributes_map(az_ext.EXTENDED_ATTRIBUTES_2_0)
return (az_ext.Availability_zone.get_resources() +
agent.Agent.get_resources())
@ -52,11 +53,16 @@ class AZTestCommon(test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
class TestAZAgentCase(AZTestCommon):
def setUp(self):
self._agent_backup = copy.deepcopy(agent.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
plugin = ('neutron.tests.unit.extensions.'
'test_availability_zone.AZTestPlugin')
ext_mgr = AZExtensionManager()
super(TestAZAgentCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def _restore(self):
agent.RESOURCE_ATTRIBUTE_MAP = self._agent_backup
def test_list_availability_zones(self):
self._register_azs()
helpers.set_agent_admin_state(self.agent3['id'], admin_state_up=False)
@ -95,9 +101,14 @@ class TestAZAgentCase(AZTestCommon):
class TestAZNetworkCase(AZTestCommon):
def setUp(self):
self._agent_backup = copy.deepcopy(agent.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
ext_mgr = AZExtensionManager()
super(TestAZNetworkCase, self).setUp(plugin='ml2', ext_mgr=ext_mgr)
def _restore(self):
agent.RESOURCE_ATTRIBUTE_MAP = self._agent_backup
def test_availability_zones_in_create_response(self):
with self.network() as net:
self.assertIn('availability_zone_hints', net['network'])

View File

@ -13,6 +13,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
from neutron_lib import constants
from neutron_lib import context
from neutron_lib.utils import helpers
@ -34,8 +36,7 @@ _get_path = test_base._get_path
class ExtraRouteTestExtensionManager(object):
def get_resources(self):
l3.RESOURCE_ATTRIBUTE_MAP['routers'].update(
extraroute.EXTENDED_ATTRIBUTES_2_0['routers'])
l3.L3().update_attributes_map(extraroute.EXTENDED_ATTRIBUTES_2_0)
return l3.L3.get_resources()
def get_actions(self):
@ -496,6 +497,8 @@ class ExtraRouteDBIntTestCase(test_l3.L3NatDBIntTestCase,
ExtraRouteDBTestCaseBase):
def setUp(self, plugin=None, ext_mgr=None):
self._backup = copy.deepcopy(l3.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
if not plugin:
plugin = ('neutron.tests.unit.extensions.test_extraroute.'
'TestExtraRouteIntPlugin')
@ -507,10 +510,15 @@ class ExtraRouteDBIntTestCase(test_l3.L3NatDBIntTestCase,
ext_mgr=ext_mgr)
self.setup_notification_driver()
def _restore(self):
l3.RESOURCE_ATTRIBUTE_MAP = self._backup
class ExtraRouteDBSepTestCase(test_l3.L3NatDBSepTestCase,
ExtraRouteDBTestCaseBase):
def setUp(self):
self._backup = copy.deepcopy(l3.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
# the plugin without L3 support
plugin = 'neutron.tests.unit.extensions.test_l3.TestNoL3NatPlugin'
# the L3 service plugin
@ -527,3 +535,6 @@ class ExtraRouteDBSepTestCase(test_l3.L3NatDBSepTestCase,
service_plugins=service_plugins)
self.setup_notification_driver()
def _restore(self):
l3.RESOURCE_ATTRIBUTE_MAP = self._backup

View File

@ -53,8 +53,8 @@ class FlavorExtensionTestCase(extension.ExtensionTestCase):
super(FlavorExtensionTestCase, self).setUp()
self._setUpExtension(
'neutron.services.flavors.flavors_plugin.FlavorsPlugin',
constants.FLAVORS, flavors.RESOURCE_ATTRIBUTE_MAP,
flavors.Flavors, '', supported_extension_aliases='flavors')
constants.FLAVORS, flavors.Flavors, '',
supported_extension_aliases='flavors')
def test_create_flavor(self):
tenant_id = uuidutils.generate_uuid()

View File

@ -39,7 +39,6 @@ from webob import exc
from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
from neutron.api.rpc.handlers import l3_rpc
from neutron.api.v2 import attributes
from neutron.db import _resource_extend as resource_extend
from neutron.db import common_db_mixin
from neutron.db import db_base_plugin_v2
@ -78,12 +77,6 @@ DEVICE_OWNER_COMPUTE = lib_constants.DEVICE_OWNER_COMPUTE_PREFIX + 'fake'
class L3TestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
l3.RESOURCE_ATTRIBUTE_MAP)
return l3.L3.get_resources()
def get_actions(self):
@ -100,9 +93,8 @@ class L3NatExtensionTestCase(test_extensions_base.ExtensionTestCase):
super(L3NatExtensionTestCase, self).setUp()
self._setUpExtension(
'neutron.services.l3_router.l3_router_plugin.L3RouterPlugin',
plugin_constants.L3, l3.RESOURCE_ATTRIBUTE_MAP,
l3.L3, '', allow_pagination=True, allow_sorting=True,
supported_extension_aliases=['router'],
plugin_constants.L3, l3.L3, '', allow_pagination=True,
allow_sorting=True, supported_extension_aliases=['router'],
use_quota=True)
def test_router_create(self):
@ -3974,14 +3966,7 @@ class L3NatDBSepTestCase(L3BaseForSepTests, L3NatTestCaseBase,
class L3TestExtensionManagerWithDNS(L3TestExtensionManager):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
l3.RESOURCE_ATTRIBUTE_MAP)
attributes.RESOURCE_ATTRIBUTE_MAP[l3.FLOATINGIPS].update(
dns_apidef.RESOURCE_ATTRIBUTE_MAP[l3.FLOATINGIPS])
l3.L3().update_attributes_map(dns_apidef.RESOURCE_ATTRIBUTE_MAP)
return l3.L3.get_resources()
@ -4001,6 +3986,8 @@ class L3NatDBFloatingIpTestCaseWithDNS(L3BaseForSepTests, L3NatTestCaseMixin):
_extension_drivers = ['dns']
def setUp(self):
self._l3_resource_backup = copy.deepcopy(l3.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
ext_mgr = L3TestExtensionManagerWithDNS()
plugin = 'neutron.plugins.ml2.plugin.Ml2Plugin'
cfg.CONF.set_override('extension_drivers',
@ -4012,6 +3999,9 @@ class L3NatDBFloatingIpTestCaseWithDNS(L3BaseForSepTests, L3NatTestCaseMixin):
self.mock_client.reset_mock()
self.mock_admin_client.reset_mock()
def _restore(self):
l3.RESOURCE_ATTRIBUTE_MAP = self._l3_resource_backup
def _create_network(self, fmt, name, admin_state_up,
arg_list=None, set_context=False, tenant_id=None,
**kwargs):

View File

@ -14,6 +14,8 @@
# under the License.
#
import copy
import mock
import netaddr
from neutron_lib import constants
@ -365,10 +367,7 @@ class ExtGwModeIntTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
def setUp(self, plugin=None, svc_plugins=None, ext_mgr=None):
# Store l3 resource attribute map as it will be updated
self._l3_attribute_map_bk = {}
for item in l3.RESOURCE_ATTRIBUTE_MAP:
self._l3_attribute_map_bk[item] = (
l3.RESOURCE_ATTRIBUTE_MAP[item].copy())
self._l3_attribute_map_bk = copy.deepcopy(l3.RESOURCE_ATTRIBUTE_MAP)
plugin = plugin or (
'neutron.tests.unit.extensions.test_l3_ext_gw_mode.'
'TestDbIntPlugin')

View File

@ -13,6 +13,8 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import copy
from neutron_lib.api.definitions import port_security as psec
from neutron_lib.api import validators
from neutron_lib import context
@ -38,6 +40,8 @@ class PortSecurityTestCase(
test_db_base_plugin_v2.NeutronDbPluginV2TestCase):
def setUp(self, plugin=None):
self._backup = copy.deepcopy(ext_sg.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
ext_mgr = (
test_securitygroup.SecurityGroupTestExtensionManager())
super(PortSecurityTestCase, self).setUp(plugin=plugin, ext_mgr=ext_mgr)
@ -47,6 +51,9 @@ class PortSecurityTestCase(
self._skip_security_group = ('security-group' not in
plugin_obj.supported_extension_aliases)
def _restore(self):
ext_sg.RESOURCE_ATTRIBUTE_MAP = self._backup
class PortSecurityTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
securitygroups_db.SecurityGroupDbMixin,

View File

@ -11,6 +11,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
from neutron_lib.plugins import constants
from neutron.db.availability_zone import router as router_az_db
@ -61,9 +63,7 @@ class TestAZRouterCase(test_az.AZTestCommon, test_l3.L3NatTestCaseMixin):
service_plugins=service_plugins)
def _backup(self):
self.contents_backup = {}
for res, attrs in l3.RESOURCE_ATTRIBUTE_MAP.items():
self.contents_backup[res] = attrs.copy()
self.contents_backup = copy.deepcopy(l3.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
def _restore(self):

View File

@ -14,6 +14,7 @@
# limitations under the License.
import contextlib
import copy
import mock
from neutron_lib.api import validators
@ -26,7 +27,6 @@ import oslo_db.exception as exc
import testtools
import webob.exc
from neutron.api.v2 import attributes as attr
from neutron.common import exceptions as n_exc
from neutron.db import api as db_api
from neutron.db import db_base_plugin_v2
@ -55,12 +55,6 @@ class SecurityGroupTestExtensionManager(object):
ext_sg.RESOURCE_ATTRIBUTE_MAP[ext_sg.SECURITYGROUPRULES])
sg_rule_attr_desc = ext_res[ext_sg.SECURITYGROUPRULES]
existing_sg_rule_attr_map.update(sg_rule_attr_desc)
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attr.RESOURCE_ATTRIBUTE_MAP.update(
ext_sg.RESOURCE_ATTRIBUTE_MAP)
return ext_sg.Securitygroup.get_resources()
def get_actions(self):
@ -247,11 +241,16 @@ class SecurityGroupTestPlugin(db_base_plugin_v2.NeutronDbPluginV2,
class SecurityGroupDBTestCase(SecurityGroupsTestCase):
def setUp(self, plugin=None, ext_mgr=None):
self._backup = copy.deepcopy(ext_sg.RESOURCE_ATTRIBUTE_MAP)
self.addCleanup(self._restore)
plugin = plugin or DB_PLUGIN_KLASS
ext_mgr = ext_mgr or SecurityGroupTestExtensionManager()
super(SecurityGroupDBTestCase,
self).setUp(plugin=plugin, ext_mgr=ext_mgr)
def _restore(self):
ext_sg.RESOURCE_ATTRIBUTE_MAP = self._backup
class TestSecurityGroups(SecurityGroupDBTestCase):
def test_create_security_group(self):

View File

@ -31,7 +31,6 @@ from oslo_config import cfg
from oslo_utils import uuidutils
import webob.exc
from neutron.api.v2 import attributes
from neutron.common import exceptions as neutron_exc
from neutron.conf.plugins.ml2.drivers import driver_type
from neutron.db import agents_db
@ -62,12 +61,6 @@ HTTP_NOT_FOUND = 404
class SegmentTestExtensionManager(object):
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
ext_segment.RESOURCE_ATTRIBUTE_MAP)
return ext_segment.Segment.get_resources()
def get_actions(self):

View File

@ -15,7 +15,6 @@
from oslo_config import cfg
from webob import exc as web_exc
from neutron.api.v2 import attributes
from neutron.db import db_base_plugin_v2
from neutron.db import vlantransparent_db as vlt_db
from neutron.extensions import vlantransparent as vlt
@ -65,14 +64,8 @@ class VlanTransparentExtensionTestCase(test_db_base_plugin_v2.TestNetworksV2):
plugin = ('neutron.tests.unit.extensions.test_vlantransparent.'
'VlanTransparentExtensionTestPlugin')
# Save the global RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {}
for res, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.items():
self.saved_attr_map[res] = attrs.copy()
# Update the plugin and extensions path
ext_mgr = VlanTransparentExtensionManager()
self.addCleanup(self._restore_attribute_map)
super(VlanTransparentExtensionTestCase, self).setUp(plugin=plugin,
ext_mgr=ext_mgr)
@ -80,10 +73,6 @@ class VlanTransparentExtensionTestCase(test_db_base_plugin_v2.TestNetworksV2):
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 test_network_create_with_vlan_transparent_attr(self):
vlantrans = {'vlan_transparent': True}
with self.network(name='net1', **vlantrans) as net:

View File

@ -15,6 +15,7 @@ from neutron_lib import context
from oslo_utils import uuidutils
from neutron.db.quota import api as quota_db_api
from neutron.tests import tools
from neutron.tests.unit.api import test_extensions
from neutron.tests.unit.extensions import test_l3
from neutron.tests.unit.extensions import test_securitygroup
@ -161,6 +162,7 @@ class TestL3ResourcesEventHandler(BaseTestEventHandler,
def setUp(self):
super(TestL3ResourcesEventHandler, self).setUp()
self.useFixture(tools.AttributeMapMemento())
ext_mgr = test_l3.L3TestExtensionManager()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)

View File

@ -19,7 +19,6 @@ from neutron_lib.plugins import directory
from oslo_utils import uuidutils
from neutron.api.rpc.agentnotifiers import metering_rpc_agent_api
from neutron.api.v2 import attributes as attr
from neutron.common import utils
from neutron.db import api as db_api
from neutron.db.metering import metering_rpc
@ -44,9 +43,6 @@ METERING_SERVICE_PLUGIN_KLASS = (
class MeteringTestExtensionManager(object):
def get_resources(self):
attr.RESOURCE_ATTRIBUTE_MAP.update(ext_metering.RESOURCE_ATTRIBUTE_MAP)
attr.RESOURCE_ATTRIBUTE_MAP.update(ext_l3.RESOURCE_ATTRIBUTE_MAP)
l3_res = ext_l3.L3.get_resources()
metering_res = ext_metering.Metering.get_resources()

View File

@ -210,15 +210,11 @@ class NeutronPolicyTestCase(base.BaseTestCase):
attributes.RESOURCE_ATTRIBUTE_MAP.update(FAKE_RESOURCES)
self._set_rules()
def remove_fake_resource():
del attributes.RESOURCE_ATTRIBUTE_MAP["%ss" % FAKE_RESOURCE_NAME]
self.patcher = mock.patch.object(neutron.policy,
'init',
new=self.fakepolicyinit)
self.patcher.start()
policy.refresh()
self.addCleanup(remove_fake_resource)
self.addCleanup(policy.refresh)
self.context = context.Context('fake', 'fake', roles=['user'])
plugin_klass = importutils.import_class(