openstack/senlin
Code Issues Proposed changes

Unit test for events in service engine

Browse Source 
This patch adds unit test cases for events in service engine.

Change-Id: I3942f6fcbabbcaf6af9971fb28c5c487fe5f4608
This commit is contained in:
tengqm
2015-08-13 09:00:28 -04:00
parent 621793e202
commit 3e04607360
5 changed files with 210 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
senlin/common/consts.py
View File

@@ -147,9 +147,11 @@ WEBHOOK_OBJ_TYPES = (
EVENT_ATTRS = (
EVENT_TIMESTAMP, EVENT_OBJ_ID, EVENT_OBJ_NAME, EVENT_OBJ_TYPE,
EVENT_USER, EVENT_ACTION, EVENT_STATUS, EVENT_STATUS_REASON,
EVENT_LEVEL,
) = (
'timestamp', 'obj_id', 'obj_name', 'obj_type',
'user', 'action', 'status', 'status_reason',
'level',
)
ACTION_ATTRS = (

1
senlin/db/sqlalchemy/api.py
View File

@@ -1016,6 +1016,7 @@ def _event_filter_paginate_query(context, query, filters=None,
sort_key_map = {
consts.EVENT_TIMESTAMP: models.Event.timestamp.key,
consts.EVENT_LEVEL: models.Event.level.key,
consts.EVENT_OBJ_TYPE: models.Event.obj_type.key,
consts.EVENT_OBJ_NAME: models.Event.obj_name.key,
consts.EVENT_USER: models.Event.user.key,

6
senlin/engine/event.py
View File

@@ -51,6 +51,12 @@ class Event(object):
self.cluster_id = kwargs.get('cluster_id', None)
self.metadata = kwargs.get('metadata', {})
ctx = kwargs.get('context', None)
if ctx is not None:
self.user = ctx.user
self.project = ctx.project
self.domain = ctx.domain
# entity not None implies an initial creation of event object,
# not a deserialization, so we try make an inference here
if entity is not None:

19
senlin/engine/service.py
View File

@@ -1496,8 +1496,9 @@ class EngineService(service.Service):
resource_id=db_action.id)
return None
@request_context
def event_find(self, context, identity, show_deleted=False):
'''Find a event with the given identity (could be name or ID).'''
'''Find a event with the given identity (could ID or short ID).'''
if uuidutils.is_uuid_like(identity):
event = db_api.event_get(context, identity)
if not event:
@@ -1506,7 +1507,7 @@ class EngineService(service.Service):
event = db_api.event_get_by_short_id(context, identity)
if not event:
raise exception.EventNotFound(action=identity)
raise exception.EventNotFound(event=identity)
return event
@@ -1514,14 +1515,14 @@ class EngineService(service.Service):
def event_list(self, context, filters=None, limit=None, marker=None,
sort_keys=None, sort_dir=None, project_safe=True,
show_deleted=False):
all_actions = event_mod.Event.load_all(context, filters=filters,
limit=limit, marker=marker,
sort_keys=sort_keys,
sort_dir=sort_dir,
project_safe=project_safe,
show_deleted=show_deleted)
all_events = event_mod.Event.load_all(context, filters=filters,
limit=limit, marker=marker,
sort_keys=sort_keys,
sort_dir=sort_dir,
project_safe=project_safe,
show_deleted=show_deleted)
results = [action.to_dict() for action in all_actions]
results = [event.to_dict() for event in all_events]
return results
@request_context

191
senlin/tests/unit/engine/service/test_events.py Normal file
View File

@@ -0,0 +1,191 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# 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 oslo_messaging.rpc import dispatcher as rpc
from oslo_utils import timeutils
import six
from senlin.common import exception
from senlin.engine import event as event_mod
from senlin.engine import service
from senlin.tests.unit.common import base
from senlin.tests.unit.common import utils
class EventTest(base.SenlinTestCase):
def setUp(self):
super(EventTest, self).setUp()
self.ctx = utils.dummy_context(project='event_test_project')
self.eng = service.EngineService('host-a', 'topic-a')
self.eng.init_tgm()
def test_event_list(self):
ts = timeutils.utcnow()
e1 = event_mod.Event(ts, 50, status='GOOD', context=self.ctx)
eid1 = e1.store(self.ctx)
e2 = event_mod.Event(ts, 50, status='BAD', context=self.ctx)
eid2 = e2.store(self.ctx)
result = self.eng.event_list(self.ctx)
self.assertIsInstance(result, list)
statuses = [e['status'] for e in result]
ids = [p['id'] for p in result]
self.assertIn(e1.status, statuses)
self.assertIn(e2.status, statuses)
self.assertIn(eid1, ids)
self.assertIn(eid2, ids)
def test_event_list_with_limit_marker(self):
e1 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
e1.store(self.ctx)
e2 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
e2.store(self.ctx)
result = self.eng.event_list(self.ctx, limit=0)
self.assertEqual(0, len(result))
result = self.eng.event_list(self.ctx, limit=1)
self.assertEqual(1, len(result))
result = self.eng.event_list(self.ctx, limit=2)
self.assertEqual(2, len(result))
result = self.eng.event_list(self.ctx, limit=3)
self.assertEqual(2, len(result))
result = self.eng.event_list(self.ctx, marker=e1.id)
self.assertEqual(1, len(result))
result = self.eng.event_list(self.ctx, marker=e2.id)
self.assertEqual(0, len(result))
e3 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
e3.store(self.ctx)
result = self.eng.event_list(self.ctx, limit=1, marker=e1.id)
self.assertEqual(1, len(result))
result = self.eng.event_list(self.ctx, limit=2, marker=e1.id)
self.assertEqual(2, len(result))
def test_event_list_with_sort_keys(self):
e1 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
e1.store(self.ctx)
e2 = event_mod.Event(timeutils.utcnow(), 40, status='GOOD',
context=self.ctx)
e2.store(self.ctx)
e3 = event_mod.Event(timeutils.utcnow(), 60, status='BAD',
context=self.ctx)
e3.store(self.ctx)
# default by timestamp
result = self.eng.event_list(self.ctx)
self.assertEqual(e1.id, result[0]['id'])
self.assertEqual(e2.id, result[1]['id'])
# use level for sorting
result = self.eng.event_list(self.ctx, sort_keys=['level'])
self.assertEqual(e2.id, result[0]['id'])
self.assertEqual(e1.id, result[1]['id'])
# use status for sorting
result = self.eng.event_list(self.ctx, sort_keys=['status'])
self.assertEqual(e3.id, result[2]['id'])
# use level and status for sorting
result = self.eng.event_list(self.ctx,
sort_keys=['status', 'level'])
self.assertEqual(e3.id, result[2]['id'])
self.assertEqual(e2.id, result[0]['id'])
self.assertEqual(e1.id, result[1]['id'])
# unknown keys will be ignored
result = self.eng.event_list(self.ctx, sort_keys=['duang'])
self.assertIsNotNone(result)
def test_event_list_with_sort_dir(self):
e1 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
e1.store(self.ctx)
e2 = event_mod.Event(timeutils.utcnow(), 40, status='GOOD',
context=self.ctx)
e2.store(self.ctx)
e3 = event_mod.Event(timeutils.utcnow(), 60, status='BAD',
context=self.ctx)
e3.store(self.ctx)
# default by timestamp , ascending
result = self.eng.event_list(self.ctx)
self.assertEqual(e1.id, result[0]['id'])
self.assertEqual(e2.id, result[1]['id'])
# sort by created_time, descending
result = self.eng.event_list(self.ctx, sort_dir='desc')
self.assertEqual(e3.id, result[0]['id'])
self.assertEqual(e2.id, result[1]['id'])
# use name for sorting, descending
result = self.eng.event_list(self.ctx, sort_keys=['name'],
sort_dir='desc')
self.assertEqual(e3.id, result[0]['id'])
self.assertEqual(e1.id, result[2]['id'])
# use permission for sorting
ex = self.assertRaises(ValueError,
self.eng.event_list, self.ctx,
sort_dir='Bogus')
self.assertEqual("Unknown sort direction, must be "
"'desc' or 'asc'", six.text_type(ex))
def test_event_list_with_filters(self):
e1 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
e1.store(self.ctx)
e2 = event_mod.Event(timeutils.utcnow(), 40, status='GOOD',
context=self.ctx)
e2.store(self.ctx)
e3 = event_mod.Event(timeutils.utcnow(), 60, status='BAD',
context=self.ctx)
e3.store(self.ctx)
result = self.eng.event_list(self.ctx, filters={'level': 50})
self.assertEqual(1, len(result))
result = self.eng.event_list(self.ctx, filters={'level': 10})
self.assertEqual(0, len(result))
filters = {'status': 'GOOD'}
result = self.eng.event_list(self.ctx, filters=filters)
self.assertEqual(2, len(result))
def test_event_list_empty(self):
result = self.eng.event_list(self.ctx)
self.assertIsInstance(result, list)
self.assertEqual(0, len(result))
def test_event_find(self):
e1 = event_mod.Event(timeutils.utcnow(), 50, status='GOOD',
context=self.ctx)
eid = e1.store(self.ctx)
result = self.eng.event_find(self.ctx, eid)
self.assertIsNotNone(result)
# short id
result = self.eng.event_find(self.ctx, eid[:5])
self.assertIsNotNone(result)
# others
ex = self.assertRaises(rpc.ExpectedException,
self.eng.event_find, self.ctx, 'Bogus')
self.assertEqual(exception.EventNotFound, ex.exc_info[0])