openstack
/
keystone
Code Issues Proposed changes

Add unit tests for disabled endpoints in catalog 

There were no tests that show how disabled endpoints are dealt
with in the catalog backend or when getting or validating a
token.

Change-Id: I42ba07e2e7a9c95469b3f2a741a809756cda9361
Related-Bug: #1273867
This commit is contained in:
Brant Knudson 2014-03-01 14:54:51 -06:00
parent b870d88c7d
commit 46070a6562
6 changed files with 305 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

46
keystone/tests/test_associate_project_endpoint_extension.py
View File

@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import uuid
from keystone.common.sql import migration_helpers
@ -458,3 +459,48 @@ class AssociateProjectEndpointFilterTokenRequestTestCase(TestExtensionCase):
ep_filter_assoc=1)
self.assertEqual(r.result['token']['project']['id'],
self.project['id'])
def test_disabled_endpoint(self):
"""The catalog will contain both enabled and disabled endpoints."""
# FIXME(blk-u): disabled endpoints should not be included in the
# catalog, see bug 1273867
# Add an enabled endpoint to the default project
self.put('/OS-EP-FILTER/projects/%(project_id)s'
'/endpoints/%(endpoint_id)s' % {
'project_id': self.project['id'],
'endpoint_id': self.endpoint_id},
expected_status=204)
# Add a disabled endpoint to the default project.
# Create a disabled endpoint that's like the enabled one.
disabled_endpoint_ref = copy.copy(self.endpoint)
disabled_endpoint_id = uuid.uuid4().hex
disabled_endpoint_ref.update({
'id': disabled_endpoint_id,
'enabled': False,
'interface': 'internal'
})
self.catalog_api.create_endpoint(disabled_endpoint_id,
disabled_endpoint_ref)
self.put('/OS-EP-FILTER/projects/%(project_id)s'
'/endpoints/%(endpoint_id)s' % {
'project_id': self.project['id'],
'endpoint_id': disabled_endpoint_id},
expected_status=204)
# Authenticate to get token with catalog
auth_data = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'],
project_id=self.project['id'])
r = self.post('/auth/tokens', body=auth_data)
endpoints = r.result['token']['catalog'][0]['endpoints']
endpoint_ids = [ep['id'] for ep in endpoints]
self.assertEqual(2, len(endpoint_ids))
self.assertIn(self.endpoint_id, endpoint_ids)
self.assertIn(disabled_endpoint_id, endpoint_ids)

108
keystone/tests/test_auth.py
View File

@ -958,6 +958,114 @@ class TokenExpirationTest(AuthTest):
self._maintain_token_expiration()
class AuthCatalog(AuthTest):
"""Tests for the catalog provided in the auth response."""
def config(self, config_files):
# We need to use a backend that supports disabled endpoints, like the
# SQL backend.
config_files.append(tests.dirs.tests('backend_sql.conf'))
super(AuthCatalog, self).config(config_files)
def _create_endpoints(self):
def create_endpoint(service_id, region, **kwargs):
id_ = uuid.uuid4().hex
ref = {
'id': id_,
'interface': 'public',
'region': region,
'service_id': service_id,
'url': 'http://localhost/%s' % uuid.uuid4().hex,
}
ref.update(kwargs)
self.catalog_api.create_endpoint(id_, ref)
return ref
# Create a service for use with the endpoints.
service_id = uuid.uuid4().hex
service_ref = {
'id': service_id,
'name': uuid.uuid4().hex,
'type': uuid.uuid4().hex,
}
self.catalog_api.create_service(service_id, service_ref)
region = uuid.uuid4().hex
# Create endpoints
enabled_endpoint_ref = create_endpoint(service_id, region)
disabled_endpoint_ref = create_endpoint(
service_id, region, enabled=False, interface='internal')
return enabled_endpoint_ref, disabled_endpoint_ref
def test_auth_catalog_disabled(self):
"""When authenticate, get back a catalog that includes both enabled and
disabled endpoints.
"""
# FIXME(blk-u): disabled endpoints should not be included in the
# catalog, see bug 1273867
enabled_endpoint_ref, disabled_endpoint_ref = self._create_endpoints()
# Authenticate
body_dict = _build_user_auth(
username='FOO',
password='foo2',
tenant_name="BAR")
token = self.controller.authenticate({}, body_dict)
# Check the catalog
endpoint = token['access']['serviceCatalog'][0]['endpoints'][0]
exp_endpoint = {
'id': enabled_endpoint_ref['id'],
'internalURL': disabled_endpoint_ref['url'],
'publicURL': enabled_endpoint_ref['url'],
'region': enabled_endpoint_ref['region'],
}
self.assertEqual(exp_endpoint, endpoint)
def test_validate_catalog_disabled(self):
"""When validate, get back a catalog that includes both enabled and
disabled endpoints.
"""
# FIXME(blk-u): disabled endpoints should not be included in the
# catalog, see bug 1273867
enabled_endpoint_ref, disabled_endpoint_ref = self._create_endpoints()
# Authenticate
body_dict = _build_user_auth(
username='FOO',
password='foo2',
tenant_name="BAR")
token = self.controller.authenticate({}, body_dict)
# Validate
token_id = token['access']['token']['id']
validate_ref = self.controller.validate_token(
dict(is_admin=True, query_string={}),
token_id=token_id)
# Check the catalog
endpoint = validate_ref['access']['serviceCatalog'][0]['endpoints'][0]
exp_endpoint = {
'id': enabled_endpoint_ref['id'],
'internalURL': disabled_endpoint_ref['url'],
'publicURL': enabled_endpoint_ref['url'],
'region': enabled_endpoint_ref['region'],
}
self.assertEqual(exp_endpoint, endpoint)
class NonDefaultAuthTest(tests.TestCase):
def test_add_non_default_auth_method(self):

