openstack/aodh
Code Issues Proposed changes

Merge "Refactor tests to remove direct access to test DBManagers"

Browse Source
This commit is contained in:
Jenkins 2014-06-10 00:31:23 +00:00 committed by Gerrit Code Review
commit 3d2723ed7d
7 changed files with 125 additions and 113 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
ceilometer/tests/api/v2/test_app.py
View File

@ -32,7 +32,6 @@ from ceilometer import service
from ceilometer.tests import api as acl
from ceilometer.tests.api.v2 import FunctionalTest
from ceilometer.tests import base
from ceilometer.tests import db as tests_db
class TestApp(base.BaseTestCase):
@ -82,8 +81,6 @@ class TestApp(base.BaseTestCase):
class TestPecanApp(FunctionalTest):
db_manager = tests_db.MongoDbManager()
def test_pecan_extension_guessing_unset(self):
# check Pecan does not assume .jpg is an extension
response = self.app.get(self.PATH_PREFIX + '/meters/meter.jpg')
@ -92,8 +89,6 @@ class TestPecanApp(FunctionalTest):
class TestApiMiddleware(FunctionalTest):
db_manager = tests_db.MongoDbManager()
no_lang_translated_error = 'No lang translated error'
en_US_translated_error = 'en-US translated error'

14
ceilometer/tests/api/v2/test_statistics_scenarios.py
View File

