29
29
from mock import patch
31
import ceilometer.openstack.common.db.sqlalchemy.session as sqlalchemy_session
32
from ceilometer.openstack.common import timeutils
31
33
from ceilometer.storage import models
32
34
from ceilometer.storage.sqlalchemy import models as sql_models
33
35
from ceilometer.tests import db as tests_db
34
from ceilometer import utils
36
from ceilometer.tests.storage import test_storage_scenarios as scenarios
37
39
class EventTestBase(tests_db.TestBase):
144
146
self.assertIsNone(trait.t_int)
145
147
self.assertIsNone(trait.t_string)
146
148
self.assertIsNone(trait.t_float)
147
self.assertEqual(trait.t_datetime, utils.dt_to_decimal(now))
149
self.assertEqual(trait.t_datetime, now)
148
150
self.assertIsNotNone(trait.trait_type.desc)
150
152
def test_bad_event(self):
176
178
def test_model_table_args(self):
177
179
self.assertIsNotNone(sql_models.table_args())
182
class RelationshipTest(scenarios.DBTestBase):
183
database_connection = 'mysql://localhost'
185
def test_clear_metering_data_meta_tables(self):
186
timeutils.utcnow.override_time = datetime.datetime(2012, 7, 2, 10, 45)
187
self.conn.clear_expired_metering_data(3 * 60)
189
session = sqlalchemy_session.get_session()
190
meta_tables = [sql_models.MetaText, sql_models.MetaFloat,
191
sql_models.MetaBigInt, sql_models.MetaBool]
192
for table in meta_tables:
193
self.assertEqual(session.query(table)
194
.filter(~table.id.in_(
195
session.query(sql_models.Meter.id)
196
.group_by(sql_models.Meter.id)
199
def test_clear_metering_data_associations(self):
200
timeutils.utcnow.override_time = datetime.datetime(2012, 7, 2, 10, 45)
201
self.conn.clear_expired_metering_data(3 * 60)
203
session = sqlalchemy_session.get_session()
204
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)
209
self.assertEqual(session.query(sql_models.sourceassoc)
210
.filter(~sql_models.sourceassoc.c.project_id.in_(
211
session.query(sql_models.Project.id)
212
.group_by(sql_models.Project.id)
214
self.assertEqual(session.query(sql_models.sourceassoc)
215
.filter(~sql_models.sourceassoc.c.user_id.in_(
216
session.query(sql_models.User.id)
217
.group_by(sql_models.User.id)