openstack
/
keystone
Code Issues Proposed changes

Rename notification for create/delete grants 

For backward compatiblity, old and new format of event
type will still be sent. The old format is marked as
deprecated and to be removed in the next release.

Co-Authored-By: Lin Hua Cheng <os.lcheng@gmail.com>
DocImpact
Change-Id: Ic19e8e6983e63789b42d8fb9677ffd115a40fc9e
Closes-Bug: #1416767
This commit is contained in:
Steve Martinelli 2015-03-25 02:18:27 -04:00
parent 2c50fb0b69
commit 8695b1adf7
2 changed files with 34 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
keystone/notifications.py
View File

@ -30,6 +30,7 @@ from pycadf import eventfactory
from pycadf import resource
from keystone.i18n import _, _LE
from keystone.openstack.common import versionutils
notifier_opts = [
@ -524,8 +525,10 @@ class CadfRoleAssignmentNotificationWrapper(object):
def __init__(self, operation):
self.action = '%s.%s' % (operation, self.ROLE_ASSIGNMENT)
self.event_type = '%s.%s.%s' % (SERVICE, operation,
self.ROLE_ASSIGNMENT)
self.deprecated_event_type = '%s.%s.%s' % (SERVICE, operation,
self.ROLE_ASSIGNMENT)
self.event_type = '%s.%s.%s' % (SERVICE, self.ROLE_ASSIGNMENT,
operation)
def __call__(self, f):
def wrapper(wrapped_self, role_id, *args, **kwargs):
@ -581,19 +584,30 @@ class CadfRoleAssignmentNotificationWrapper(object):
audit_kwargs['inherited_to_projects'] = inherited
audit_kwargs['role'] = role_id
# For backward compatability, send both old and new event_type.
# Deprecate old format and remove it in the next release.
event_types = [self.deprecated_event_type, self.event_type]
versionutils.deprecated(
as_of=versionutils.deprecated.KILO,
remove_in=+1,
what=('sending duplicate %s notification event type' %
self.deprecated_event_type),
in_favor_of='%s notification event type' % self.event_type)
try:
result = f(wrapped_self, role_id, *args, **kwargs)
except Exception:
_send_audit_notification(self.action, initiator,
taxonomy.OUTCOME_FAILURE,
target, self.event_type,
**audit_kwargs)
for event_type in event_types:
_send_audit_notification(self.action, initiator,
taxonomy.OUTCOME_FAILURE,
target, event_type,
**audit_kwargs)
raise
else:
_send_audit_notification(self.action, initiator,
taxonomy.OUTCOME_SUCCESS,
target, self.event_type,
**audit_kwargs)
for event_type in event_types:
_send_audit_notification(self.action, initiator,
taxonomy.OUTCOME_SUCCESS,
target, event_type,
**audit_kwargs)
return result
return wrapper

13
keystone/tests/unit/common/test_notifications.py
View File

@ -759,13 +759,14 @@ class CadfNotificationsWrapperTestCase(test_v3.RestfulTestCase):
'action': action,
'initiator': initiator,
'event': event,
'event_type': event_type,
'send_notification_called': True}
self._notifications.append(note)
self.useFixture(mockpatch.PatchObject(
notifications, '_send_audit_notification', fake_notify))
def _assert_last_note(self, action, user_id):
def _assert_last_note(self, action, user_id, event_type=None):
self.assertTrue(self._notifications)
note = self._notifications[-1]
self.assertEqual(note['action'], action)
@ -773,6 +774,8 @@ class CadfNotificationsWrapperTestCase(test_v3.RestfulTestCase):
self.assertEqual(initiator.id, user_id)
self.assertEqual(initiator.host.address, self.LOCAL_HOST)
self.assertTrue(note['send_notification_called'])
if event_type:
self.assertEqual(note['event_type'], event_type)
def _assert_event(self, role_id, project=None, domain=None,
user=None, group=None, inherit=False):
@ -857,11 +860,15 @@ class CadfNotificationsWrapperTestCase(test_v3.RestfulTestCase):
user=None, group=None):
self.put(url)
action = "%s.%s" % (CREATED_OPERATION, self.ROLE_ASSIGNMENT)
self._assert_last_note(action, self.user_id)
event_type = '%s.%s.%s' % (notifications.SERVICE,
self.ROLE_ASSIGNMENT, CREATED_OPERATION)
self._assert_last_note(action, self.user_id, event_type)
self._assert_event(role, project, domain, user, group)
self.delete(url)
action = "%s.%s" % (DELETED_OPERATION, self.ROLE_ASSIGNMENT)
self._assert_last_note(action, self.user_id)
event_type = '%s.%s.%s' % (notifications.SERVICE,
self.ROLE_ASSIGNMENT, DELETED_OPERATION)
self._assert_last_note(action, self.user_id, event_type)
self._assert_event(role, project, domain, user, group)
def test_user_project_grant(self):