openstack/keystone
Code Issues Proposed changes

Merge "add internal delete notification for endpoint"

Browse Source
This commit is contained in:
Jenkins 2014-07-25 10:23:56 +00:00 committed by Gerrit Code Review
commit 2d3d00e29c
2 changed files with 10 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
keystone/catalog/core.py
View File

@ -26,6 +26,7 @@ from keystone.common import manager
from keystone import config from keystone import config
from keystone import exception from keystone import exception
from keystone.i18n import _ from keystone.i18n import _
from keystone import notifications
from keystone.openstack.common import log from keystone.openstack.common import log
@ -70,6 +71,7 @@ class Manager(manager.Manager):
dynamically calls the backend. dynamically calls the backend.
""" """
_ENDPOINT = 'endpoint'
def __init__(self): def __init__(self):
super(Manager, self).__init__(CONF.catalog.driver) super(Manager, self).__init__(CONF.catalog.driver)
@ -145,6 +147,7 @@ class Manager(manager.Manager):
service_id = endpoint_ref.get('service_id') service_id = endpoint_ref.get('service_id')
raise exception.ServiceNotFound(service_id=service_id) raise exception.ServiceNotFound(service_id=service_id)
@notifications.deleted(_ENDPOINT, public=False)
def delete_endpoint(self, endpoint_id): def delete_endpoint(self, endpoint_id):
try: try:
ret = self.driver.delete_endpoint(endpoint_id) ret = self.driver.delete_endpoint(endpoint_id)

7
keystone/tests/test_notifications.py
View File

@ -308,6 +308,13 @@ class NotificationsForEntities(test_v3.RestfulTestCase):
self.assignment_api.delete_domain(domain_ref['id']) self.assignment_api.delete_domain(domain_ref['id'])
self._assertLastNotify(domain_ref['id'], DELETED_OPERATION, 'domain') self._assertLastNotify(domain_ref['id'], DELETED_OPERATION, 'domain')
def test_delete_endpoint(self):
endpoint_ref = self.new_endpoint_ref(service_id=self.service_id)
self.catalog_api.create_endpoint(endpoint_ref['id'], endpoint_ref)
self.catalog_api.delete_endpoint(endpoint_ref['id'])
self._assertNotifySent(endpoint_ref['id'], DELETED_OPERATION,
'endpoint', public=False)
def test_disable_domain(self): def test_disable_domain(self):
domain_ref = self.new_domain_ref() domain_ref = self.new_domain_ref()
self.assignment_api.create_domain(domain_ref['id'], domain_ref) self.assignment_api.create_domain(domain_ref['id'], domain_ref)