openstack/keystone
Code Issues Proposed changes

Fixes LP #954089 - Service list templated catalog

Browse Source 
* Adds missing test cases for the TemplatedCatalog
* Adds a base CatalogTest that different backends
  can use
* Updates kvs.Catalog to raise ServiceNotFound where
  appropriate
* Updates the tests.test_keystoneclient_sql to actually
  test the SQL catalog backend
* Removes old test for incorrect endpoints listing
* Removes the keystone.catalog.core.Driver.service_exists
  method since it was only implemented in the SQL driver
  and wasn't required now that get_service and delete_service
  properly raise ServiceNotFound exception.

Change-Id: I35690cc147e56007be27bacf94eeff360e727e5d
This commit is contained in:
Jay Pipes 2012-03-13 17:30:07 -04:00
parent 88ac1edec0
commit 193374af38
12 changed files with 171 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
keystone/catalog/backends/kvs.py
View File

@ -16,6 +16,7 @@
from keystone import catalog
from keystone import exception
from keystone.common import kvs
@ -25,7 +26,10 @@ class Catalog(kvs.Base, catalog.Driver):
return self.db.get('catalog-%s-%s' % (tenant_id, user_id))
def get_service(self, service_id):
return self.db.get('service-%s' % service_id)
res = self.db.get('service-%s' % service_id)
if not res:
raise exception.ServiceNotFound(service_id=service_id)
return res
def list_services(self):
return self.db.get('service_list', [])
@ -42,6 +46,9 @@ class Catalog(kvs.Base, catalog.Driver):
return service
def delete_service(self, service_id):
if not self.db.get('service-%s' % service_id):
raise exception.ServiceNotFound(service_id=service_id)
self.db.delete('service-%s' % service_id)
service_list = set(self.db.get('service_list', []))
service_list.remove(service_id)

10
keystone/catalog/backends/sql.py
View File

@ -90,11 +90,15 @@ class Catalog(sql.Base, catalog.Driver):
def get_service(self, service_id):
session = self.get_session()
service_ref = session.query(Service).filter_by(id=service_id).first()
if not service_ref:
raise exception.ServiceNotFound(service_id=service_id)
return service_ref.to_dict()
def delete_service(self, service_id):
session = self.get_session()
service_ref = session.query(Service).filter_by(id=service_id).first()
if not service_ref:
raise exception.ServiceNotFound(service_id=service_id)
with session.begin():
session.delete(service_ref)
session.flush()
@ -107,12 +111,6 @@ class Catalog(sql.Base, catalog.Driver):
session.flush()
return service.to_dict()
def service_exists(self, service_id):
session = self.get_session()
if not session.query(Service).filter_by(id=service_id).first():
return False
return True
# Endpoints
def create_endpoint(self, endpoint_id, endpoint_ref):
session = self.get_session()

3
keystone/catalog/backends/templated.py
View File

