openstack
/
neutron
Code Issues Proposed changes

Remove auto-generation of db schema from models at startup 

This patch removes the Neutron capability of creating database tables
from sqlalchemy models for all those model classes for which
a table is not found in the database schema.
Migrations should be the official and only solution for creating and
managing the Neutron db schema.
This patch also adapts unit tests in order to ensure test schemas
are still correctly created.

DocImpact
Update deployment documentation accordingly.

Closes-Bug: #1207402

Change-Id: Ie4ee5507888ecad5f6dc32ce7a029c43014687a2
Co-Authored-By: Henry Gessau <gessau@cisco.com>
This commit is contained in:
Salvatore Orlando 2013-08-05 08:07:02 -07:00 committed by Henry Gessau
parent 08529376f1
commit 9bf0e66544
73 changed files with 118 additions and 358 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
neutron/db/api.py
View File

@ -15,14 +15,6 @@
from oslo.config import cfg
from oslo.db.sqlalchemy import session
import sqlalchemy as sql
from neutron.db import model_base
from neutron.openstack.common import log as logging
LOG = logging.getLogger(__name__)
BASE = model_base.BASEV2
_FACADE = None
@ -36,19 +28,6 @@ def _create_facade_lazily():
return _FACADE
def configure_db():
"""Configure database.
Establish the database, create an engine if needed, and register
the models.
"""
register_models()
def clear_db(base=BASE):
unregister_models(base)
def get_engine():
"""Helper method to grab engine."""
facade = _create_facade_lazily()
@ -60,25 +39,3 @@ def get_session(autocommit=True, expire_on_commit=False):
facade = _create_facade_lazily()
return facade.get_session(autocommit=autocommit,
expire_on_commit=expire_on_commit)
def register_models(base=BASE):
"""Register Models and create properties."""
try:
facade = _create_facade_lazily()
engine = facade.get_engine()
base.metadata.create_all(engine)
except sql.exc.OperationalError as e:
LOG.info(_("Database registration exception: %s"), e)
return False
return True
def unregister_models(base=BASE):
"""Unregister Models, useful clearing out data before testing."""
try:
facade = _create_facade_lazily()
engine = facade.get_engine()
base.metadata.drop_all(engine)
except Exception:
LOG.exception(_("Database exception"))

2
neutron/db/db_base_plugin_v2.py
View File