@ -1214,6 +1214,7 @@ class TestGroupByInstance(FunctionalTest,
u'2013-08-01T14:00:00'])
@tests_db.run_with('mongodb', 'hbase', 'db2')
class TestGroupBySource(FunctionalTest,
tests_db.MixinTestsWithBackendScenarios):
@ -1223,12 +1224,6 @@ class TestGroupBySource(FunctionalTest,
# moved to TestGroupByInstance with all the other group by statistics
# tests.
scenarios = [
('mongodb', {'db_manager': tests_db.MongoDbManager()}),
('hbase', {'db_manager': tests_db.HBaseManager()}),
('db2', {'db_manager': tests_db.DB2Manager()}),
]
PATH = '/meters/instance/statistics'
def setUp(self):
@ -1567,6 +1562,7 @@ class TestSelectableAggregates(FunctionalTest,
'Bad aggregate: cardinality.injection_attack')
@tests_db.run_with('mongodb', 'hbase', 'db2')
class TestUnparameterizedAggregates(FunctionalTest,
tests_db.MixinTestsWithBackendScenarios):
@ -1578,12 +1574,6 @@ class TestUnparameterizedAggregates(FunctionalTest,
# For hbase & db2, the skip on NotImplementedError logic works
# in the usual way.
scenarios = [
('mongodb', {'db_manager': tests_db.MongoDbManager()}),
('hbase', {'db_manager': tests_db.HBaseManager()}),
('db2', {'db_manager': tests_db.DB2Manager()}),
]
PATH = '/meters/instance/statistics'
def setUp(self):

159
ceilometer/tests/db.py
View File

@ -21,11 +21,13 @@
"""Base classes for API tests."""
import fixtures
import os
import urlparse
import uuid
import warnings
import six
import testscenarios.testcase
from testtools import testcase
from ceilometer.openstack.common.fixture import config
import ceilometer.openstack.common.fixture.mockpatch as oslo_mock
@ -33,32 +35,95 @@ from ceilometer import storage
from ceilometer.tests import base as test_base
class TestBase(testscenarios.testcase.WithScenarios, test_base.BaseTestCase):
class MongoDbManager(fixtures.Fixture):
def __init__(self, url):
self._url = url
def setUp(self):
super(TestBase, self).setUp()
self.useFixture(self.db_manager)
self.CONF = self.useFixture(config.Config()).conf
super(MongoDbManager, self).setUp()
with warnings.catch_warnings():
warnings.filterwarnings(
action='ignore',
message='.*you must provide a username and password.*')
try:
self.conn = storage.get_connection(self.db_manager.connection)
self.connection = storage.get_connection(self.url)
except storage.StorageBadVersion as e:
self.skipTest(six.text_type(e))
raise testcase.TestSkipped(six.text_type(e))
@property
def url(self):
return '%(url)s_%(db)s' % {
'url': self._url,
'db': uuid.uuid4().hex
}
class HBaseManager(fixtures.Fixture):
def __init__(self, url):
self._url = url
def setUp(self):
super(HBaseManager, self).setUp()
self.connection = storage.get_connection(self.url)
@property
def url(self):
return '%s?table_prefix=%s' % (
self._url,
uuid.uuid4().hex
)
class SQLiteManager(fixtures.Fixture):
def __init__(self, url):
self.url = url
def setUp(self):
super(SQLiteManager, self).setUp()
self.connection = storage.get_connection(self.url)
class TestBase(testscenarios.testcase.WithScenarios, test_base.BaseTestCase):
DRIVER_MANAGERS = {
'mongodb': MongoDbManager,
'db2': MongoDbManager,
'sqlite': SQLiteManager,
'hbase': HBaseManager,
}
db_url = 'sqlite://' # NOTE(Alexei_987) Set default db url
def setUp(self):
super(TestBase, self).setUp()
engine = urlparse.urlparse(self.db_url).scheme
# NOTE(Alexei_987) Shortcut to skip expensive db setUp
test_method = self._get_test_method()
if (hasattr(test_method, '_run_with')
and engine not in test_method._run_with):
raise testcase.TestSkipped(
'Test is not applicable for %s' % engine)
self.db_manager = self._get_driver_manager(engine)(self.db_url)
self.useFixture(self.db_manager)
self.conn = self.db_manager.connection
self.conn.upgrade()
self.useFixture(oslo_mock.Patch('ceilometer.storage.get_connection',
return_value=self.conn))
self.CONF = self.useFixture(config.Config()).conf
self.CONF([], project='ceilometer')
# Set a default location for the pipeline config file so the
# tests work even if ceilometer is not installed globally on
# the system.
self.CONF.import_opt('pipeline_cfg_file', 'ceilometer.pipeline')
self.CONF.set_override(
'pipeline_cfg_file',
self.path_get('etc/ceilometer/pipeline.yaml')
@ -69,64 +134,38 @@ class TestBase(testscenarios.testcase.WithScenarios, test_base.BaseTestCase):
self.conn = None
super(TestBase, self).tearDown()
class MongoDbManager(fixtures.Fixture):
def __init__(self):
self.url = os.environ.get('CEILOMETER_TEST_MONGODB_URL')
if not self.url:
raise RuntimeError(
"No MongoDB test URL set,"
"export CEILOMETER_TEST_MONGODB_URL environment variable")
def setUp(self):
super(MongoDbManager, self).setUp()
self.connection = '%(url)s_%(db)s' % {
'url': self.url,
'db': uuid.uuid4().hex
}
def _get_driver_manager(self, engine):
manager = self.DRIVER_MANAGERS.get(engine)
if not manager:
raise ValueError('No manager available for %s' % engine)
return manager
class DB2Manager(MongoDbManager):
def __init__(self):
self.url = (os.environ.get('CEILOMETER_TEST_DB2_URL') or
os.environ.get('CEILOMETER_TEST_MONGODB_URL'))
if not self.url:
raise RuntimeError(
"No DB2 test URL set, "
"export CEILOMETER_TEST_DB2_URL environment variable")
def run_with(*drivers):
"""Used to mark tests that are only applicable for certain db driver.
Skips test if driver is not available
"""
def decorator(test):
if isinstance(test, type) and issubclass(test, TestBase):
# Decorate all test methods
for attr in dir(test):
value = getattr(test, attr)
if callable(value) and attr.startswith('test_'):
value.__func__._run_with = drivers
else:
# This is to make sure that the db2 driver is used when
# CEILOMETER_TEST_DB2_URL was not set
self.url = self.url.replace('mongodb:', 'db2:', 1)
class HBaseManager(fixtures.Fixture):
def __init__(self):
self.url = os.environ.get('CEILOMETER_TEST_HBASE_URL')
if not self.url:
self.url = 'hbase://__test__'
def setUp(self):
super(HBaseManager, self).setUp()
self.connection = '%s?table_prefix=%s' % (
self.url,
uuid.uuid4().hex)
class SQLiteManager(fixtures.Fixture):
def setUp(self):
super(SQLiteManager, self).setUp()
self.connection = 'sqlite://'
test._run_with = drivers
return test
return decorator
@six.add_metaclass(test_base.SkipNotImplementedMeta)
class MixinTestsWithBackendScenarios(object):
scenarios = [
('sqlite', {'db_manager': SQLiteManager()}),
('mongodb', {'db_manager': MongoDbManager()}),
('hbase', {'db_manager': HBaseManager()}),
('db2', {'db_manager': DB2Manager()})
('sqlite', {'db_url': 'sqlite://'}),
('mongodb', {'db_url': os.environ.get('CEILOMETER_TEST_MONGODB_URL')}),
('hbase', {'db_url': os.environ.get('CEILOMETER_TEST_HBASE_URL',
'hbase://__test__')}),
('db2', {'db_url': (os.environ.get('CEILOMETER_TEST_DB2_URL') or
os.environ.get('CEILOMETER_TEST_MONGODB_URL'))})
]

10
ceilometer/tests/storage/test_impl_hbase.py
View File

@ -31,14 +31,12 @@ from ceilometer.tests import base as test_base
from ceilometer.tests import db as tests_db
class HBaseEngineTestBase(tests_db.TestBase):
db_manager = tests_db.HBaseManager()
class ConnectionTest(HBaseEngineTestBase):
class ConnectionTest(tests_db.TestBase,
tests_db.MixinTestsWithBackendScenarios):
@tests_db.run_with('hbase')
def test_hbase_connection(self):
conn = hbase.Connection(self.db_manager.connection)
conn = hbase.Connection(self.db_manager.url)
self.assertIsInstance(conn.conn_pool.connection(), hbase.MConnection)
class TestConn(object):

22
ceilometer/tests/storage/test_impl_mongodb.py
View File

@ -31,17 +31,14 @@ from ceilometer.tests import db as tests_db
from ceilometer.tests.storage import test_storage_scenarios
class MongoDBEngineTestBase(tests_db.TestBase):
db_manager = tests_db.MongoDbManager()
class MongoDBConnection(MongoDBEngineTestBase):
@tests_db.run_with('mongodb')
class MongoDBConnection(tests_db.TestBase):
def test_connection_pooling(self):
test_conn = impl_mongodb.Connection(self.db_manager.connection)
test_conn = impl_mongodb.Connection(self.db_manager.url)
self.assertEqual(self.conn.conn, test_conn.conn)
def test_replica_set(self):
url = self.db_manager.connection + '?replicaSet=foobar'
url = self.db_manager._url + '?replicaSet=foobar'
conn = impl_mongodb.Connection(url)
self.assertTrue(conn.conn)
@ -56,8 +53,8 @@ class MongoDBConnection(MongoDBEngineTestBase):
self.assertEqual(expect, ret)
class MongoDBTestMarkerBase(test_storage_scenarios.DBTestBase,
MongoDBEngineTestBase):
@tests_db.run_with('mongodb')
class MongoDBTestMarkerBase(test_storage_scenarios.DBTestBase):
#NOTE(Fengqian): All these three test case are the same for resource
#and meter collection. As to alarm, we will set up in AlarmTestPagination.
def test_get_marker(self):
@ -85,7 +82,8 @@ class MongoDBTestMarkerBase(test_storage_scenarios.DBTestBase,
self.assertTrue(True)
class IndexTest(MongoDBEngineTestBase):
@tests_db.run_with('mongodb')
class IndexTest(tests_db.TestBase):
def test_meter_ttl_index_absent(self):
# create a fake index and check it is deleted
self.conn.db.meter.ensure_index('foo', name='meter_ttl')
@ -113,8 +111,8 @@ class IndexTest(MongoDBEngineTestBase):
name='meter_ttl'))
class AlarmTestPagination(test_storage_scenarios.AlarmTestBase,
MongoDBEngineTestBase):
@tests_db.run_with('mongodb')
class AlarmTestPagination(test_storage_scenarios.AlarmTestBase):
def test_alarm_get_marker(self):
self.add_some_alarms()
marker_pairs = {'name': 'red-alert'}

19
ceilometer/tests/storage/test_impl_sqlalchemy.py
View File

@ -38,20 +38,17 @@ from ceilometer.tests import db as tests_db
from ceilometer.tests.storage import test_storage_scenarios as scenarios
class EventTestBase(tests_db.TestBase):
# Note: Do not derive from SQLAlchemyEngineTestBase, since we
# don't want to automatically inherit all the Meter setup.
db_manager = tests_db.SQLiteManager()
@tests_db.run_with('sqlite')
class CeilometerBaseTest(tests_db.TestBase):
class CeilometerBaseTest(EventTestBase):
def test_ceilometer_base(self):
base = sql_models.CeilometerBase()
base['key'] = 'value'
self.assertEqual('value', base['key'])
class TraitTypeTest(EventTestBase):
@tests_db.run_with('sqlite')
class TraitTypeTest(tests_db.TestBase):
# TraitType is a construct specific to sqlalchemy.
# Not applicable to other drivers.
@ -83,7 +80,8 @@ class TraitTypeTest(EventTestBase):
self.assertTrue(repr.repr(tt2))
class EventTypeTest(EventTestBase):
@tests_db.run_with('sqlite')
class EventTypeTest(tests_db.TestBase):
# EventType is a construct specific to sqlalchemy
# Not applicable to other drivers.
@ -108,7 +106,8 @@ class MyException(Exception):
pass
class EventTest(EventTestBase):
@tests_db.run_with('sqlite')
class EventTest(tests_db.TestBase):
def test_string_traits(self):
model = models.Trait("Foo", models.Trait.TEXT_TYPE, "my_text")
trait = self.conn._make_trait(model, None)
@ -174,10 +173,10 @@ class EventTest(EventTestBase):
self.assertTrue(repr.repr(ev))
@tests_db.run_with('sqlite')
class RelationshipTest(scenarios.DBTestBase):
# Note: Do not derive from SQLAlchemyEngineTestBase, since we
# don't want to automatically inherit all the Meter setup.
db_manager = tests_db.SQLiteManager()
@patch.object(timeutils, 'utcnow')
def test_clear_metering_data_meta_tables(self, mock_utcnow):

9
ceilometer/tests/storage/test_pymongo_base.py
View File

@ -20,7 +20,6 @@ import datetime
from mock import call
from mock import patch
import pymongo
import testscenarios
from ceilometer.openstack.common.gettextutils import _
from ceilometer.publisher import utils
@ -29,17 +28,11 @@ from ceilometer.storage import pymongo_base
from ceilometer.tests import db as tests_db
from ceilometer.tests.storage import test_storage_scenarios
load_tests = testscenarios.load_tests_apply_scenarios
@tests_db.run_with('mongodb', 'db2')
class CompatibilityTest(test_storage_scenarios.DBTestBase,
tests_db.MixinTestsWithBackendScenarios):
scenarios = [
('mongodb', {'db_manager': tests_db.MongoDbManager()}),
('db2', {'db_manager': tests_db.DB2Manager()}),
]
def prepare_data(self):
def old_record_metering_data(self, data):
self.db.user.update(