__native_sorting_support = True
def __init__(self):
- # NOTE(jkoelker) This is an incomplete implementation. Subclasses
- # must override __init__ and setup the database
- # and not call into this class's __init__.
- # This connection is setup as memory for the tests.
db.configure_db()
@classmethod
from neutron import context as qcontext
from neutron.db import agents_db
from neutron.db import agentschedulers_db
-from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import dhcp_rpc_base
from neutron.db import external_net_db
servers = None
def __init__(self, server_timeout=None):
+ super(NeutronRestProxyV2Base, self).__init__()
# This base class is not intended to be instantiated directly.
# Extending class should set ServerPool.
if not self.servers:
"dhcp_agent_scheduler", "agent"]
def __init__(self, server_timeout=None):
+ super(NeutronRestProxyV2, self).__init__()
LOG.info(_('NeutronRestProxy: Starting plugin. Version=%s'),
version_string_with_vcs())
pl_config.register_config()
- # init DB, proxy's persistent store defaults to in-memory sql-lite DB
- db.configure_db()
# Include the BigSwitch Extensions path in the api_extensions
neutron_extensions.append_api_extensions_path(extensions.__path__)
Specify switch address and db configuration.
"""
+ super(BrocadePluginV2, self).__init__()
self.supported_extension_aliases = ["binding", "security-group",
"external-net", "router",
"extraroute", "agent",
physical_interface)
self.base_binding_dict = self._get_base_binding_dict()
portbindings_base.register_port_dict_function()
- db.configure_db()
self.ctxt = context.get_admin_context()
self.ctxt.session = db.get_session()
self._vlan_bitmap = vbm.VlanBitmap(self.ctxt)
LOG = logging.getLogger(__name__)
-def initialize():
- """Initialize the database."""
- db.configure_db()
-
-
def del_trunk_segment_binding(db_session, trunk_segment_id, segment_pairs):
"""
Delete a trunk network binding.
1. Initialize Nexus1000v and Credential DB
2. Establish communication with Cisco Nexus1000V
"""
- n1kv_db_v2.initialize()
+ super(N1kvNeutronPluginV2, self).__init__()
c_cred.Store.initialize()
self._initialize_network_ranges()
self._setup_vsm()
LOG = logging.getLogger(__name__)
-def initialize():
- db.configure_db()
-
-
def sync_network_states(network_vlan_ranges):
"""Synchronize network_states table with current configured VLAN ranges."""
return self._aliases
def __init__(self):
+ super(LinuxBridgePluginV2, self).__init__()
self.base_binding_dict = {
portbindings.VIF_TYPE: portbindings.VIF_TYPE_BRIDGE,
portbindings.CAPABILITIES: {
portbindings.CAP_PORT_FILTER:
'security-group' in self.supported_extension_aliases}}
- db.initialize()
self._parse_network_vlan_ranges()
db.sync_network_states(self.network_vlan_ranges)
self.tenant_network_type = cfg.CONF.VLANS.tenant_network_type
extraroute_db.ExtraRoute_db_mixin):
def __init__(self, configfile=None):
+ super(MetaPluginV2, self).__init__()
LOG.debug(_("Start initializing metaplugin"))
self.supported_extension_aliases = ['flavor', 'external-net',
'router', 'ext-gw-mode',
raise exc.Invalid(_('default_l3_flavor %s is not plugin list') %
self.default_l3_flavor)
- db.configure_db()
-
self.extension_map = {}
if not cfg.CONF.META.extension_map == '':
extension_list = [method_set.split(':')
from oslo.config import cfg
-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
supported_extension_aliases = ["external-net", "router"]
def __init__(self, configfile=None):
- db.configure_db()
+ super(ProxyPluginV2, self).__init__()
self.neutron = client.Client(
username=cfg.CONF.PROXY.admin_user,
password=cfg.CONF.PROXY.admin_password,
from neutron.common import topics
from neutron.db import agents_db
from neutron.db import agentschedulers_db
-from neutron.db import api as db
from neutron.db import db_base_plugin_v2
from neutron.db import dhcp_rpc_base
from neutron.db import external_net_db
__native_bulk_support = False
def __init__(self):
+ super(MidonetPluginV2, self).__init__()
# Read config values
midonet_conf = cfg.CONF.MIDONET
midonet_uri = midonet_conf.midonet_uri
raise MidonetPluginException(msg=msg)
self.setup_rpc()
- db.configure_db()
self.base_binding_dict = {
portbindings.VIF_TYPE: portbindings.VIF_TYPE_MIDONET,
LOG = log.getLogger(__name__)
-def initialize():
- db_api.configure_db()
-
-
def add_network_segment(session, network_id, segment):
with session.begin(subtransactions=True):
record = models.NetworkSegment(
return self._aliases
def __init__(self):
+ super(Ml2Plugin, self).__init__()
# First load drivers, then initialize DB, then initialize drivers
self.type_manager = managers.TypeManager()
self.mechanism_manager = managers.MechanismManager()
- db.initialize()
self.type_manager.initialize()
self.mechanism_manager.initialize()
# bulk support depends on the underlying drivers
LOG = logging.getLogger(__name__)
-def initialize():
- db.configure_db()
-
-
def _remove_non_allocatable_vlans(session, allocations,
physical_network, vlan_ids):
if physical_network in allocations:
def __init__(self):
"""Start Mellanox Neutron Plugin."""
- db.initialize()
+ super(MellanoxEswitchPlugin, self).__init__()
self._parse_network_vlan_ranges()
db.sync_network_states(self.network_vlan_ranges)
self._set_tenant_network_type()
return resource_map[resource]
-def initialize():
- db.configure_db()
-
-
def clear_db(base=model_base.BASEV2):
db.clear_db(base)
return self._aliases
def __init__(self):
-
- ndb.initialize()
+ super(NECPluginV2, self).__init__()
self.ofc = ofc_manager.OFCManager()
self.base_binding_dict = self._get_base_binding_dict()
portbindings_base.register_port_dict_function()
from neutron import context as q_context
from neutron.db import agentschedulers_db
from neutron.db import allowedaddresspairs_db as addr_pair_db
-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
novazone_cluster_map = {}
def __init__(self):
-
+ super(NvpPluginV2, self).__init__()
# TODO(salv-orlando): Replace These dicts with
# collections.defaultdict for better handling of default values
# Routines for managing logical ports in NVP
pbin.CAP_PORT_FILTER:
'security-group' in self.supported_extension_aliases}}
- db.configure_db()
self._extend_fault_map()
self.setup_dhcpmeta_access()
# Set this flag to false as the default gateway has not
LOG = logging.getLogger(__name__)
-def initialize():
- db.configure_db()
-
-
def get_network_binding(session, network_id):
session = session or db.get_session()
try:
attributes.NETWORKS, ['_extend_network_dict_provider_ovs'])
def __init__(self, configfile=None):
+ super(OVSNeutronPluginV2, self).__init__()
self.base_binding_dict = {
portbindings.VIF_TYPE: portbindings.VIF_TYPE_OVS,
portbindings.CAPABILITIES: {
portbindings.CAP_PORT_FILTER:
'security-group' in self.supported_extension_aliases}}
- ovs_db_v2.initialize()
self._parse_network_vlan_ranges()
ovs_db_v2.sync_vlan_allocations(self.network_vlan_ranges)
self.tenant_network_type = cfg.CONF.OVS.tenant_network_type
from oslo.config import cfg
from neutron.api.v2 import attributes
-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
def __init__(self):
LOG.info(_('Neutron PLUMgrid Director: Starting Plugin'))
- # Plugin DB initialization
- db.configure_db()
-
+ super(NeutronPluginPLUMgridV2, self).__init__()
self.plumgrid_init()
LOG.debug(_('Neutron PLUMgrid Director: Neutron server with '
return self._aliases
def __init__(self, configfile=None):
+ super(RyuNeutronPluginV2, self).__init__()
self.base_binding_dict = {
portbindings.VIF_TYPE: portbindings.VIF_TYPE_OVS,
portbindings.CAPABILITIES: {
portbindings.CAP_PORT_FILTER:
'security-group' in self.supported_extension_aliases}}
portbindings_base.register_port_dict_function()
- db.configure_db()
self.tunnel_key = db_api_v2.TunnelKey(
cfg.CONF.OVS.tunnel_key_min, cfg.CONF.OVS.tunnel_key_max)
self.ofp_api_host = cfg.CONF.OVS.openflow_rest_api
def setUp(self):
super(VlanAllocationsTest, self).setUp()
- n1kv_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
n1kv_db_v2.sync_vlan_allocations(self.session, VLAN_RANGES)
def setUp(self):
super(VxlanAllocationsTest, self).setUp()
- n1kv_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
n1kv_db_v2.sync_vxlan_allocations(self.session, VXLAN_RANGES)
def setUp(self):
super(NetworkBindingsTest, self).setUp()
- n1kv_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
def tearDown(self):
def setUp(self):
super(NetworkProfileTests, self).setUp()
- n1kv_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
def tearDown(self):
def setUp(self):
super(PolicyProfileTests, self).setUp()
- n1kv_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
def tearDown(self):
def setUp(self):
super(ProfileBindingTests, self).setUp()
- n1kv_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
def tearDown(self):
"""
def test_db(self):
- n1kv_db_v2.initialize()
+ db.configure_db()
class NetworkStatesTest(base.BaseTestCase):
def setUp(self):
super(NetworkStatesTest, self).setUp()
- lb_db.initialize()
+ db.configure_db()
lb_db.sync_network_states(VLAN_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
cfg.CONF.set_override('network_vlan_ranges', ['physnet1:1000:2999'],
group='VLANS')
super(NetworkBindingsTest, self).setUp(plugin=PLUGIN_NAME)
- lb_db.initialize()
+ db.configure_db()
self.session = db.get_session()
def test_add_network_binding(self):
from neutron.common import exceptions as exc
import neutron.db.api as db
-from neutron.plugins.ml2 import db as ml2_db
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_gre
from neutron.tests import base
def setUp(self):
super(GreTypeTest, self).setUp()
- ml2_db.initialize()
+ db.configure_db()
self.driver = type_gre.GreTypeDriver()
self.driver.gre_id_ranges = TUNNEL_RANGES
self.driver._sync_gre_allocations()
def setUp(self):
super(GreTypeMultiRangeTest, self).setUp()
- ml2_db.initialize()
+ db.configure_db()
self.driver = type_gre.GreTypeDriver()
self.driver.gre_id_ranges = self.TUNNEL_MULTI_RANGES
self.driver._sync_gre_allocations()
from neutron.common import exceptions as exc
from neutron.db import api as db
from neutron.plugins.common import constants as p_const
-from neutron.plugins.ml2 import db as ml2_db
from neutron.plugins.ml2 import driver_api as api
from neutron.plugins.ml2.drivers import type_vxlan
from neutron.tests import base
class VxlanTypeTest(base.BaseTestCase):
def setUp(self):
super(VxlanTypeTest, self).setUp()
- ml2_db.initialize()
+ db.configure_db()
cfg.CONF.set_override('vni_ranges', [TUNNEL_RANGES],
group='ml2_type_vxlan')
cfg.CONF.set_override('vxlan_group', MULTICAST_GROUP,
def setUp(self):
super(VxlanTypeMultiRangeTest, self).setUp()
- ml2_db.initialize()
+ db.configure_db()
self.driver = type_vxlan.VxlanTypeDriver()
self.driver.vxlan_vni_ranges = self.TUNNEL_MULTI_RANGES
self.driver._sync_vxlan_allocations()
class SegmentationIdAllocationTest(base.BaseTestCase):
def setUp(self):
super(SegmentationIdAllocationTest, self).setUp()
- mlnx_db.initialize()
+ db.configure_db()
mlnx_db.sync_network_states(VLAN_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
class NetworkBindingsTest(test_plugin.NeutronDbPluginV2TestCase):
def setUp(self):
super(NetworkBindingsTest, self).setUp()
- mlnx_db.initialize()
+ db.configure_db()
self.session = db.get_session()
def test_add_network_binding(self):
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
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')
- ndb.initialize()
self.addCleanup(ndb.clear_db)
self.ofc = ofc_manager.OFCManager()
# NOTE: enable_autocheck() is a feature of StubOFCDriver
class VlanAllocationsTest(base.BaseTestCase):
def setUp(self):
super(VlanAllocationsTest, self).setUp()
- ovs_db_v2.initialize()
+ db.configure_db()
ovs_db_v2.sync_vlan_allocations(VLAN_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
class TunnelAllocationsTest(base.BaseTestCase):
def setUp(self):
super(TunnelAllocationsTest, self).setUp()
- ovs_db_v2.initialize()
+ db.configure_db()
ovs_db_v2.sync_tunnel_allocations(TUNNEL_RANGES)
self.session = db.get_session()
self.addCleanup(db.clear_db)
cfg.CONF.set_override('network_vlan_ranges', ['physnet1:1000:2999'],
group='OVS')
super(NetworkBindingsTest, self).setUp(plugin=PLUGIN_NAME)
- ovs_db_v2.initialize()
+ db.configure_db()
self.session = db.get_session()
def test_add_network_binding(self):
from neutron import context
from neutron.db import api as db
from neutron.db import quota_db
-from neutron.plugins.linuxbridge.db import l2network_db_v2
from neutron import quota
from neutron.tests import base
from neutron.tests.unit import test_api_v2
# extra1 here is added later, so have to do it manually
quota.QUOTAS.register_resource_by_name('extra1')
ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
- l2network_db_v2.initialize()
+ 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)
Code Review / openstack-build / neutron-build.git / commitdiff