81
keystone/tests/test_backend.py
View File

@ -3676,6 +3676,87 @@ class CatalogTests(object):
}
self.catalog_api.create_endpoint(endpoint['id'], endpoint.copy())
def _create_endpoints(self):
# Creates a service and 2 endpoints for the service in the same region.
# The 'public' interface is enabled and the 'internal' interface is
# disabled.
def create_endpoint(service_id, region, **kwargs):
id_ = uuid.uuid4().hex
ref = {
'id': id_,
'interface': 'public',
'region': region,
'service_id': service_id,
'url': 'http://localhost/%s' % uuid.uuid4().hex,
}
ref.update(kwargs)
self.catalog_api.create_endpoint(id_, ref)
return ref
# Create a service for use with the endpoints.
service_id = uuid.uuid4().hex
service_ref = {
'id': service_id,
'name': uuid.uuid4().hex,
'type': uuid.uuid4().hex,
}
self.catalog_api.create_service(service_id, service_ref)
region = uuid.uuid4().hex
# Create endpoints
enabled_endpoint_ref = create_endpoint(service_id, region)
disabled_endpoint_ref = create_endpoint(
service_id, region, enabled=False, interface='internal')
return service_ref, enabled_endpoint_ref, disabled_endpoint_ref
def test_get_catalog_endpoint_disabled(self):
"""Get back both enabled and disabled endpoints when get the v2
catalog.
"""
# FIXME(blk-u): disabled endpoints should not be included in the
# catalog, see bug 1273867
service_ref, enabled_endpoint_ref, disabled_endpoint_ref = (
self._create_endpoints())
user_id = uuid.uuid4().hex
project_id = uuid.uuid4().hex
catalog = self.catalog_api.get_catalog(user_id, project_id)
exp_entry = {
'id': enabled_endpoint_ref['id'],
'internalURL': disabled_endpoint_ref['url'],
'name': service_ref['name'],
'publicURL': enabled_endpoint_ref['url'],
}
region = enabled_endpoint_ref['region']
self.assertEqual(exp_entry, catalog[region][service_ref['type']])
def test_get_v3_catalog_endpoint_disabled(self):
"""Get back both enabled and disabled endpoints when get the v3
catalog.
"""
# FIXME(blk-u): disabled endpoints should not be included in the
# catalog, see bug 1273867
dummy_service_ref, enabled_endpoint_ref, disabled_endpoint_ref = (
self._create_endpoints())
user_id = uuid.uuid4().hex
project_id = uuid.uuid4().hex
catalog = self.catalog_api.get_v3_catalog(user_id, project_id)
endpoint_ids = [x['id'] for x in catalog[0]['endpoints']]
self.assertIn(enabled_endpoint_ref['id'], endpoint_ids)
self.assertIn(disabled_endpoint_ref['id'], endpoint_ids)
self.assertEqual(2, len(endpoint_ids))
class PolicyTests(object):
def _new_policy_ref(self):

