207
208
"\'float\', \'string\', \'datetime\']",
208
209
resp.json['error_message']['faultstring'])
211
def test_get_events_filter_operator_invalid_type(self):
212
resp = self.get_json(self.PATH, headers=headers,
213
q=[{'field': 'trait_A',
214
'value': 'my_Foo_text',
217
self.assertEqual(400, resp.status_code)
218
self.assertEqual("operator whats-up is not supported. the "
219
"supported operators are: (\'lt\', \'le\', "
220
"\'eq\', \'ne\', \'ge\', \'gt\')",
221
resp.json['error_message']['faultstring'])
210
223
def test_get_events_filter_text_trait(self):
211
224
data = self.get_json(self.PATH, headers=headers,
212
225
q=[{'field': 'trait_A',
439
452
'type': 'integer',
456
class EventRestrictionTestBase(v2.FunctionalTest,
457
tests_db.MixinTestsWithBackendScenarios):
460
super(EventRestrictionTestBase, self).setUp()
461
self.CONF.set_override('default_api_return_limit', 10, group='api')
462
self._generate_models()
464
def _generate_models(self):
467
self.s_time = datetime.datetime(2013, 12, 31, 5, 0)
468
self.trait_time = datetime.datetime(2013, 12, 31, 5, 0)
470
trait_models = [models.Trait(name, type, value)
471
for name, type, value in [
472
('trait_A', models.Trait.TEXT_TYPE,
474
('trait_B', models.Trait.INT_TYPE,
476
('trait_C', models.Trait.FLOAT_TYPE,
477
float(base) + 0.123456),
478
('trait_D', models.Trait.DATETIME_TYPE,
482
models.Event(message_id=str(uuid.uuid4()),
483
event_type='foo.bar',
484
generated=self.trait_time,
486
raw={'status': {'nested': 'started'}}))
487
self.trait_time += datetime.timedelta(seconds=1)
488
self.event_conn.record_events(event_models)
491
class TestEventRestriction(EventRestrictionTestBase):
493
def test_get_limit(self):
494
data = self.get_json('/events?limit=1', headers=headers)
495
self.assertEqual(1, len(data))
497
def test_get_limit_negative(self):
498
self.assertRaises(webtest.app.AppError,
499
self.get_json, '/events?limit=-2', headers=headers)
501
def test_get_limit_bigger(self):
502
data = self.get_json('/events?limit=100', headers=headers)
503
self.assertEqual(20, len(data))
505
def test_get_default_limit(self):
506
data = self.get_json('/events', headers=headers)
507
self.assertEqual(10, len(data))