cleanup unit test usage of api extension maps
Today our unit test code uses various ways to "patch" the global RESOURCE_ATTRIBUTE_MAP as well as extension specific maps in some cases. This patch consolidates such patching whereby tests should use neutron's AttributeMapMemento in their setup() chain (only once) if they update the global map and they should individually handle backup/restore of per extension map updates. This change will simplify the code and make it easier to phase-in API definition usage with neutron-lib where we have some as API definitions and others not. Longer term the AttributeMapMemento will be replace with neutron-lib's fixture as we move all extension maps to API definitions in neutron-lib. Change-Id: I2586f0b11b107d7f57214a0d65bcf7c38a5f0ebb
This commit is contained in:
parent
32d9ae2393
commit
277996b68d
|
@ -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)
|
||||
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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()
|
||||
|
|
|
@ -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'])
|
||||
|
|
|
@ -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
|
||||
|
|
|
@ -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()
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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')
|
||||
|
|
|
@ -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,
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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):
|
||||
|
|
|
@ -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:
|
||||
|
|
|
@ -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)
|
||||
|
||||
|
|
|
@ -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()
|
||||
|
||||
|
|
|
@ -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(
|
||||
|
|
Loading…
Reference in New Issue