Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.test.engine = self.test.transaction_engine
self.test.sessionmaker = session.get_maker(
self.test.transaction_engine)
else:
self.test.engine = self.test.db.engine
self.test.sessionmaker = session.get_maker(self.test.engine)
self.addCleanup(setattr, self.test, 'sessionmaker', None)
self.addCleanup(setattr, self.test, 'engine', None)
self.test.enginefacade = enginefacade._TestTransactionFactory(
self.test.engine, self.test.sessionmaker, apply_global=True)
self.addCleanup(self.test.enginefacade.dispose_global)
@debtcollector.removals.removed_class(
"DbTestCase",
message="Please use oslo_db.sqlalchemy.test_fixtures directly")
class DbTestCase(test_base.BaseTestCase):
"""Base class for testing of DB code.
"""
FIXTURE = DbFixture
SCHEMA_SCOPE = None
SKIP_ON_UNAVAILABLE_DB = True
_db_not_available = {}
_schema_resources = {}
_database_resources = {}
def _get_db_resource_not_available_reason(self):
'only on %s. Current engine is %s.')
args = (reflection.get_callable_name(f), ', '.join(dialects),
self.engine.name)
self.skipTest(msg % args)
else:
return f(self)
return ins_wrap
return wrap
@debtcollector.removals.removed_class("MySQLOpportunisticFixture")
class MySQLOpportunisticFixture(DbFixture):
DRIVER = 'mysql'
@debtcollector.removals.removed_class("PostgreSQLOpportunisticFixture")
class PostgreSQLOpportunisticFixture(DbFixture):
DRIVER = 'postgresql'
@debtcollector.removals.removed_class("MySQLOpportunisticTestCase")
class MySQLOpportunisticTestCase(OpportunisticTestCase):
FIXTURE = MySQLOpportunisticFixture
@debtcollector.removals.removed_class("PostgreSQLOpportunisticTestCase")
class PostgreSQLOpportunisticTestCase(OpportunisticTestCase):
FIXTURE = PostgreSQLOpportunisticFixture
try:
from oslotest import base as test_base
except ImportError:
raise NameError('Oslotest is not installed. Please add oslotest in your'
' test-requirements')
from oslo_utils import reflection
from oslo_db import exception
from oslo_db.sqlalchemy import enginefacade
from oslo_db.sqlalchemy import provision
from oslo_db.sqlalchemy import session
@debtcollector.removals.removed_class(
"DbFixture",
message="Please use oslo_db.sqlalchemy.test_fixtures directly")
class DbFixture(fixtures.Fixture):
"""Basic database fixture.
Allows to run tests on various db backends, such as SQLite, MySQL and
PostgreSQL. By default use sqlite backend. To override default backend
uri set env variable OS_TEST_DBAPI_ADMIN_CONNECTION with database admin
credentials for specific backend.
"""
DRIVER = "sqlite"
# these names are deprecated, and are not used by DbFixture.
# they are here for backwards compatibility with test suites that
# are referring to them directly.
@removals.removed_class('MaximumEntropy',
replacement="dit.algorithms.scipy_optimizers.MaxEntOptimizer",
message="Please see methods in dit.algorithms.distribution_optimizers.py.",
version='1.0.1')
class MaximumEntropy(CVXOPT_Template):
"""
Find maximum entropy distribution.
"""
def build_function(self):
self.func = negentropy
@removals.removed_class('MarginalMaximumEntropy',
replacement="dit.algorithms.scipy_optimizers.MaxEntOptimizer",
message="Please see methods in dit.algorithms.distribution_optimizers.py.",
version='1.0.1')
class MarginalMaximumEntropy(MaximumEntropy):
"""
Find maximum entropy distribution subject to `k`-way marginal constraints.
`k = 0` should reproduce the behavior of MaximumEntropy.
"""
def __init__(self, dist, k, tol=None, prng=None):
"""
Initialize optimizer.
Parameters
----------
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from debtcollector import removals
from stevedore import extension
from designateclient import exceptions
from designateclient import utils
from designateclient import version
@removals.removed_class(
'designateclient.v1.Client',
replacement='designateclient.v2.client.Client',
message='Designate v1 API is being retired, and the v1 Client class will '
'stop functioning. Please update code to the '
'designateclient.v2.client.Client class. The API is deprecated',
version='2.2.0',
removal_version='?',
stacklevel=3)
class Client(object):
"""Client for the Designate v1 API"""
def __init__(self, endpoint=None, username=None, user_id=None,
user_domain_id=None, user_domain_name=None, password=None,
tenant_name=None, tenant_id=None, domain_name=None,
domain_id=None, project_name=None,
project_id=None, project_domain_name=None,
OPTS = [
cfg.StrOpt('file_path',
help='Name and the location of the file to record '
'meters.'),
cfg.IntOpt('max_bytes',
default=0,
help='The max size of the file.'),
cfg.IntOpt('backup_count',
default=0,
help='The max number of the files to keep.'),
]
cfg.CONF.register_opts(OPTS, group="dispatcher_file")
@removals.removed_class("FileDispatcher", message="Use file publisher instead",
removal_version="9.0.0")
class FileDispatcher(dispatcher.MeterDispatcherBase,
dispatcher.EventDispatcherBase):
"""Dispatcher class for recording metering data to a file.
The dispatcher class which logs each meter and/or event into a file
configured in ceilometer configuration file. An example configuration may
look like the following:
[dispatcher_file]
file_path = /tmp/meters
To enable this dispatcher, the following section needs to be present in
ceilometer.conf file
[DEFAULT]
class CliOptRegisteredError(cfg.Error):
"""Raised when registering cli opt not in original ConfigOpts.
.. versionadded:: 1.12
"""
def __str__(self):
ret = "Cannot register a cli option that was not present in the" \
" original ConfigOpts."
if self.msg:
ret += ": " + self.msg
return ret
@removals.removed_class('ConfigFilter')
class ConfigFilter(collections.Mapping):
"""A helper class which wraps a ConfigOpts object.
ConfigFilter enforces the explicit declaration of dependencies on external
options and allows private options which are not registered with the
wrapped Configopts object.
"""
def __init__(self, conf):
"""Construct a ConfigFilter object.
:param conf: a ConfigOpts object
"""
self._conf = conf
self._fconf = cfg.ConfigOpts()
self._sync()
from networking_bgpvpn.neutron.services.service_drivers.opencontrail import \
exceptions as oc_exc
from networking_bgpvpn.neutron.services.service_drivers.opencontrail import \
opencontrail_client
OPENCONTRAIL_BGPVPN_DRIVER_NAME = 'OpenContrail'
LOG = log.getLogger(__name__)
MESSAGE = ("replaced by a new driver which could be found in the Contrail "
"Neutron monolithic core plugin tree: 'neutron_plugin_contrail."
"plugins.opencontrail.networking_bgpvpn.contrail."
"ContrailBGPVPNDriver'")
@removals.removed_class('OpenContrailBGPVPNDriver', version='Queens',
removal_version='Rocky', message=MESSAGE)
class OpenContrailBGPVPNDriver(driver_api.BGPVPNDriverBase):
"""BGP VPN Service Driver class for OpenContrail."""
def __init__(self, service_plugin):
super(OpenContrailBGPVPNDriver, self).__init__(service_plugin)
LOG.debug("OpenContrailBGPVPNDriver service_plugin : %s",
service_plugin)
def _get_opencontrail_api_client(self, context):
return opencontrail_client.OpenContrailAPIBaseClient()
def _locate_rt(self, oc_client, rt_fq_name):
try:
rt_uuid = oc_client.fqname_to_id('route-target', rt_fq_name)
except oc_exc.OpenContrailAPINotFound:
import logging
from debtcollector import removals
from six.moves.urllib import parse as urlparse
from keystoneclient import exceptions
from keystoneclient import httpclient
from keystoneclient.i18n import _, _LE
_logger = logging.getLogger(__name__)
# NOTE(jamielennox): To be removed after Pike.
@removals.removed_class('keystoneclient.generic.client.Client',
message='Use keystoneauth discovery',
version='3.9.0',
removal_version='4.0.0')
class Client(httpclient.HTTPClient):
"""Client for the OpenStack Keystone pre-version calls API.
:param string endpoint: A user-supplied endpoint URL for the keystone
service.
:param integer timeout: Allows customization of the timeout for client
http requests. (optional)
Example::
>>> from keystoneclient.generic import client
>>> root = client.Client(auth_url=KEYSTONE_URL)
>>> versions = root.discover()