@ -52,6 +52,9 @@ def parse_templates(template_lines):
return o
# TODO(jaypipes): should be templated.Catalog,
# not templated.TemplatedCatalog to be consistent with
# other catalog backends
class TemplatedCatalog(kvs.Catalog):
"""A backend that generates endpoints for the Catalog based on templates.

12
keystone/catalog/core.py
View File

@ -69,14 +69,6 @@ class Driver(object):
def create_service(self, service_id, service_ref):
raise exception.NotImplemented()
def service_exists(self, service_id):
"""Query existence of a service by id.
Returns: True if the service exists or False.
"""
raise exception.NotImplemented()
def create_endpoint(self, endpoint_id, endpoint_ref):
raise exception.NotImplemented()
@ -176,7 +168,9 @@ class EndpointController(wsgi.Application):
endpoint_ref['id'] = endpoint_id
service_id = endpoint_ref['service_id']
if not self.catalog_api.service_exists(context, service_id):
try:
service = self.catalog_api.get_service(context, service_id)
except exception.ServiceNotFound:
msg = 'No service exists with id %s' % service_id
raise webob.exc.HTTPBadRequest(msg)

4
keystone/exception.py
View File

@ -77,3 +77,7 @@ class NotImplemented(Error):
class TokenNotFound(NotFound):
"""Could not find token: %(token_id)s"""
class ServiceNotFound(NotFound):
"""Could not find service: %(service_id)s"""

69
keystone/test.py
View File

@ -121,6 +121,7 @@ class TestCase(unittest.TestCase):
self._paths = []
self._memo = {}
self._overrides = []
self._group_overrides = {}
def setUp(self):
super(TestCase, self).setUp()
@ -146,15 +147,26 @@ class TestCase(unittest.TestCase):
kvs.INMEMDB.clear()
self.reset_opts()
def opt_in_group(self, group, **kw):
for k, v in kw.iteritems():
CONF.set_override(k, v, group)
if group not in self._group_overrides:
self._group_overrides[group] = []
self._group_overrides[group].append(k)
def opt(self, **kw):
for k, v in kw.iteritems():
CONF.set_override(k, v)
self._overrides.append(k)
def reset_opts(self):
for group, opt_list in self._group_overrides.iteritems():
for k in opt_list:
CONF.set_override(k, None, group)
for k in self._overrides:
CONF.set_override(k, None)
self._overrides = []
self._group_overrides = {}
CONF.reset()
def load_backends(self):
@ -172,34 +184,41 @@ class TestCase(unittest.TestCase):
"""
# TODO(termie): doing something from json, probably based on Django's
# loaddata will be much preferred.
for tenant in fixtures.TENANTS:
rv = self.identity_api.create_tenant(tenant['id'], tenant)
setattr(self, 'tenant_%s' % tenant['id'], rv)
if hasattr(self, 'catalog_api'):
for service in fixtures.SERVICES:
rv = self.catalog_api.create_service(service['id'], service)
setattr(self, 'service_%s' % service['id'], rv)
for user in fixtures.USERS:
user_copy = user.copy()
tenants = user_copy.pop('tenants')
rv = self.identity_api.create_user(user['id'], user_copy.copy())
for tenant_id in tenants:
self.identity_api.add_user_to_tenant(tenant_id, user['id'])
setattr(self, 'user_%s' % user['id'], user_copy)
if hasattr(self, 'identity_api'):
for tenant in fixtures.TENANTS:
rv = self.identity_api.create_tenant(tenant['id'], tenant)
setattr(self, 'tenant_%s' % tenant['id'], rv)
for role in fixtures.ROLES:
rv = self.identity_api.create_role(role['id'], role)
setattr(self, 'role_%s' % role['id'], rv)
for user in fixtures.USERS:
user_copy = user.copy()
tenants = user_copy.pop('tenants')
rv = self.identity_api.create_user(user['id'],
user_copy.copy())
for tenant_id in tenants:
self.identity_api.add_user_to_tenant(tenant_id, user['id'])
setattr(self, 'user_%s' % user['id'], user_copy)
for metadata in fixtures.METADATA:
metadata_ref = metadata.copy()
# TODO(termie): these will probably end up in the model anyway,
# so this may be futile
del metadata_ref['user_id']
del metadata_ref['tenant_id']
rv = self.identity_api.create_metadata(metadata['user_id'],
metadata['tenant_id'],
metadata_ref)
setattr(self,
'metadata_%s%s' % (metadata['user_id'],
metadata['tenant_id']), rv)
for role in fixtures.ROLES:
rv = self.identity_api.create_role(role['id'], role)
setattr(self, 'role_%s' % role['id'], rv)
for metadata in fixtures.METADATA:
metadata_ref = metadata.copy()
# TODO(termie): these will probably end up in the model anyway,
# so this may be futile
del metadata_ref['user_id']
del metadata_ref['tenant_id']
rv = self.identity_api.create_metadata(metadata['user_id'],
metadata['tenant_id'],
metadata_ref)
setattr(self,
'metadata_%s%s' % (metadata['user_id'],
metadata['tenant_id']), rv)
def _paste_config(self, config):
if not config.startswith('config:'):

3
tests/backend_sql.conf
View File

@ -13,3 +13,6 @@ driver = keystone.token.backends.sql.Token
[ec2]
driver = keystone.contrib.ec2.backends.sql.Ec2
[catalog]
driver = keystone.catalog.backends.sql.Catalog

21
tests/default_fixtures.py
View File