@ -27,7 +27,6 @@ from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron.common import ipv6_utils
from neutron import context as ctx
from neutron.db import api as db
from neutron.db import common_db_mixin
from neutron.db import models_v2
from neutron.db import sqlalchemyutils
@ -70,7 +69,6 @@ class NeutronDbPluginV2(neutron_plugin_base_v2.NeutronPluginBaseV2,
__native_sorting_support = True
def __init__(self):
db.configure_db()
if cfg.CONF.notify_nova_on_port_status_changes:
from neutron.notifiers import nova
# NOTE(arosen) These event listeners are here to hook into when

3
neutron/db/metering/metering_db.py
View File

@ -21,7 +21,6 @@ from sqlalchemy import sql
from neutron.api.rpc.agentnotifiers import metering_rpc_agent_api
from neutron.common import constants
from neutron.db import api as dbapi
from neutron.db import common_db_mixin as base_db
from neutron.db import l3_db
from neutron.db import model_base
@ -61,8 +60,6 @@ class MeteringDbMixin(metering.MeteringPluginBase,
base_db.CommonDbMixin):
def __init__(self):
dbapi.register_models()
self.meter_rpc = metering_rpc_agent_api.MeteringAgentNotifyAPI()
def _make_metering_label_dict(self, metering_label, fields=None):

10
neutron/db/migration/alembic_migrations/common_ext_ops.py
View File

@ -59,10 +59,7 @@ def upgrade_l3():
)
def upgrade_quota(options=None):
if not (options or {}).get('folsom_quota_db_enabled'):
return
def upgrade_quota():
op.create_table(
'quotas',
sa.Column('id', sa.String(length=36), nullable=False),
@ -78,6 +75,5 @@ def downgrade_l3():
op.drop_table(table)
def downgrade_quota(options=None):
if (options or {}).get('folsom_quota_db_enabled'):
op.drop_table('quotas')
def downgrade_quota():
op.drop_table('quotas')

8
neutron/db/migration/alembic_migrations/env.py
View File

@ -27,8 +27,6 @@ from neutron.openstack.common import importutils
MYSQL_ENGINE = None
DATABASE_QUOTA_DRIVER = 'neutron.extensions._quotav2_driver.DbQuotaDriver'
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
config = context.config
@ -118,11 +116,7 @@ def run_migrations_online():
def build_options():
return {'folsom_quota_db_enabled': is_db_quota_enabled()}
def is_db_quota_enabled():
return neutron_config.QUOTAS.quota_driver == DATABASE_QUOTA_DRIVER
return
if context.is_offline_mode():

4
neutron/db/migration/alembic_migrations/versions/folsom_initial.py
View File

@ -83,7 +83,7 @@ def upgrade(active_plugins=None, options=None):
common_ext_ops.upgrade_l3()
if migration.should_run(active_plugins, FOLSOM_QUOTA):
common_ext_ops.upgrade_quota(options)
common_ext_ops.upgrade_quota()
if PLUGINS['lbr'] in active_plugins:
upgrade_linuxbridge()
@ -484,7 +484,7 @@ def downgrade(active_plugins=None, options=None):
downgrade_linuxbridge()
if migration.should_run(active_plugins, FOLSOM_QUOTA):
common_ext_ops.downgrade_quota(options)
common_ext_ops.downgrade_quota()
if migration.should_run(active_plugins, L3_CAPABLE):
common_ext_ops.downgrade_l3()

7
neutron/db/servicetype_db.py
View File

@ -18,9 +18,7 @@
import sqlalchemy as sa
from neutron.db import api as db
from neutron.db import model_base
from neutron.db import models_v2
from neutron.openstack.common import log as logging
from neutron.services import provider_configuration as pconf
@ -47,13 +45,8 @@ class ServiceTypeManager(object):
return cls._instance
def __init__(self):
self._initialize_db()
self._load_conf()
def _initialize_db(self):
db.configure_db()
db.register_models(models_v2.model_base.BASEV2)
def _load_conf(self):
self.conf = pconf.ProviderConfiguration(
pconf.parse_service_provider_opt())

5
neutron/db/vpn/vpn_db.py
View File

@ -21,7 +21,6 @@ from sqlalchemy import orm
from sqlalchemy.orm import exc
from neutron.common import constants as n_constants
from neutron.db import api as qdbapi
from neutron.db import common_db_mixin as base_db
from neutron.db import l3_agentschedulers_db as l3_agent_db
from neutron.db import l3_db
@ -162,10 +161,6 @@ class VPNService(model_base.BASEV2, models_v2.HasId, models_v2.HasTenant):
class VPNPluginDb(vpnaas.VPNPluginBase, base_db.CommonDbMixin):
"""VPN plugin database class using SQLAlchemy models."""
def __init__(self):
"""Do the initialization for the vpn service plugin here."""
qdbapi.register_models()
def _get_validator(self):
"""Obtain validator to use for attribute validation.

3
neutron/plugins/cisco/db/network_db_v2.py
View File

@ -22,9 +22,6 @@ from neutron.openstack.common import uuidutils
from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.db import network_models_v2
# Do NOT remove this import. It is required for all the models to be seen
# by db.initialize() when called from VirtualPhysicalSwitchModelV2.__init__.
from neutron.plugins.cisco.db import nexus_models_v2 # noqa
from neutron.plugins.openvswitch import ovs_models_v2

6
neutron/plugins/cisco/models/virt_phy_sw_v2.py
View File

@ -21,7 +21,6 @@ import inspect
import sys
from neutron.api.v2 import attributes
from neutron.db import api as db_api
from neutron.extensions import portbindings
from neutron.extensions import providernet as provider
from neutron import neutron_plugin_base_v2
@ -78,11 +77,6 @@ class VirtualPhysicalSwitchModelV2(neutron_plugin_base_v2.NeutronPluginBaseV2):
self.supported_extension_aliases.extend(
self._plugins[const.VSWITCH_PLUGIN].
supported_extension_aliases)
# At this point, all the database models should have been loaded. It's
# possible that configure_db() may have been called by one of the
# plugins loaded in above. Otherwise, this call is to make sure that
# the database is initialized
db_api.configure_db()
# Initialize credential store after database initialization
cred.Store.initialize()

2
neutron/plugins/hyperv/db.py
View File

@ -28,8 +28,6 @@ LOG = logging.getLogger(__name__)
class HyperVPluginDB(object):
def initialize(self):
db_api.configure_db()
def reserve_vlan(self, session):
with session.begin(subtransactions=True):

1
neutron/plugins/hyperv/hyperv_neutron_plugin.py
View File

@ -157,7 +157,6 @@ class HyperVNeutronPlugin(agents_db.AgentDbMixin,
def __init__(self, configfile=None):
self._db = hyperv_db.HyperVPluginDB()
self._db.initialize()
self.base_binding_dict = {
portbindings.VIF_TYPE: portbindings.VIF_TYPE_HYPERV}
portbindings_base.register_port_dict_function()

4
neutron/plugins/metaplugin/meta_neutron_plugin.py
View File

@ -18,7 +18,6 @@ from oslo.config import cfg
from neutron.common import exceptions as exc
from neutron.common import topics
from neutron import context as neutron_context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import external_net_db
from neutron.db import extraroute_db
@ -85,9 +84,6 @@ class MetaPluginV2(db_base_plugin_v2.NeutronDbPluginV2,
cfg._is_opt_registered = _is_opt_registered
# Keep existing tables if multiple plugin use same table name.
db.model_base.NeutronBase.__table_args__ = {'keep_existing': True}
self.plugins = {}
plugin_list = [plugin_set.split(':')

5
neutron/plugins/nec/db/api.py
View File

@ -16,7 +16,6 @@
import sqlalchemy as sa
from neutron.db import api as db
from neutron.db import model_base
from neutron.db import models_v2
from neutron.db import securitygroups_db as sg_db
from neutron.extensions import securitygroup as ext_sg
@ -44,10 +43,6 @@ def _get_resource_model(resource):
return resource_map[resource]
def clear_db(base=model_base.BASEV2):
db.clear_db(base)
def get_ofc_item(session, resource, neutron_id):
model = _get_resource_model(resource)
if not model:

2
neutron/services/firewall/fwaas_plugin.py
View File

@ -21,7 +21,6 @@ from neutron.common import exceptions as n_exception
from neutron.common import rpc as n_rpc
from neutron.common import topics
from neutron import context as neutron_context
from neutron.db import api as qdbapi
from neutron.db.firewall import firewall_db
from neutron.extensions import firewall as fw_ext
from neutron.openstack.common import log as logging
@ -155,7 +154,6 @@ class FirewallPlugin(firewall_db.Firewall_db_mixin):
def __init__(self):
"""Do the initialization for the firewall service plugin here."""
qdbapi.register_models()
self.endpoints = [FirewallCallbacks(self)]

3
neutron/services/l3_router/l3_apic.py
View File

@ -15,11 +15,9 @@
#
# @author: Arvind Somya (asomya@cisco.com), Cisco Systems Inc.
from neutron.db import api as qdbapi
from neutron.db import db_base_plugin_v2
from neutron.db import extraroute_db
from neutron.db import l3_gwmode_db
from neutron.db import model_base
from neutron.openstack.common import excutils
from neutron.openstack.common import log as logging
from neutron.plugins.common import constants
@ -41,7 +39,6 @@ class ApicL3ServicePlugin(db_base_plugin_v2.NeutronDbPluginV2,
def __init__(self):
super(ApicL3ServicePlugin, self).__init__()
qdbapi.register_models(base=model_base.BASEV2)
self.manager = apic_manager.APICManager()
@staticmethod

3
neutron/services/l3_router/l3_router_plugin.py
View File

@ -21,14 +21,12 @@ from neutron.api.rpc.agentnotifiers import l3_rpc_agent_api
from neutron.common import constants as q_const
from neutron.common import rpc as n_rpc
from neutron.common import topics
from neutron.db import api as qdbapi
from neutron.db import common_db_mixin
from neutron.db import extraroute_db
from neutron.db import l3_dvr_db
from neutron.db import l3_dvrscheduler_db
from neutron.db import l3_gwmode_db
from neutron.db import l3_rpc_base
from neutron.db import model_base
from neutron.openstack.common import importutils
from neutron.plugins.common import constants
@ -61,7 +59,6 @@ class L3RouterPlugin(common_db_mixin.CommonDbMixin,
"extraroute", "l3_agent_scheduler"]
def __init__(self):
qdbapi.register_models(base=model_base.BASEV2)
self.setup_rpc()
self.router_scheduler = importutils.import_object(
cfg.CONF.router_scheduler_driver)

5
neutron/services/loadbalancer/drivers/embrane/db.py
View File

@ -15,15 +15,10 @@
#
# @author: Ivar Lazzaro, Embrane, Inc. ivar@embrane.com
import neutron.db.api as db
from neutron.db import models_v2 as nmodel
from neutron.services.loadbalancer.drivers.embrane import models
def initialize():
db.configure_db()
def add_pool_port(context, pool_id, port_id):
session = context.session
with session.begin(subtransactions=True):

1
neutron/services/loadbalancer/drivers/embrane/driver.py
View File

@ -56,7 +56,6 @@ def get_conf(x):
class EmbraneLbaas(abstract_driver.LoadBalancerAbstractDriver):
def __init__(self, plugin):
edb.initialize()
config_esm_mgmt = get_conf('esm_mgmt')
config_admin_username = get_conf('admin_username')
config_admin_password = get_conf('admin_password')

2
neutron/services/loadbalancer/plugin.py
View File

@ -18,7 +18,6 @@
from neutron.api.v2 import attributes as attrs
from neutron.common import exceptions as n_exc
from neutron import context
from neutron.db import api as qdbapi
from neutron.db.loadbalancer import loadbalancer_db as ldb
from neutron.db import servicetype_db as st_db
from neutron.extensions import loadbalancer
@ -52,7 +51,6 @@ class LoadBalancerPlugin(ldb.LoadBalancerPluginDb,
def __init__(self):
"""Initialization for the loadbalancer service plugin."""
qdbapi.register_models()
self.service_type_manager = st_db.ServiceTypeManager.get_instance()
self._load_drivers()

7
neutron/tests/unit/bigswitch/test_base.py
View File

@ -20,15 +20,9 @@ import mock
from oslo.config import cfg
import neutron.common.test_lib as test_lib
from neutron.db import api as db
from neutron.plugins.bigswitch import config
from neutron.tests.unit.bigswitch import fake_server
# REVISIT(kevinbenton): This needs to be imported here to create the
# portbindings table since it's not imported until function call time
# in the porttracker_db module, which will cause unit test failures when
# the unit tests are being run by testtools
from neutron.db import portbindings_db # noqa
RESTPROXY_PKG_PATH = 'neutron.plugins.bigswitch.plugin'
NOTIFIER = 'neutron.plugins.bigswitch.plugin.AgentNotifierApi'
@ -62,7 +56,6 @@ class BigSwitchTestBase(object):
self.spawn_p = mock.patch(SPAWN, new=lambda *args, **kwargs: None)
# prevent the consistency watchdog from starting
self.watch_p = mock.patch(CWATCH, new=lambda *args, **kwargs: None)
self.addCleanup(db.clear_db)
self.plugin_notifier_p.start()
self.spawn_p.start()
self.watch_p.start()

6
neutron/tests/unit/brocade/test_brocade_vlan.py
View File

@ -21,16 +21,14 @@ Test vlans alloc/dealloc.
from neutron.db import api as db
from neutron.openstack.common import context
from neutron.plugins.brocade import vlanbm as vlan_bitmap
from neutron.tests import base
from neutron.tests.unit import testlib_api
class TestVlanBitmap(base.BaseTestCase):
class TestVlanBitmap(testlib_api.SqlTestCase):
"""exercise Vlan bitmap ."""
def setUp(self):
super(TestVlanBitmap, self).setUp()
db.configure_db()
self.addCleanup(db.clear_db)
self.context = context.get_admin_context()
self.context.session = db.get_session()

24
neutron/tests/unit/cisco/n1kv/test_n1kv_db.py
View File

@ -28,8 +28,8 @@ from neutron.plugins.cisco.common import cisco_constants as c_const
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.db import n1kv_db_v2
from neutron.plugins.cisco.db import n1kv_models_v2
from neutron.tests import base
from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import testlib_api
PHYS_NET = 'physnet1'
@ -91,15 +91,13 @@ def _create_test_policy_profile_if_not_there(session,
return _profile
class VlanAllocationsTest(base.BaseTestCase):
class VlanAllocationsTest(testlib_api.SqlTestCase):
def setUp(self):
super(VlanAllocationsTest, self).setUp()
db.configure_db()
self.session = db.get_session()
self.net_p = _create_test_network_profile_if_not_there(self.session)
n1kv_db_v2.sync_vlan_allocations(self.session, self.net_p)
self.addCleanup(db.clear_db)
def test_sync_vlan_allocations_outside_segment_range(self):
self.assertRaises(c_exc.VlanIDNotFound,
@ -205,17 +203,15 @@ class VlanAllocationsTest(base.BaseTestCase):
vlan_id)
class VxlanAllocationsTest(base.BaseTestCase,
class VxlanAllocationsTest(testlib_api.SqlTestCase,
n1kv_db_v2.NetworkProfile_db_mixin):
def setUp(self):
super(VxlanAllocationsTest, self).setUp()
db.configure_db()
self.session = db.get_session()
self.net_p = _create_test_network_profile_if_not_there(
self.session, TEST_NETWORK_PROFILE_VXLAN)
n1kv_db_v2.sync_vxlan_allocations(self.session, self.net_p)
self.addCleanup(db.clear_db)
def test_sync_vxlan_allocations_outside_segment_range(self):
self.assertRaises(c_exc.VxlanIDNotFound,
@ -296,9 +292,7 @@ class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
super(NetworkBindingsTest, self).setUp()
db.configure_db()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_add_network_binding(self):
with self.network() as network:
@ -547,14 +541,12 @@ class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
self.assertEqual(t_members, [])
class NetworkProfileTests(base.BaseTestCase,
class NetworkProfileTests(testlib_api.SqlTestCase,
n1kv_db_v2.NetworkProfile_db_mixin):
def setUp(self):
super(NetworkProfileTests, self).setUp()
db.configure_db()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_create_network_profile(self):
_db_profile = n1kv_db_v2.create_network_profile(self.session,
@ -731,13 +723,11 @@ class NetworkProfileTests(base.BaseTestCase,
self.assertEqual(len(test_profiles), len(list(profiles)))
class PolicyProfileTests(base.BaseTestCase):
class PolicyProfileTests(testlib_api.SqlTestCase):
def setUp(self):
super(PolicyProfileTests, self).setUp()
db.configure_db()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_create_policy_profile(self):
_db_profile = n1kv_db_v2.create_policy_profile(TEST_POLICY_PROFILE)
@ -775,15 +765,13 @@ class PolicyProfileTests(base.BaseTestCase):
self.assertEqual(profile.name, got_profile.name)
class ProfileBindingTests(base.BaseTestCase,
class ProfileBindingTests(testlib_api.SqlTestCase,
n1kv_db_v2.NetworkProfile_db_mixin,
common_db_mixin.CommonDbMixin):
def setUp(self):
super(ProfileBindingTests, self).setUp()
db.configure_db()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def _create_test_binding_if_not_there(self, tenant_id, profile_id,
profile_type):

1
neutron/tests/unit/cisco/n1kv/test_n1kv_plugin.py
View File

@ -213,7 +213,6 @@ class N1kvPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
attributes.RESOURCE_ATTRIBUTE_MAP["ports"].update(
n1kv.EXTENDED_ATTRIBUTES_2_0["ports"])
self.addCleanup(self.restore_resource_attribute_map)
self.addCleanup(db.clear_db)
super(N1kvPluginTestCase, self).setUp(self._plugin_name,
ext_mgr=ext_mgr)
# Create some of the database entries that we require.

15
neutron/tests/unit/cisco/test_network_db.py
View File

@ -17,23 +17,21 @@ import collections
import mock
import testtools
from neutron.db import api as db
from neutron.plugins.cisco.common import cisco_constants
from neutron.plugins.cisco.common import cisco_credentials_v2
from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.common import config as config
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco import network_plugin
from neutron.tests import base
from neutron.tests.unit import testlib_api
class CiscoNetworkDbTest(base.BaseTestCase):
class CiscoNetworkDbTest(testlib_api.SqlTestCase):
"""Base class for Cisco network database unit tests."""
def setUp(self):
super(CiscoNetworkDbTest, self).setUp()
db.configure_db()
# The Cisco network plugin includes a thin layer of QoS and
# credential API methods which indirectly call Cisco QoS and
@ -47,8 +45,6 @@ class CiscoNetworkDbTest(base.BaseTestCase):
'__init__', new=new_network_plugin_init):
self._network_plugin = network_plugin.PluginV2()
self.addCleanup(db.clear_db)
class CiscoNetworkQosDbTest(CiscoNetworkDbTest):
@ -290,15 +286,10 @@ class CiscoNetworkCredentialDbTest(CiscoNetworkDbTest):
cdb.get_credential, cred_n1kv_2_id)
class CiscoCredentialStoreTest(base.BaseTestCase):
class CiscoCredentialStoreTest(testlib_api.SqlTestCase):
"""Cisco Credential Store unit tests."""
def setUp(self):
super(CiscoCredentialStoreTest, self).setUp()
db.configure_db()
self.addCleanup(db.clear_db)
def test_cred_store_init_duplicate_creds_ignored(self):
"""Check that with multi store instances, dup creds are ignored."""
# Create a device dictionary containing credentials for 1 switch.

6
neutron/tests/unit/cisco/test_nexus_db.py
View File

@ -22,10 +22,10 @@ from neutron.plugins.cisco.common import cisco_exceptions as c_exc
from neutron.plugins.cisco.common import config
from neutron.plugins.cisco.db import nexus_db_v2 as nxdb
from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
from neutron.tests import base
from neutron.tests.unit import testlib_api
class CiscoNexusDbTest(base.BaseTestCase):
class CiscoNexusDbTest(testlib_api.SqlTestCase):
"""Unit tests for cisco.db.nexus_models_v2.NexusPortBinding model."""
@ -33,9 +33,7 @@ class CiscoNexusDbTest(base.BaseTestCase):
def setUp(self):
super(CiscoNexusDbTest, self).setUp()
db.configure_db()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def _npb_test_obj(self, pnum, vnum, switch=None, instance=None):
"""Create a Nexus port binding test object from a pair of numbers."""

8
neutron/tests/unit/cisco/test_nexus_plugin.py
View File

@ -18,7 +18,6 @@ import mock
from oslo.config import cfg
from neutron.db import api as db
from neutron.extensions import providernet as provider
from neutron.openstack.common import importutils
from neutron.plugins.cisco.common import cisco_constants as const
@ -26,8 +25,7 @@ from neutron.plugins.cisco.common import cisco_exceptions as cisco_exc
from neutron.plugins.cisco.common import config as cisco_config
from neutron.plugins.cisco.db import network_db_v2 as cdb
from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
from neutron.tests import base
from neutron.tests.unit import testlib_api
NEXUS_IP_ADDRESS = '1.1.1.1'
HOSTNAME1 = 'testhost1'
@ -50,7 +48,7 @@ NET_ATTRS = [const.NET_ID,
const.NET_VLAN_ID]
class TestCiscoNexusPlugin(base.BaseTestCase):
class TestCiscoNexusPlugin(testlib_api.SqlTestCase):
def setUp(self):
"""Set up function."""
@ -144,9 +142,7 @@ class TestCiscoNexusPlugin(base.BaseTestCase):
'password': 'password'
},
}
db.configure_db()
self.addCleanup(db.clear_db)
# Use a mock netconf client
self.mock_ncclient = mock.Mock()

4
neutron/tests/unit/cisco/test_plugin_model.py
View File

@ -22,10 +22,10 @@ from neutron.plugins.cisco.common import cisco_constants as const
from neutron.plugins.cisco.common import config as cisco_config
from neutron.plugins.cisco.models import virt_phy_sw_v2
from neutron.plugins.cisco.nexus import cisco_nexus_plugin_v2
from neutron.tests import base
from neutron.tests.unit import testlib_api
class TestCiscoPluginModel(base.BaseTestCase):
class TestCiscoPluginModel(testlib_api.SqlTestCase):
def setUp(self):
# Point config file to: neutron/tests/etc/neutron.conf.test

1
neutron/tests/unit/db/loadbalancer/test_db_loadbalancer.py
View File

@ -24,7 +24,6 @@ from neutron.api import extensions
from neutron.common import config
from neutron.common import exceptions as n_exc
from neutron import context
import neutron.db.l3_db # noqa
from neutron.db.loadbalancer import loadbalancer_db as ldb
from neutron.db import servicetype_db as sdb
import neutron.extensions

6
neutron/tests/unit/db/test_agent_db.py
View File

@ -18,22 +18,20 @@ from oslo.db import exception as exc
from neutron import context
from neutron.db import agents_db
from neutron.db import api as db
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.tests import base
from neutron.tests.unit import testlib_api
class FakePlugin(base_plugin.NeutronDbPluginV2, agents_db.AgentDbMixin):
"""A fake plugin class containing all DB methods."""
class TestAgentsDbMixin(base.BaseTestCase):
class TestAgentsDbMixin(testlib_api.SqlTestCase):
def setUp(self):
super(TestAgentsDbMixin, self).setUp()
self.context = context.get_admin_context()
self.plugin = FakePlugin()
self.addCleanup(db.clear_db)
self.agent_status = {
'agent_type': 'Open vSwitch agent',

7
neutron/tests/unit/db/test_dvr_mac_db.py
View File

@ -17,10 +17,9 @@ import mock
from oslo.config import cfg
from neutron import context
from neutron.db import api as db
from neutron.db import dvr_mac_db
from neutron.extensions import dvr
from neutron.tests import base
from neutron.tests.unit import testlib_api
class DVRDbMixinImpl(dvr_mac_db.DVRDbMixin):
@ -29,13 +28,11 @@ class DVRDbMixinImpl(dvr_mac_db.DVRDbMixin):
self.notifier = notifier
class DvrDbMixinTestCase(base.BaseTestCase):
class DvrDbMixinTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(DvrDbMixinTestCase, self).setUp()
db.configure_db()
self.ctx = context.get_admin_context()
self.addCleanup(db.clear_db)
self.mixin = DVRDbMixinImpl(mock.Mock())
def _create_dvr_mac_entry(self, host, mac_address):

7
neutron/tests/unit/db/test_l3_dvr_db.py
View File

@ -18,19 +18,16 @@ import mock
from neutron.common import constants as l3_const
from neutron import context
from neutron.db import api as db
from neutron.db import l3_dvr_db
from neutron import manager
from neutron.tests import base
from neutron.tests.unit import testlib_api
class L3DvrTestCase(base.BaseTestCase):
class L3DvrTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(L3DvrTestCase, self).setUp()
db.configure_db()
self.ctx = context.get_admin_context()
self.addCleanup(db.clear_db)
self.mixin = l3_dvr_db.L3_NAT_with_dvr_db_mixin()
def _create_router(self, router):

6
neutron/tests/unit/db/test_quota_db.py
View File

@ -17,10 +17,9 @@
from neutron.common import exceptions
from neutron import context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2 as base_plugin
from neutron.db import quota_db
from neutron.tests import base
from neutron.tests.unit import testlib_api
class FakePlugin(base_plugin.NeutronDbPluginV2, quota_db.DbQuotaDriver):
@ -42,12 +41,11 @@ PROJECT = 'prj_test'
RESOURCE = 'res_test'
class TestDbQuotaDriver(base.BaseTestCase):
class TestDbQuotaDriver(testlib_api.SqlTestCase):
def setUp(self):
super(TestDbQuotaDriver, self).setUp()
self.plugin = FakePlugin()
self.context = context.get_admin_context()
self.addCleanup(db.clear_db)
def test_create_quota_limit(self):
defaults = {RESOURCE: TestResource(RESOURCE, 4)}

2
neutron/tests/unit/embrane/test_embrane_l3_plugin.py
View File

@ -17,7 +17,6 @@
from oslo.config import cfg
from neutron.db import api as db
from neutron.plugins.embrane.common import config # noqa
from neutron.tests.unit import test_extension_extraroute as extraroute_test
from neutron.tests.unit import test_l3_plugin as router_test
@ -31,7 +30,6 @@ class TestEmbraneL3NatDBTestCase(router_test.L3NatDBIntTestCase):
def setUp(self):
cfg.CONF.set_override('admin_password', "admin123", 'heleos')
self.addCleanup(db.clear_db)
super(TestEmbraneL3NatDBTestCase, self).setUp()

2
neutron/tests/unit/embrane/test_embrane_neutron_plugin.py
View File

@ -19,7 +19,6 @@ import sys
import mock
from oslo.config import cfg
from neutron.db import api as db
from neutron.plugins.embrane.common import config # noqa
from neutron.tests.unit import test_db_plugin as test_plugin
@ -34,7 +33,6 @@ class EmbranePluginV2TestCase(test_plugin.NeutronDbPluginV2TestCase):
cfg.CONF.set_override('admin_password', "admin123", 'heleos')
p = mock.patch.dict(sys.modules, {'heleosapi': mock.Mock()})
p.start()
self.addCleanup(db.clear_db)
# dict patches must be explicitly stopped
self.addCleanup(p.stop)
super(EmbranePluginV2TestCase, self).setUp(self._plugin_name)

7
neutron/tests/unit/linuxbridge/test_lb_db.py
View File

@ -21,8 +21,8 @@ from testtools import matchers
from neutron.common import exceptions as n_exc
from neutron.db import api as db
from neutron.plugins.linuxbridge.db import l2network_db_v2 as lb_db
from neutron.tests import base
from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import testlib_api
PHYS_NET = 'physnet1'
PHYS_NET_2 = 'physnet2'
@ -36,13 +36,11 @@ PLUGIN_NAME = ('neutron.plugins.linuxbridge.'
'lb_neutron_plugin.LinuxBridgePluginV2')
class NetworkStatesTest(base.BaseTestCase):
class NetworkStatesTest(testlib_api.SqlTestCase):
def setUp(self):
super(NetworkStatesTest, self).setUp()
db.configure_db()
lb_db.sync_network_states(VLAN_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_sync_network_states(self):
self.assertIsNone(lb_db.get_network_state(PHYS_NET,
@ -155,7 +153,6 @@ class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
cfg.CONF.set_override('network_vlan_ranges', ['physnet1:1000:2999'],
group='VLANS')
super(NetworkBindingsTest, self).setUp(plugin=PLUGIN_NAME)
db.configure_db()
self.session = db.get_session()
def test_add_network_binding(self):

17
neutron/tests/unit/metaplugin/test_metaplugin.py
View File

@ -20,13 +20,12 @@ import testtools
from neutron.common import exceptions as exc
from neutron.common import topics
from neutron import context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import models_v2
from neutron.extensions import flavor as ext_flavor
from neutron.openstack.common import uuidutils
from neutron.plugins.metaplugin import meta_neutron_plugin
from neutron.tests import base
from neutron.tests.unit import testlib_api
CONF_FILE = ""
META_PATH = "neutron.plugins.metaplugin"
@ -57,8 +56,6 @@ def setup_metaplugin_conf(has_l3=True):
cfg.CONF.set_override('base_mac', "12:34:56:78:90:ab")
#TODO(nati) remove this after subnet quota change is merged
cfg.CONF.set_override('max_dns_nameservers', 10)
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
# Hooks registered by metaplugin must not exist for other plugins UT.
@ -70,20 +67,16 @@ def unregister_meta_hooks():
models_v2.Port, 'metaplugin_port', None, None, None)
class MetaNeutronPluginV2Test(base.BaseTestCase):
class MetaNeutronPluginV2Test(testlib_api.SqlTestCase):
"""Class conisting of MetaNeutronPluginV2 unit tests."""
has_l3 = True
def setUp(self):
super(MetaNeutronPluginV2Test, self).setUp()
db._ENGINE = None
db._MAKER = None
self.fake_tenant_id = uuidutils.generate_uuid()
self.context = context.get_admin_context()
db.configure_db()
self.addCleanup(db.clear_db)
self.addCleanup(unregister_meta_hooks)
setup_metaplugin_conf(self.has_l3)
@ -365,15 +358,11 @@ class MetaNeutronPluginV2TestWithoutL3(MetaNeutronPluginV2Test):
self.skipTest("Test case without router")
class MetaNeutronPluginV2TestRpcFlavor(base.BaseTestCase):
class MetaNeutronPluginV2TestRpcFlavor(testlib_api.SqlTestCase):
"""Tests for rpc_flavor."""
def setUp(self):
super(MetaNeutronPluginV2TestRpcFlavor, self).setUp()
db._ENGINE = None
db._MAKER = None
db.configure_db()
self.addCleanup(db.clear_db)
self.addCleanup(unregister_meta_hooks)
def test_rpc_flavor(self):

7
neutron/tests/unit/ml2/db/test_ml2_dvr_db.py
View File

@ -18,22 +18,19 @@ import mock
from sqlalchemy.orm import query
from neutron import context
from neutron.db import api as db_api
from neutron.db import l3_db
from neutron.db import models_v2
from neutron.extensions import portbindings
from neutron.plugins.ml2 import db as ml2_db
from neutron.plugins.ml2 import models as ml2_models
from neutron.tests import base
from neutron.tests.unit import testlib_api
class Ml2DBTestCase(base.BaseTestCase):
class Ml2DBTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(Ml2DBTestCase, self).setUp()
db_api.configure_db()
self.ctx = context.get_admin_context()
self.addCleanup(db_api.clear_db)
def _setup_neutron_network(self, network_id, port_ids):
with self.ctx.session.begin(subtransactions=True):

12
neutron/tests/unit/ml2/drivers/arista/test_arista_mechanism_driver.py
View File

@ -17,12 +17,12 @@ import mock
from oslo.config import cfg
from neutron.common import constants as n_const
import neutron.db.api as ndb
from neutron.extensions import portbindings
from neutron.plugins.ml2.drivers.arista import db
from neutron.plugins.ml2.drivers.arista import exceptions as arista_exc
from neutron.plugins.ml2.drivers.arista import mechanism_arista as arista
from neutron.tests import base
from neutron.tests.unit import testlib_api
def setup_arista_wrapper_config(value=''):
@ -36,18 +36,13 @@ def setup_valid_config():
setup_arista_wrapper_config('value')
class AristaProvisionedVlansStorageTestCase(base.BaseTestCase):
class AristaProvisionedVlansStorageTestCase(testlib_api.SqlTestCase):
"""Test storing and retriving functionality of Arista mechanism driver.
Tests all methods of this class by invoking them separately as well
as a group.
"""
def setUp(self):
super(AristaProvisionedVlansStorageTestCase, self).setUp()
ndb.configure_db()
self.addCleanup(ndb.clear_db)
def test_tenant_is_remembered(self):
tenant_id = 'test'
@ -539,7 +534,7 @@ class NegativeRPCWrapperTestCase(base.BaseTestCase):
self.assertRaises(arista_exc.AristaRpcError, drv.get_tenants)
class RealNetStorageAristaDriverTestCase(base.BaseTestCase):
class RealNetStorageAristaDriverTestCase(testlib_api.SqlTestCase):
"""Main test cases for Arista Mechanism driver.
Tests all mechanism driver APIs supported by Arista Driver. It invokes
@ -549,7 +544,6 @@ class RealNetStorageAristaDriverTestCase(base.BaseTestCase):
def setUp(self):
super(RealNetStorageAristaDriverTestCase, self).setUp()
self.fake_rpc = mock.MagicMock()
ndb.configure_db()
self.drv = arista.AristaDriver(self.fake_rpc)
def tearDown(self):

9
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_nexus.py
View File

@ -18,7 +18,6 @@ import mock
import testtools
from neutron.common import constants as n_const
from neutron.db import api as db
from neutron.extensions import portbindings
from neutron.openstack.common import importutils
from neutron.plugins.ml2 import driver_api as api
@ -27,7 +26,7 @@ from neutron.plugins.ml2.drivers.cisco.nexus import exceptions
from neutron.plugins.ml2.drivers.cisco.nexus import mech_cisco_nexus
from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2
from neutron.plugins.ml2.drivers.cisco.nexus import nexus_network_driver
from neutron.tests import base
from neutron.tests.unit import testlib_api
NEXUS_IP_ADDRESS = '1.1.1.1'
@ -93,7 +92,7 @@ class FakePortContext(object):
return self._segment
class TestCiscoNexusDevice(base.BaseTestCase):
class TestCiscoNexusDevice(testlib_api.SqlTestCase):
"""Unit tests for Cisco ML2 Nexus device driver."""
@ -151,15 +150,11 @@ class TestCiscoNexusDevice(base.BaseTestCase):
mech_instance.driver.nexus_switches = (
mech_instance._nexus_switches)
db.configure_db()
mock.patch.object(mech_cisco_nexus.CiscoNexusMechanismDriver,
'__init__', new=new_nexus_init).start()
self._cisco_mech_driver = (mech_cisco_nexus.
CiscoNexusMechanismDriver())
self.addCleanup(db.clear_db)
def _create_delete_port(self, port_config):
"""Tests creation and deletion of a virtual port."""
nexus_ip_addr = port_config.nexus_ip_addr

10
neutron/tests/unit/ml2/drivers/cisco/nexus/test_cisco_nexus_db.py
View File

@ -16,23 +16,17 @@
import collections
import testtools
from neutron.db import api as db
from neutron.plugins.ml2.drivers.cisco.nexus import exceptions
from neutron.plugins.ml2.drivers.cisco.nexus import nexus_db_v2
from neutron.tests import base
from neutron.tests.unit import testlib_api
class CiscoNexusDbTest(base.BaseTestCase):
class CiscoNexusDbTest(testlib_api.SqlTestCase):
"""Unit tests for Cisco mechanism driver's Nexus port binding database."""
NpbObj = collections.namedtuple('NpbObj', 'port vlan switch instance')
def setUp(self):
super(CiscoNexusDbTest, self).setUp()
db.configure_db()
self.addCleanup(db.clear_db)
def _npb_test_obj(self, pnum, vnum, switch='10.9.8.7', instance=None):
"""Creates a Nexus port binding test object from a pair of numbers."""
if pnum is 'router':

3
neutron/tests/unit/ml2/drivers/test_l2population.py
View File

@ -23,7 +23,6 @@ from neutron.common import constants
from neutron.common import topics
from neutron import context
from neutron.db import agents_db
from neutron.db import api as db_api
from neutron.extensions import portbindings
from neutron.extensions import providernet as pnet
from neutron import manager
@ -131,8 +130,6 @@ class TestL2PopulationRpcTestCase(test_plugin.NeutronDbPluginV2TestCase):
uptime_patch = mock.patch(uptime, return_value=190)
uptime_patch.start()
self.addCleanup(db_api.clear_db)
def tearDown(self):
l2_consts.SUPPORTED_AGENT_TYPES = self.orig_supported_agents
super(TestL2PopulationRpcTestCase, self).tearDown()

5
neutron/tests/unit/ml2/test_helpers.py
View File

@ -23,6 +23,7 @@ import neutron.db.api as db
from neutron.plugins.ml2.drivers import helpers
from neutron.plugins.ml2.drivers import type_vlan
from neutron.tests import base
from neutron.tests.unit import testlib_api
TENANT_NET = 'phys_net2'
@ -34,16 +35,14 @@ NETWORK_VLAN_RANGES = {
}
class HelpersTest(base.BaseTestCase):
class HelpersTest(testlib_api.SqlTestCase):
def setUp(self):
super(HelpersTest, self).setUp()
db.configure_db()
self.driver = type_vlan.VlanTypeDriver()
self.driver.network_vlan_ranges = NETWORK_VLAN_RANGES
self.driver._sync_vlan_allocations()
self.session = db.get_session()
self.addCleanup(db.clear_db)
self.useFixture(
fixtures.FakeLogger(
name=helpers.__name__,

3
neutron/tests/unit/ml2/test_mechanism_odl.py
View File

@ -24,6 +24,7 @@ from neutron.plugins.ml2.drivers import mechanism_odl
from neutron.plugins.ml2 import plugin
from neutron.tests import base
from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import testlib_api
PLUGIN_NAME = 'neutron.plugins.ml2.plugin.Ml2Plugin'
@ -70,7 +71,7 @@ class OpenDaylightTestCase(test_plugin.NeutronDbPluginV2TestCase):
self.assertFalse(self.mech.check_segment(self.segment))
class OpenDayLightMechanismConfigTests(base.BaseTestCase):
class OpenDayLightMechanismConfigTests(testlib_api.SqlTestCase):
def _set_config(self, url='http://127.0.0.1:9999', username='someuser',
password='somepass'):

6
neutron/tests/unit/ml2/test_type_flat.py
View File

@ -18,23 +18,21 @@ import neutron.db.api as db
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_flat
from neutron.tests import base
from neutron.tests.unit import testlib_api
from oslo.config import cfg
FLAT_NETWORKS = ['flat_net1', 'flat_net2']
class FlatTypeTest(base.BaseTestCase):
class FlatTypeTest(testlib_api.SqlTestCase):
def setUp(self):
super(FlatTypeTest, self).setUp()
db.configure_db()
cfg.CONF.set_override('flat_networks', FLAT_NETWORKS,
group='ml2_type_flat')
self.driver = type_flat.FlatTypeDriver()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def _get_allocation(self, session, segment):
return session.query(type_flat.FlatAllocation).filter_by(

10
neutron/tests/unit/ml2/test_type_gre.py
View File

@ -22,7 +22,7 @@ from neutron.common import exceptions as exc
import neutron.db.api as db
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_gre
from neutron.tests import base
from neutron.tests.unit import testlib_api
TUNNEL_IP_ONE = "10.10.10.10"
TUNNEL_IP_TWO = "10.10.10.20"
@ -32,16 +32,14 @@ TUNNEL_RANGES = [(TUN_MIN, TUN_MAX)]
UPDATED_TUNNEL_RANGES = [(TUN_MIN + 5, TUN_MAX + 5)]
class GreTypeTest(base.BaseTestCase):
class GreTypeTest(testlib_api.SqlTestCase):
def setUp(self):
super(GreTypeTest, self).setUp()
db.configure_db()
self.driver = type_gre.GreTypeDriver()
self.driver.gre_id_ranges = TUNNEL_RANGES
self.driver._sync_gre_allocations()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_validate_provider_segment(self):
segment = {api.NETWORK_TYPE: 'gre',
@ -233,7 +231,7 @@ class GreTypeTest(base.BaseTestCase):
log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
class GreTypeMultiRangeTest(base.BaseTestCase):
class GreTypeMultiRangeTest(testlib_api.SqlTestCase):
TUN_MIN0 = 100
TUN_MAX0 = 101
@ -243,12 +241,10 @@ class GreTypeMultiRangeTest(base.BaseTestCase):
def setUp(self):
super(GreTypeMultiRangeTest, self).setUp()
db.configure_db()
self.driver = type_gre.GreTypeDriver()
self.driver.gre_id_ranges = self.TUNNEL_MULTI_RANGES
self.driver._sync_gre_allocations()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_release_segment(self):
segments = [self.driver.allocate_tenant_segment(self.session)

6
neutron/tests/unit/ml2/test_type_vlan.py
View File

@ -21,7 +21,7 @@ import neutron.db.api as db
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_vlan
from neutron.tests import base
from neutron.tests.unit import testlib_api
PROVIDER_NET = 'phys_net1'
TENANT_NET = 'phys_net2'
@ -37,16 +37,14 @@ UPDATED_VLAN_RANGES = {
}
class VlanTypeTest(base.BaseTestCase):
class VlanTypeTest(testlib_api.SqlTestCase):
def setUp(self):
super(VlanTypeTest, self).setUp()
db.configure_db()
self.driver = type_vlan.VlanTypeDriver()
self.driver.network_vlan_ranges = NETWORK_VLAN_RANGES
self.driver._sync_vlan_allocations()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def _get_allocation(self, session, segment):
return session.query(type_vlan.VlanAllocation).filter_by(

10
neutron/tests/unit/ml2/test_type_vxlan.py
View File

@ -25,7 +25,7 @@ from neutron.db import api as db
from neutron.plugins.common import constants as p_const
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_vxlan
from neutron.tests import base
from neutron.tests.unit import testlib_api
TUNNEL_IP_ONE = "10.10.10.10"
@ -40,10 +40,9 @@ VXLAN_UDP_PORT_ONE = 9999
VXLAN_UDP_PORT_TWO = 8888
class VxlanTypeTest(base.BaseTestCase):
class VxlanTypeTest(testlib_api.SqlTestCase):
def setUp(self):
super(VxlanTypeTest, self).setUp()
db.configure_db()
cfg.CONF.set_override('vni_ranges', [TUNNEL_RANGES],
group='ml2_type_vxlan')
cfg.CONF.set_override('vxlan_group', MULTICAST_GROUP,
@ -52,7 +51,6 @@ class VxlanTypeTest(base.BaseTestCase):
self.driver.vxlan_vni_ranges = TUNNEL_RANGES
self.driver._sync_vxlan_allocations()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_vxlan_tunnel_type(self):
self.assertEqual(self.driver.get_type(), p_const.TYPE_VXLAN)
@ -254,7 +252,7 @@ class VxlanTypeTest(base.BaseTestCase):
log_warn.assert_called_once_with(mock.ANY, TUNNEL_IP_ONE)
class VxlanTypeMultiRangeTest(base.BaseTestCase):
class VxlanTypeMultiRangeTest(testlib_api.SqlTestCase):
TUN_MIN0 = 100
TUN_MAX0 = 101
@ -264,12 +262,10 @@ class VxlanTypeMultiRangeTest(base.BaseTestCase):
def setUp(self):
super(VxlanTypeMultiRangeTest, self).setUp()
db.configure_db()
self.driver = type_vxlan.VxlanTypeDriver()
self.driver.vxlan_vni_ranges = self.TUNNEL_MULTI_RANGES
self.driver._sync_vxlan_allocations()
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_release_segment(self):
segments = [self.driver.allocate_tenant_segment(self.session)

7
neutron/tests/unit/mlnx/test_mlnx_db.py
View File

@ -19,8 +19,8 @@ from testtools import matchers
from neutron.common import exceptions as n_exc
from neutron.db import api as db
from neutron.plugins.mlnx.db import mlnx_db_v2 as mlnx_db
from neutron.tests import base
from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import testlib_api
PHYS_NET = 'physnet1'
PHYS_NET_2 = 'physnet2'
@ -33,13 +33,11 @@ UPDATED_VLAN_RANGES = {PHYS_NET: [(VLAN_MIN + 5, VLAN_MAX + 5)],
TEST_NETWORK_ID = 'abcdefghijklmnopqrstuvwxyz'
class SegmentationIdAllocationTest(base.BaseTestCase):
class SegmentationIdAllocationTest(testlib_api.SqlTestCase):
def setUp(self):
super(SegmentationIdAllocationTest, self).setUp()
db.configure_db()
mlnx_db.sync_network_states(VLAN_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_sync_segmentationIdAllocation(self):
self.assertIsNone(mlnx_db.get_network_state(PHYS_NET,
@ -159,7 +157,6 @@ class SegmentationIdAllocationTest(base.BaseTestCase):
class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
super(NetworkBindingsTest, self).setUp()
db.configure_db()
self.session = db.get_session()
def test_add_network_binding(self):

4
neutron/tests/unit/mlnx/test_mlnx_plugin_config.py
View File

@ -20,10 +20,10 @@ from oslo.config import cfg
from neutron.plugins.mlnx.common import config # noqa
from neutron.plugins.mlnx.common import constants
from neutron.plugins.mlnx import mlnx_plugin
from neutron.tests import base
from neutron.tests.unit import testlib_api
class TestMlnxPluginConfig(base.BaseTestCase):
class TestMlnxPluginConfig(testlib_api.SqlTestCase):
expected_vlan_mappings = {'physnet1': [(1, 1000)],
'physnet2': [(1, 1000)]}
expected_network_types = {'physnet1': constants.TYPE_ETH,

1
neutron/tests/unit/nec/test_db.py
View File

@ -20,7 +20,6 @@ from neutron.common import constants as q_const
from neutron.openstack.common import uuidutils
from neutron.plugins.nec.common import exceptions as nexc
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec.db import models as nmodels # noqa
from neutron.tests.unit.nec import test_nec_plugin

8
neutron/tests/unit/nec/test_ofc_manager.py
View File

@ -16,13 +16,11 @@
import mock
from neutron import context
from neutron.db import api as db
from neutron.openstack.common import uuidutils
from neutron.plugins.nec.common import config
from neutron.plugins.nec.db import api as ndb
from neutron.plugins.nec.db import models as nmodels # noqa
from neutron.plugins.nec import ofc_manager
from neutron.tests import base
from neutron.tests.unit import testlib_api
class FakePortInfo(object):
@ -38,15 +36,13 @@ class FakePortInfo(object):
raise AttributeError(name)
class OFCManagerTestBase(base.BaseTestCase):
class OFCManagerTestBase(testlib_api.SqlTestCase):
"""Class conisting of OFCManager unit tests."""
def setUp(self):
super(OFCManagerTestBase, self).setUp()
db.configure_db()
driver = "neutron.tests.unit.nec.stub_ofc_driver.StubOFCDriver"
config.CONF.set_override('driver', driver, 'OFC')
self.addCleanup(ndb.clear_db)
self.plugin = mock.Mock()
self.plugin.get_packet_filters_for_port.return_value = None
self.ofc = ofc_manager.OFCManager(self.plugin)

1
neutron/tests/unit/nec/test_security_group.py
View File

@ -20,7 +20,6 @@ import mock
from neutron.api.v2 import attributes
from neutron.extensions import securitygroup as ext_sg
from neutron import manager
from neutron.plugins.nec.db import api as ndb # noqa
from neutron.tests.unit.nec import test_nec_plugin
from neutron.tests.unit import test_extension_security_group as test_sg
from neutron.tests.unit import test_security_groups_rpc as test_sg_rpc

3
neutron/tests/unit/opencontrail/test_contrail_plugin.py
View File

@ -28,11 +28,9 @@ from neutron.api.v2 import attributes as attr
from neutron.api.v2 import base as api_base
from neutron.common import exceptions as exc
from neutron import context as neutron_context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import external_net_db
from neutron.db import l3_db
from neutron.db import quota_db # noqa
from neutron.db import securitygroups_db
from neutron.extensions import portbindings
from neutron.extensions import securitygroup as ext_sg
@ -219,7 +217,6 @@ class ContrailPluginTestCase(test_plugin.NeutronDbPluginV2TestCase):
cfg.CONF.keystone_authtoken = KeyStoneInfo()
mock.patch('requests.post').start().side_effect = FAKE_SERVER.request
db.configure_db()
super(ContrailPluginTestCase, self).setUp(self._plugin_name)

11
neutron/tests/unit/openvswitch/test_ovs_db.py
View File

@ -25,8 +25,8 @@ from neutron.common import exceptions as n_exc
from neutron.db import api as db
from neutron.plugins.openvswitch import ovs_db_v2
from neutron.plugins.openvswitch import ovs_models_v2 as ovs_models
from neutron.tests import base
from neutron.tests.unit import test_db_plugin as test_plugin
from neutron.tests.unit import testlib_api
PHYS_NET = 'physnet1'
PHYS_NET_2 = 'physnet2'
@ -44,13 +44,11 @@ PLUGIN_NAME = ('neutron.plugins.openvswitch.'
'ovs_neutron_plugin.OVSNeutronPluginV2')
class VlanAllocationsTest(base.BaseTestCase):
class VlanAllocationsTest(testlib_api.SqlTestCase):
def setUp(self):
super(VlanAllocationsTest, self).setUp()
db.configure_db()
ovs_db_v2.sync_vlan_allocations(VLAN_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_sync_vlan_allocations(self):
self.assertIsNone(ovs_db_v2.get_vlan_allocation(PHYS_NET,
@ -188,13 +186,11 @@ class VlanAllocationsTest(base.BaseTestCase):
ovs_db_v2.sync_vlan_allocations({})
class TunnelAllocationsTest(base.BaseTestCase):
class TunnelAllocationsTest(testlib_api.SqlTestCase):
def setUp(self):
super(TunnelAllocationsTest, self).setUp()
db.configure_db()
ovs_db_v2.sync_tunnel_allocations(TUNNEL_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
def test_sync_tunnel_allocations(self):
self.assertIsNone(ovs_db_v2.get_tunnel_allocation(TUN_MIN - 1))
@ -301,7 +297,6 @@ class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
cfg.CONF.set_override('network_vlan_ranges', ['physnet1:1000:2999'],
group='OVS')
super(NetworkBindingsTest, self).setUp(plugin=PLUGIN_NAME)
db.configure_db()
self.session = db.get_session()
def test_add_network_binding(self):

1
neutron/tests/unit/ryu/test_ryu_db.py
View File

@ -19,7 +19,6 @@ import operator
from neutron.db import api as db
from neutron.plugins.ryu.common import config # noqa
from neutron.plugins.ryu.db import api_v2 as db_api_v2
from neutron.plugins.ryu.db import models_v2 as ryu_models_v2 # noqa
from neutron.tests.unit import test_db_plugin as test_plugin

1
neutron/tests/unit/ryu/test_ryu_plugin.py
View File

@ -16,7 +16,6 @@
import mock
from neutron import manager
from neutron.plugins.ryu.db import models_v2 as ryu_models_v2 # noqa
from neutron.tests.unit.ryu import fake_ryu
from neutron.tests.unit import test_db_plugin as test_plugin

4
neutron/tests/unit/services/l3_router/test_l3_apic_plugin.py
View File

@ -18,7 +18,7 @@
import mock
from neutron.services.l3_router import l3_apic
from neutron.tests import base
from neutron.tests.unit import testlib_api
TENANT = 'tenant1'
TENANT_CONTRACT = 'abcd'
@ -55,7 +55,7 @@ class FakePort(object):
self.subnet_id = SUBNET
class TestCiscoApicL3Plugin(base.BaseTestCase):
class TestCiscoApicL3Plugin(testlib_api.SqlTestCase):
def setUp(self):
super(TestCiscoApicL3Plugin, self).setUp()

6
neutron/tests/unit/services/vpn/service_drivers/test_cisco_ipsec.py
View File

@ -16,7 +16,6 @@ import mock
# from oslo.config import cfg
from neutron import context as n_ctx
from neutron.db import api as dbapi
from neutron.openstack.common import uuidutils
from neutron.plugins.common import constants
# from neutron.services.vpn import plugin as vpn_plugin
@ -24,6 +23,7 @@ from neutron.services.vpn.service_drivers import cisco_csr_db as csr_db
from neutron.services.vpn.service_drivers import cisco_ipsec as ipsec_driver
from neutron.services.vpn.service_drivers import cisco_validator as validator
from neutron.tests import base
from neutron.tests.unit import testlib_api
_uuid = uuidutils.generate_uuid
@ -328,14 +328,12 @@ class TestCiscoIPsecDriverMapping(base.BaseTestCase):
tenant_id='1000')
class TestCiscoIPsecDriver(base.BaseTestCase):
class TestCiscoIPsecDriver(testlib_api.SqlTestCase):
"""Test that various incoming requests are sent to device driver."""
def setUp(self):
super(TestCiscoIPsecDriver, self).setUp()
dbapi.configure_db()
self.addCleanup(dbapi.clear_db)
mock.patch('neutron.common.rpc.create_connection').start()
l3_agent = mock.Mock()

21
neutron/tests/unit/test_db_plugin.py
View File

@ -32,7 +32,6 @@ from neutron.common import ipv6_utils
from neutron.common import test_lib
from neutron.common import utils
from neutron import context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import models_v2
from neutron import manager
@ -88,12 +87,6 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
if not plugin:
plugin = DB_PLUGIN_KLASS
# Create the default configurations
args = ['--config-file', base.etcdir('neutron.conf.test')]
# If test_config specifies some config-file, use it, as well
for config_file in test_lib.test_config.get('config_files', []):
args.extend(['--config-file', config_file])
self.config_parse(args=args)
# Update the plugin
self.setup_coreplugin(plugin)
cfg.CONF.set_override(
@ -149,13 +142,18 @@ class NeutronDbPluginV2TestCase(testlib_api.WebTestCase):
self._skip_native_pagination = None
self._skip_native_sortin = None
self.ext_api = None
# NOTE(jkoelker) for a 'pluggable' framework, Neutron sure
# doesn't like when the plugin changes ;)
db.clear_db()
# Restore the original attribute map
attributes.RESOURCE_ATTRIBUTE_MAP = self._attribute_map_bk
super(NeutronDbPluginV2TestCase, self).tearDown()
def setup_config(self):
# Create the default configurations
args = ['--config-file', base.etcdir('neutron.conf.test')]
# If test_config specifies some config-file, use it, as well
for config_file in test_lib.test_config.get('config_files', []):
args.extend(['--config-file', config_file])
self.config_parse(args=args)
def _req(self, method, resource, data=None, fmt=None, id=None, params=None,
action=None, subresource=None, sub_id=None, context=None):
fmt = fmt or self.fmt
@ -3938,7 +3936,7 @@ class TestNeutronDbPluginV2(base.BaseTestCase):
['b', '192.168.1.112', '192.168.1.120']], actual)
class NeutronDbPluginV2AsMixinTestCase(base.BaseTestCase):
class NeutronDbPluginV2AsMixinTestCase(testlib_api.SqlTestCase):
"""Tests for NeutronDbPluginV2 as Mixin.
While NeutronDbPluginV2TestCase checks NeutronDbPlugin and all plugins as
@ -3957,7 +3955,6 @@ class NeutronDbPluginV2AsMixinTestCase(base.BaseTestCase):
'admin_state_up': True,
'tenant_id': 'test-tenant',
'shared': False}}
self.addCleanup(db.clear_db)
def test_create_network_with_default_status(self):
net = self.plugin.create_network(self.context, self.net_data)

3
neutron/tests/unit/test_db_plugin_level.py
View File

@ -18,12 +18,11 @@ from neutron.common import constants
from neutron.common import exceptions as n_exc
from neutron import context
from neutron import manager
from neutron.tests import base
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import testlib_api
class TestNetworks(base.BaseTestCase):
class TestNetworks(testlib_api.SqlTestCase):
def setUp(self):
super(TestNetworks, self).setUp()
self._tenant_id = 'test-tenant'

7
neutron/tests/unit/test_dhcp_scheduler.py
View File

@ -20,22 +20,19 @@ from neutron.common import topics
from neutron import context
from neutron.db import agents_db
from neutron.db import agentschedulers_db
from neutron.db import api as db
from neutron.db import models_v2
from neutron.openstack.common import timeutils
from neutron.scheduler import dhcp_agent_scheduler
from neutron.tests import base
from neutron.tests.unit import testlib_api
class DhcpSchedulerTestCase(base.BaseTestCase):
class DhcpSchedulerTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(DhcpSchedulerTestCase, self).setUp()
db.configure_db()
self.ctx = context.get_admin_context()
self.network_id = 'foo_network_id'
self._save_networks([self.network_id])
self.addCleanup(db.clear_db)
def _get_agents(self, hosts):
return [

5
neutron/tests/unit/test_extension_ext_gw_mode.py
View File

@ -27,9 +27,9 @@ from neutron.db import models_v2
from neutron.extensions import l3
from neutron.extensions import l3_ext_gw_mode
from neutron.openstack.common import uuidutils
from neutron.tests import base
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_l3_plugin
from neutron.tests.unit import testlib_api
_uuid = uuidutils.generate_uuid
FAKE_GW_PORT_ID = _uuid()
@ -74,7 +74,7 @@ class TestDbSepPlugin(test_l3_plugin.TestL3NatServicePlugin,
supported_extension_aliases = ["router", "ext-gw-mode"]
class TestL3GwModeMixin(base.BaseTestCase):
class TestL3GwModeMixin(testlib_api.SqlTestCase):
def setUp(self):
super(TestL3GwModeMixin, self).setUp()
@ -84,7 +84,6 @@ class TestL3GwModeMixin(base.BaseTestCase):
# Patch the context
ctx_patcher = mock.patch('neutron.context', autospec=True)
mock_context = ctx_patcher.start()
self.addCleanup(db_api.clear_db)
self.context = mock_context.get_admin_context()
# This ensure also calls to elevated work in unit tests
self.context.elevated.return_value = self.context

5
neutron/tests/unit/test_l3_plugin.py
View File

@ -26,7 +26,6 @@ from neutron.api.v2 import attributes
from neutron.common import constants as l3_constants
from neutron.common import exceptions as n_exc
from neutron import context
from neutron.db import api as qdbapi
from neutron.db import common_db_mixin
from neutron.db import db_base_plugin_v2
from neutron.db import external_net_db
@ -35,7 +34,6 @@ from neutron.db import l3_attrs_db
from neutron.db import l3_db
from neutron.db import l3_dvr_db
from neutron.db import l3_rpc_base
from neutron.db import model_base
from neutron.extensions import external_net
from neutron.extensions import l3
from neutron.extensions import portbindings
@ -294,9 +292,6 @@ class TestL3NatServicePlugin(common_db_mixin.CommonDbMixin,
supported_extension_aliases = ["router"]
def __init__(self):
qdbapi.register_models(base=model_base.BASEV2)
def get_plugin_type(self):
return service_constants.L3_ROUTER_NAT

4
neutron/tests/unit/test_l3_schedulers.py
View File

@ -35,9 +35,9 @@ from neutron.extensions import l3 as ext_l3
from neutron import manager
from neutron.openstack.common import timeutils
from neutron.scheduler import l3_agent_scheduler
from neutron.tests import base
from neutron.tests.unit import test_db_plugin
from neutron.tests.unit import test_l3_plugin
from neutron.tests.unit import testlib_api
HOST = 'my_l3_host'
FIRST_L3_AGENT = {
@ -351,7 +351,7 @@ class L3DvrScheduler(l3_db.L3_NAT_db_mixin,
pass
class L3DvrSchedulerTestCase(base.BaseTestCase):
class L3DvrSchedulerTestCase(testlib_api.SqlTestCase):
def setUp(self):
plugin = 'neutron.plugins.ml2.plugin.Ml2Plugin'

4
neutron/tests/unit/test_quota_ext.py
View File

@ -25,7 +25,6 @@ from neutron.api.v2 import attributes
from neutron.common import config
from neutron.common import exceptions
from neutron import context
from neutron.db import api as db
from neutron.db import quota_db
from neutron import quota
from neutron.tests import base
@ -68,7 +67,6 @@ class QuotaExtensionTestCase(testlib_api.WebTestCase):
# extra1 here is added later, so have to do it manually
quota.QUOTAS.register_resource_by_name('extra1')
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
db.configure_db()
app = config.load_paste_app('extensions_test_app')
ext_middleware = extensions.ExtensionMiddleware(app, ext_mgr=ext_mgr)
self.api = webtest.TestApp(ext_middleware)
@ -76,8 +74,6 @@ class QuotaExtensionTestCase(testlib_api.WebTestCase):
def tearDown(self):
self.api = None
self.plugin = None
db.clear_db()
# Restore the global RESOURCE_ATTRIBUTE_MAP
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
super(QuotaExtensionTestCase, self).tearDown()

5
neutron/tests/unit/test_routerserviceinsertion.py
View File

@ -21,7 +21,6 @@ from neutron.api.v2 import attributes
from neutron.api.v2 import router
from neutron.common import config
from neutron import context as q_context
from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import l3_db
from neutron.db.loadbalancer import loadbalancer_db as lb_db
@ -31,7 +30,6 @@ from neutron.db import servicetype_db as st_db
from neutron.extensions import routedserviceinsertion as rsi
from neutron.extensions import routerservicetype as rst
from neutron.plugins.common import constants
from neutron.tests import base
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import testlib_api
from neutron import wsgi
@ -152,7 +150,7 @@ class RouterServiceInsertionTestPlugin(
pass
class RouterServiceInsertionTestCase(base.BaseTestCase):
class RouterServiceInsertionTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(RouterServiceInsertionTestCase, self).setUp()
plugin = (
@ -192,7 +190,6 @@ class RouterServiceInsertionTestCase(base.BaseTestCase):
def tearDown(self):
self.api = None
db.clear_db()
super(RouterServiceInsertionTestCase, self).tearDown()
def _setup_core_resources(self):

5
neutron/tests/unit/test_servicetype.py
View File

@ -24,12 +24,10 @@ import webtest
from neutron.api import extensions
from neutron.common import exceptions as n_exc
from neutron import context
from neutron.db import api as db_api
from neutron.db import servicetype_db as st_db
from neutron.extensions import servicetype
from neutron.plugins.common import constants
from neutron.services import provider_configuration as provconf
from neutron.tests import base
from neutron.tests.unit import dummy_plugin as dp
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import test_db_plugin
@ -44,7 +42,7 @@ _uuid = test_api_v2._uuid
_get_path = test_api_v2._get_path
class ServiceTypeManagerTestCase(base.BaseTestCase):
class ServiceTypeManagerTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(ServiceTypeManagerTestCase, self).setUp()
st_db.ServiceTypeManager._instance = None
@ -219,7 +217,6 @@ class ServiceTypeManagerExtTestCase(ServiceTypeExtensionTestCaseBase):
':lbaas:driver_path',
constants.DUMMY + ':dummy:dummy_dr'],
'service_providers')
self.addCleanup(db_api.clear_db)
super(ServiceTypeManagerExtTestCase, self).setUp()
def _list_service_providers(self):

21
neutron/tests/unit/testlib_api.py
View File

@ -16,6 +16,10 @@
import testtools
from neutron.api.v2 import attributes
from neutron.db import api as db_api
# Import all data models
from neutron.db.migration.models import head # noqa
from neutron.db import model_base
from neutron.tests import base
from neutron import wsgi
@ -49,7 +53,22 @@ def create_request(path, body, content_type, method='GET',
return req
class WebTestCase(base.BaseTestCase):
class SqlTestCase(base.BaseTestCase):
def setUp(self):
super(SqlTestCase, self).setUp()
# Register all data models
engine = db_api.get_engine()
model_base.BASEV2.metadata.create_all(engine)
def unregister_models():
"""Unregister all data models."""
model_base.BASEV2.metadata.drop_all(engine)
self.addCleanup(unregister_models)
class WebTestCase(SqlTestCase):
fmt = 'json'
def setUp(self):

7
neutron/tests/unit/vmware/db/test_lsn_db.py
View File

@ -16,19 +16,16 @@
from sqlalchemy import orm
from neutron import context
from neutron.db import api as db
from neutron.plugins.vmware.common import exceptions as p_exc
from neutron.plugins.vmware.dbexts import lsn_db
from neutron.tests import base
from neutron.tests.unit import testlib_api
class LSNTestCase(base.BaseTestCase):
class LSNTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(LSNTestCase, self).setUp()
db.configure_db()
self.ctx = context.get_admin_context()
self.addCleanup(db.clear_db)
self.net_id = 'foo_network_id'
self.lsn_id = 'foo_lsn_id'
self.lsn_port_id = 'foo_port_id'

7
neutron/tests/unit/vmware/db/test_nsx_db.py
View File

@ -16,20 +16,17 @@
from oslo.db import exception as d_exc
from neutron import context
from neutron.db import api as db
from neutron.db import models_v2
from neutron.plugins.vmware.dbexts import db as nsx_db
from neutron.plugins.vmware.dbexts import models
from neutron.tests import base
from neutron.tests.unit import testlib_api
class NsxDBTestCase(base.BaseTestCase):
class NsxDBTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(NsxDBTestCase, self).setUp()
db.configure_db()
self.ctx = context.get_admin_context()
self.addCleanup(db.clear_db)
def _setup_neutron_network_and_port(self, network_id, port_id):
with self.ctx.session.begin(subtransactions=True):

6
neutron/tests/unit/vmware/test_dhcpmeta.py
View File

@ -20,7 +20,6 @@ from oslo.config import cfg
from neutron.common import constants as n_consts
from neutron.common import exceptions as n_exc
from neutron import context
from neutron.db import api as db
from neutron.plugins.vmware.api_client import exception
from neutron.plugins.vmware.common import exceptions as p_exc
from neutron.plugins.vmware.dbexts import lsn_db
@ -30,6 +29,7 @@ from neutron.plugins.vmware.dhcp_meta import migration as mig_man
from neutron.plugins.vmware.dhcp_meta import nsx
from neutron.plugins.vmware.dhcp_meta import rpc
from neutron.tests import base
from neutron.tests.unit import testlib_api
class DhcpMetadataBuilderTestCase(base.BaseTestCase):
@ -704,7 +704,7 @@ class LsnManagerTestCase(base.BaseTestCase):
mock.ANY, mock.ANY, mock.ANY, mock.ANY)
class PersistentLsnManagerTestCase(base.BaseTestCase):
class PersistentLsnManagerTestCase(testlib_api.SqlTestCase):
def setUp(self):
super(PersistentLsnManagerTestCase, self).setUp()
@ -715,7 +715,6 @@ class PersistentLsnManagerTestCase(base.BaseTestCase):
self.mac = 'aa:bb:cc:dd:ee:ff'
self.lsn_port_id = 'foo_lsn_port_id'
self.tenant_id = 'foo_tenant_id'
db.configure_db()
nsx.register_dhcp_opts(cfg)
nsx.register_metadata_opts(cfg)
lsn_man.register_lsn_opts(cfg)
@ -723,7 +722,6 @@ class PersistentLsnManagerTestCase(base.BaseTestCase):
self.context = context.get_admin_context()
self.mock_lsn_api_p = mock.patch.object(lsn_man, 'lsn_api')
self.mock_lsn_api = self.mock_lsn_api_p.start()
self.addCleanup(db.clear_db)
def test_lsn_get(self):
lsn_db.lsn_add(self.context, self.net_id, self.lsn_id)

3
neutron/tests/unit/vmware/test_nsx_sync.py
View File

@ -37,6 +37,7 @@ from neutron.plugins.vmware import nsxlib
from neutron.plugins.vmware import plugin
from neutron.tests import base
from neutron.tests.unit import test_api_v2
from neutron.tests.unit import testlib_api
from neutron.tests.unit import vmware
from neutron.tests.unit.vmware.apiclient import fake
@ -259,7 +260,7 @@ class SyncLoopingCallTestCase(base.BaseTestCase):
self.assertTrue(synchronizer._synchronize_state.call_count)
class SyncTestCase(base.BaseTestCase):
class SyncTestCase(testlib_api.SqlTestCase):
def setUp(self):
# mock api client