← Back to branch summary

~ubuntu-branches/ubuntu/vivid/ceilometer/vivid

« back to all changes in this revision

Viewing changes to ceilometer/tests/storage/test_impl_sqlalchemy.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2014-03-06 14:44:28 UTC
  • mto: (28.1.1 utopic-proposed) (1.2.1)
  • mto: This revision was merged to the branch mainline in revision 19.
  • Revision ID: package-import@ubuntu.com-20140306144428-rvphsh4igwyulzf0
Tags: upstream-2014.1~b3
Import upstream version 2014.1~b3

Show diffs side-by-side

added added

removed removed

Lines of Context:
ceilometer/tests/storage/test_impl_sqlalchemy.py
28
28
 
29
29
from mock import patch
30
30
 
31
 
import ceilometer.openstack.common.db.sqlalchemy.session as sqlalchemy_session
 
31
from ceilometer.openstack.common.fixture import config
32
32
from ceilometer.openstack.common import timeutils
33
33
from ceilometer.storage import models
34
34
from ceilometer.storage.sqlalchemy import models as sql_models
 
35
from ceilometer.tests import base as tests_base
35
36
from ceilometer.tests import db as tests_db
36
37
from ceilometer.tests.storage import test_storage_scenarios as scenarios
37
38
 
172
173
        self.assertTrue(repr.repr(ev))
173
174
 
174
175
 
175
 
class ModelTest(tests_db.TestBase):
 
176
class ModelTest(tests_base.BaseTestCase):
176
177
    database_connection = 'mysql://localhost'
177
178
 
178
179
    def test_model_table_args(self):
 
180
        self.CONF = self.useFixture(config.Config()).conf
 
181
        self.CONF.set_override('connection', self.database_connection,
 
182
                               group='database')
179
183
        self.assertIsNotNone(sql_models.table_args())
180
184
 
181
185
 
182
186
class RelationshipTest(scenarios.DBTestBase):
183
 
    database_connection = 'mysql://localhost'
 
187
    # Note: Do not derive from SQLAlchemyEngineTestBase, since we
 
188
    # don't want to automatically inherit all the Meter setup.
 
189
    database_connection = 'sqlite://'
184
190
 
185
191
    def test_clear_metering_data_meta_tables(self):
186
192
        timeutils.utcnow.override_time = datetime.datetime(2012, 7, 2, 10, 45)
187
193
        self.conn.clear_expired_metering_data(3 * 60)
188
194
 
189
 
        session = sqlalchemy_session.get_session()
 
195
        session = self.conn._get_db_session()
190
196
        meta_tables = [sql_models.MetaText, sql_models.MetaFloat,
191
197
                       sql_models.MetaBigInt, sql_models.MetaBool]
192
198
        for table in meta_tables:
193
199
            self.assertEqual(session.query(table)
194
200
                .filter(~table.id.in_(
195
 
                    session.query(sql_models.Meter.id)
196
 
                        .group_by(sql_models.Meter.id)
 
201
                    session.query(sql_models.Sample.id)
 
202
                        .group_by(sql_models.Sample.id)
197
203
                        )).count(), 0)
198
204
 
199
205
    def test_clear_metering_data_associations(self):
200
206
        timeutils.utcnow.override_time = datetime.datetime(2012, 7, 2, 10, 45)
201
207
        self.conn.clear_expired_metering_data(3 * 60)
202
208
 
203
 
        session = sqlalchemy_session.get_session()
 
209
        session = self.conn._get_db_session()
204
210
        self.assertEqual(session.query(sql_models.sourceassoc)
205
 
            .filter(~sql_models.sourceassoc.c.meter_id.in_(
206
 
                session.query(sql_models.Meter.id)
207
 
                    .group_by(sql_models.Meter.id)
 
211
            .filter(~sql_models.sourceassoc.c.sample_id.in_(
 
212
                session.query(sql_models.Sample.id)
 
213
                    .group_by(sql_models.Sample.id)
208
214
                    )).count(), 0)
209
215
        self.assertEqual(session.query(sql_models.sourceassoc)
210
216
            .filter(~sql_models.sourceassoc.c.project_id.in_(