@ -35,3 +35,24 @@ ROLES = [
{'id': 'keystone_admin', 'name': 'Keystone Admin'},
{'id': 'useless', 'name': 'Useless'},
]
SERVICES = [
{
'id': 'COMPUTE_ID',
'type': 'compute',
'name': 'Nova',
'description': 'OpenStack Compute service'
},
{
'id': 'IDENTITY_ID',
'type': 'identity',
'name': 'Keystone',
'description': 'OpenStack Identity service'
},
{
'id': 'IMAGE_ID',
'type': 'image',
'name': 'Glance',
'description': 'OpenStack Image service'
},
]

20
tests/test_backend.py
View File

@ -302,3 +302,23 @@ class TokenTests(object):
self.assertDictEquals(data_ref, data)
new_data_ref = self.token_api.get_token(token_id)
self.assertEqual(data_ref, new_data_ref)
class CatalogTests(object):
def test_service_crud(self):
new_service = {'id': 'MY_SERVICE', 'type': 'myservice',
'name': 'My Service', 'description': 'My description'}
res = self.catalog_api.create_service(new_service['id'], new_service)
self.assertDictEquals(res, new_service)
service_id = new_service['id']
self.catalog_api.delete_service(service_id)
self.assertRaises(exception.ServiceNotFound,
self.catalog_api.delete_service, service_id)
self.assertRaises(exception.ServiceNotFound,
self.catalog_api.get_service, service_id)
def test_service_list(self):
services = self.catalog_api.list_services()
self.assertEqual(3, len(services))

7
tests/test_backend_kvs.py
View File

@ -35,13 +35,14 @@ class KvsToken(test.TestCase, test_backend.TokenTests):
self.token_api = token_kvs.Token(db={})
class KvsCatalog(test.TestCase):
class KvsCatalog(test.TestCase, test_backend.CatalogTests):
def setUp(self):
super(KvsCatalog, self).setUp()
self.catalog_api = catalog_kvs.Catalog(db={})
self._load_fixtures()
self.load_fixtures(default_fixtures)
self._load_fake_catalog()
def _load_fixtures(self):
def _load_fake_catalog(self):
self.catalog_foobar = self.catalog_api._create_catalog(
'foo', 'bar',
{'RegionFoo': {'service_bar': {'foo': 'bar'}}})

57
tests/test_backend_templated.py Normal file
View File

@ -0,0 +1,57 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 OpenStack LLC
#
# 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.
import os
from keystone import test
from keystone.catalog.backends import templated as catalog_templated
import test_backend
import default_fixtures
DEFAULT_CATALOG_TEMPLATES = os.path.abspath(os.path.join(
os.path.dirname(__file__),
'default_catalog.templates'))
class TestTemplatedCatalog(test.TestCase, test_backend.CatalogTests):
DEFAULT_FIXTURE = {
'RegionOne': {
'compute': {
'adminURL': 'http://localhost:8774/v1.1/bar',
'publicURL': 'http://localhost:8774/v1.1/bar',
'internalURL': 'http://localhost:8774/v1.1/bar',
'name': "'Compute Service'"
},
'identity': {
'adminURL': 'http://localhost:35357/v2.0',
'publicURL': 'http://localhost:5000/v2.0',
'internalURL': 'http://localhost:35357/v2.0',
'name': "'Identity Service'"
}
}
}
def setUp(self):
super(TestTemplatedCatalog, self).setUp()
self.opt_in_group('catalog', template_file=DEFAULT_CATALOG_TEMPLATES)
self.catalog_api = catalog_templated.TemplatedCatalog()
self.load_fixtures(default_fixtures)
def test_get_catalog(self):
catalog_ref = self.catalog_api.get_catalog('foo', 'bar')
self.assertDictEquals(catalog_ref, self.DEFAULT_FIXTURE)

8
tests/test_keystoneclient.py
View File

@ -175,14 +175,6 @@ class KeystoneClientTests(object):
self.get_client,
user_ref)
# TODO(termie): I'm not really sure that this is testing much
def test_endpoints(self):
raise nose.exc.SkipTest('Not implemented due to bug 933555')
client = self.get_client(admin=True)
token = client.auth_token
endpoints = client.tokens.endpoints(token=token)
# FIXME(ja): this test should require the "keystone:admin" roled
# (probably the role set via --keystone_admin_role flag)
# FIXME(ja): add a test that admin endpoint is only sent to admin user