12
keystone/tests/test_backend_kvs.py
View File

@ -192,6 +192,18 @@ class KvsCatalog(tests.TestCase, test_backend.CatalogTests):
catalog_ref = self.catalog_api.get_catalog('foo', 'bar')
self.assertDictEqual(catalog_ref, self.catalog_foobar)
def test_get_catalog_endpoint_disabled(self):
# This test doesn't apply to KVS because with the KVS backend the
# application creates the catalog (including the endpoints) for each
# user and project. Whether endpoints are enabled or disabled isn't
# a consideration.
f = super(KvsCatalog, self).test_get_catalog_endpoint_disabled
self.assertRaises(exception.NotFound, f)
def test_get_v3_catalog_endpoint_disabled(self):
f = super(KvsCatalog, self).test_get_v3_catalog_endpoint_disabled
self.assertRaises(exception.NotImplemented, f)
class KvsTokenCacheInvalidation(tests.TestCase,
test_backend.TokenCacheInvalidation):

8
keystone/tests/test_backend_templated.py
View File

@ -64,3 +64,11 @@ class TestTemplatedCatalog(tests.TestCase, test_backend.CatalogTests):
self.catalog_api.get_catalog,
'fake-user',
'fake-tenant')
def test_get_catalog_endpoint_disabled(self):
self.skipTest("Templated backend doesn't have disabled endpoints")
def test_get_v3_catalog_endpoint_disabled(self):
f = (super(TestTemplatedCatalog, self).
test_get_v3_catalog_endpoint_disabled)
self.assertRaises(exception.NotImplemented, f)

50
keystone/tests/test_v3_auth.py
View File

@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import copy
import datetime
import json
import uuid
@ -1619,6 +1620,42 @@ class TestAuthJSON(test_v3.RestfulTestCase):
self.assertEqual(r.result['token']['project']['id'],
self.project['id'])
def _check_disabled_endpoint_result(self, catalog, disabled_endpoint_id):
endpoints = catalog[0]['endpoints']
endpoint_ids = [ep['id'] for ep in endpoints]
self.assertEqual(2, len(endpoint_ids))
self.assertIn(self.endpoint_id, endpoint_ids)
self.assertIn(disabled_endpoint_id, endpoint_ids)
def test_auth_catalog_disabled_endpoint(self):
"""When authenticate, get back a catalog that includes both enabled
and disabled endpoints.
"""
# FIXME(blk-u): disabled endpoints should not be included in the
# catalog, see bug 1273867
# Create a disabled endpoint that's like the enabled one.
disabled_endpoint_ref = copy.copy(self.endpoint)
disabled_endpoint_id = uuid.uuid4().hex
disabled_endpoint_ref.update({
'id': disabled_endpoint_id,
'enabled': False,
'interface': 'internal'
})
self.catalog_api.create_endpoint(disabled_endpoint_id,
disabled_endpoint_ref)
auth_data = self.build_authentication_request(
user_id=self.user['id'],
password=self.user['password'],
project_id=self.project['id'])
r = self.post('/auth/tokens', body=auth_data)
self._check_disabled_endpoint_result(r.result['token']['catalog'],
disabled_endpoint_id)
def test_project_id_scoped_token_with_user_id_401(self):
project_id = uuid.uuid4().hex
project = self.new_project_ref(domain_id=self.domain_id)
@ -2197,6 +2234,19 @@ class TestAuthJSON(test_v3.RestfulTestCase):
class TestAuthXML(TestAuthJSON):
content_type = 'xml'
def _check_disabled_endpoint_result(self, catalog, disabled_endpoint_id):
# FIXME(blk-u): As far as I can tell the catalog in the XML result is
# broken. Looks like it includes only one endpoint or the other, and
# which one is included is random.
endpoint = catalog['service']['endpoint']
if endpoint['id'] == self.endpoint_id:
pass
elif endpoint['id'] == disabled_endpoint_id:
pass
else:
self.fail("Didn't find either enabled or disabled endpoint!")
class TestTrustOptional(test_v3.RestfulTestCase):
def setUp(self, *args, **kwargs):