x
/
tobiko
Code Issues Proposed changes

Add Keystone client API 

Add helper functions to get Keystone client, list services
and list endpoints.

Change-Id: I99abdd94338c7b4579f37a4580076f3c52ca9b7a
This commit is contained in:
Federico Ressi 2019-07-15 15:29:21 +02:00
parent 50efb0691d
commit 225a26de03
7 changed files with 413 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tobiko/openstack/_client.py
View File

@ -18,7 +18,6 @@ import abc
from oslo_log import log
import tobiko
from tobiko.openstack import keystone
LOG = log.getLogger(__name__)
@ -47,6 +46,7 @@ class OpenstackClientFixture(tobiko.SharedFixture):
return client
def get_session(self):
from tobiko.openstack import keystone
return keystone.keystone_session(self.session)
@abc.abstractmethod

11
tobiko/openstack/keystone/__init__.py
View File

@ -13,9 +13,20 @@
# under the License.
from __future__ import absolute_import
from tobiko.openstack.keystone import _client
from tobiko.openstack.keystone import _credentials
from tobiko.openstack.keystone import _session
keystone_client = _client.keystone_client
get_keystone_client = _client.get_keystone_client
find_service = _client.find_service
find_endpoint = _client.find_endpoint
list_endpoints = _client.list_endpoints
list_services = _client.list_services
KeystoneClientFixture = _client.KeystoneClientFixture
KeystoneResourceNotFound = _client.KeystoneResourceNotFound
MultipleKeystoneResourcesFound = _client.MultipleKeystoneResourcesFound
keystone_credentials = _credentials.keystone_credentials
get_keystone_credentials = _credentials.get_keystone_credentials
default_keystone_credentials = _credentials.default_keystone_credentials

182
tobiko/openstack/keystone/_client.py Normal file
View File

@ -0,0 +1,182 @@
# Copyright 2019 Red Hat
#
# 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 __future__ import absolute_import
from keystoneclient import base
from keystoneclient import client as keystoneclient
from keystoneclient.v2_0 import client as v2_client
from keystoneclient.v3 import client as v3_client
from keystoneclient.v3 import endpoints as v3_endpoints
import tobiko
from tobiko.openstack import _client
class KeystoneClientFixture(_client.OpenstackClientFixture):
def init_client(self, session):
return keystoneclient.Client(session=session)
class KeystoneClientManatger(_client.OpenstackClientManager):
def create_client(self, session):
return KeystoneClientFixture(session=session)
CLIENTS = KeystoneClientManatger()
CLIENT_CLASSES = (v2_client.Client, v3_client.Client)
def keystone_client(obj):
if not obj:
return get_keystone_client()
if isinstance(obj, CLIENT_CLASSES):
return obj
fixture = tobiko.setup_fixture(obj)
if isinstance(fixture, KeystoneClientFixture):
return fixture.client
message = "Object {!r} is not a KeystoneClientFixture".format(obj)
raise TypeError(message)
def get_keystone_client(session=None, shared=True, init_client=None,
manager=None):
manager = manager or CLIENTS
client = manager.get_client(session=session, shared=shared,
init_client=init_client)
tobiko.setup_fixture(client)
return client.client
def find_endpoint(client=None, check_found=True, check_unique=False,
**params):
endpoints = list_endpoints(client=client, **params)
return find_resource(resources=endpoints, check_found=check_found,
check_unique=check_unique)
def find_service(client=None, check_found=True, check_unique=False,
**params):
services = list_services(client=client, **params)
return find_resource(resources=services, check_found=check_found,
check_unique=check_unique)
def list_endpoints(client=None, service=None, interface=None, region=None,
translate=True, **params):
client = keystone_client(client)
service = service or params.pop('service_id', None)
if service:
params['service_id'] = base.getid(service)
region = region or params.pop('region_id', None)
if region:
params['region_id'] = base.getid(region)
if client.version == 'v2.0':
endpoints = client.endpoints.list()
if translate:
endpoints = translate_v2_endpoints(v2_endpoints=endpoints,
interface=interface)
else:
endpoints = client.endpoints.list(service=service,
interface=interface,
region=region)
if params:
endpoints = find_resources(endpoints, **params)
return list(endpoints)
def list_services(client=None, name=None, service_type=None, **params):
client = keystone_client(client)
service_type = service_type or params.pop('type', None)
if service_type:
params['type'] = base.getid(service_type)
if name:
params['name'] = name
if client.version == 'v2.0':
services = client.services.list()
else:
services = client.services.list(name=name,
service_type=service_type)
if params:
services = find_resources(services, **params)
return list(services)
def translate_v2_endpoints(v2_endpoints, interface=None):
interfaces = interface and [interface] or v3_endpoints.VALID_INTERFACES
endpoints = []
for endpoint in v2_endpoints:
for interface in interfaces:
url = getattr(endpoint, interface + 'url')
info = dict(id=endpoint.id,
interface=interface,
region_id=endpoint.region,
service_id=endpoint.service_id,
url=url,
enabled=endpoint.enabled)
endpoints.append(v3_endpoints.Endpoint(manager=None,
info=info))
return endpoints
def find_resource(resources, check_found=True, check_unique=True, **params):
"""Look for a service matching some property values"""
resource_it = find_resources(resources, **params)
try:
resource = next(resource_it)
except StopIteration:
resource = None
if check_found and resource is None:
raise KeystoneResourceNotFound(params=params)
if check_unique:
duplicate_ids = [s.id for s in resource_it]
if duplicate_ids:
raise MultipleKeystoneResourcesFound(params=params)
return resource
def find_resources(resources, **params):
"""Look for a service matching some property values"""
# Remove parameters with None value
for resource in resources:
for name, match in params.items():
value = getattr(resource, name)
if match is not None and match != value:
break
else:
yield resource
class KeystoneResourceNotFound(tobiko.TobikoException):
message = 'No such resource found with parameters {params!r}'
class MultipleKeystoneResourcesFound(tobiko.TobikoException):
message = 'Multiple resources found with parameters {params!r}'

5
tobiko/openstack/keystone/_session.py
View File

@ -68,9 +68,10 @@ class KeystoneSessionFixture(tobiko.SharedFixture):
credentials.validate()
loader = loading.get_plugin_loader('password')
params = credentials.to_dict()
del params['api_version'] # parameter not required
# api version parameter is not accepted
params.pop('api_version', None)
auth = loader.load_from_options(**params)
self.session = _session.Session(auth=auth, verify=False)
self.session = session = _session.Session(auth=auth, verify=False)
self.credentials = credentials

126
tobiko/tests/functional/openstack/test_keystone.py
View File

@ -15,6 +15,8 @@
# under the License.
from __future__ import absolute_import
from keystoneclient.v2_0 import client as v2_client
from keystoneclient.v3 import client as v3_client
from oslo_log import log
import testtools
import yaml
@ -25,6 +27,8 @@ from tobiko.shell import sh
LOG = log.getLogger(__name__)
CIENT_CLASSSES = v2_client.Client, v3_client.Client
class TobikoKeystoneCredentialsCommandTest(testtools.TestCase):
@ -34,3 +38,125 @@ class TobikoKeystoneCredentialsCommandTest(testtools.TestCase):
process.check_exit_status()
expected = keystone.default_keystone_credentials().to_dict()
self.assertEqual(expected, actual)
class KeystoneClientAPITest(testtools.TestCase):
def test_get_keystone_client(self):
client = keystone.get_keystone_client()
self.assertIsInstance(client, CIENT_CLASSSES)
def test_list_services(self):
services = keystone.list_services()
self.assertTrue(services)
def test_list_services_by_name(self):
services = keystone.list_services(name='keystone')
self.assertTrue(services)
for s in services:
self.assertEqual('keystone', s.name)
def test_list_services_by_type(self):
services = keystone.list_services(type='identity')
self.assertTrue(services)
for s in services:
self.assertEqual('identity', s.type)
def test_find_service(self):
service = keystone.find_service()
self.assertTrue(service.id)
def test_find_service_with_check_unique(self):
self.assertRaises(keystone.MultipleKeystoneResourcesFound,
keystone.find_service, check_unique=True)
def test_find_service_not_found(self):
self.assertRaises(keystone.KeystoneResourceNotFound,
keystone.find_service, name='never-never-land')
def test_find_service_without_check_found(self):
service = keystone.find_service(check_found=False,
name='never-never-land')
self.assertIsNone(service)
def test_find_service_by_name(self):
service = keystone.find_service(name='keystone')
self.assertEqual('keystone', service.name)
def test_find_service_by_type(self):
service = keystone.find_service(type='identity')
self.assertEqual('identity', service.type)
def test_list_endpoints(self):
service = keystone.find_service(name='keystone')
endpoints = keystone.list_endpoints()
self.assertIn(service.id, [e.service_id for e in endpoints])
def test_list_endpoints_by_service(self):
service = keystone.find_service(name='keystone')
endpoints = keystone.list_endpoints(service=service)
self.assertTrue(endpoints)
self.assertEqual([service.id] * len(endpoints),
[e.service_id for e in endpoints])
def test_list_endpoints_by_service_id(self):
service = keystone.find_service(name='keystone')
endpoints = keystone.list_endpoints(service_id=service.id)
self.assertTrue(endpoints)
for e in endpoints:
self.assertEqual(service.id, e.service_id)
def test_list_endpoints_by_interface(self):
endpoints = keystone.list_endpoints(interface='public')
self.assertTrue(endpoints)
for e in endpoints:
self.assertEqual('public', e.interface)
def test_list_endpoints_by_url(self):
url = keystone.list_endpoints()[-1].url
endpoints = keystone.list_endpoints(url=url)
self.assertTrue(endpoints)
for e in endpoints:
self.assertEqual(url, e.url)
def test_find_endpoint(self):
endpoint = keystone.find_endpoint()
self.assertTrue(endpoint.id)
def test_find_endpoint_with_check_unique(self):
self.assertRaises(keystone.MultipleKeystoneResourcesFound,
keystone.find_endpoint, check_unique=True)
def test_find_endpoint_not_found(self):
self.assertRaises(keystone.KeystoneResourceNotFound,
keystone.find_endpoint,
service='never-never-land')
def test_find_endpoint_without_check_found(self):
service = keystone.find_endpoint(check_found=False,
service='never-never-land')
self.assertIsNone(service)
def test_find_endpoint_by_service(self):
service = keystone.find_service(name='keystone')
endpoint = keystone.find_endpoint(service=service)
self.assertEqual(endpoint.service_id, service.id)
def test_find_endpoint_by_service_id(self):
service = keystone.find_service(name='keystone')
endpoint = keystone.find_endpoint(service_id=service.id)
self.assertEqual(endpoint.service_id, service.id)
def test_find_endpoint_by_url(self):
url = keystone.list_endpoints()[-1].url
endpoint = keystone.find_endpoint(url=url)
self.assertEqual(url, endpoint.url)
def test_find_keystone_public_endpoint(self):
service = keystone.find_service(name='keystone')
endpoint = keystone.find_endpoint(service=service,
interface='public',
enabled=True)
self.assertEqual(service.id, endpoint.service_id)
self.assertEqual('public', endpoint.interface)
self.assertTrue(endpoint.enabled)

20
tobiko/tests/unit/openstack/__init__.py
View File

@ -14,16 +14,33 @@
# under the License.
from __future__ import absolute_import
from keystoneclient import discover
from keystoneclient.v3 import Client
from oslo_log import log
import mock
from tobiko.openstack import keystone
from tobiko.tests import unit
LOG = log.getLogger(__name__)
class KeystoneDiscoverMock(object):
def __init__(self, session, **kwargs):
self.session = session
self.kwargs = kwargs
def create_client(self, version, unstable):
LOG.debug("Create a mock keystone client for version {!r} "
"(unestable = {!r})", version, unstable)
return Client(session=self.session)
class OpenstackTest(unit.TobikoUnitTest):
default_keystone_credentials = keystone.keystone_credentials(
auth_url='http://127.0.0.1:5000/identiy/v3',
auth_url='http://127.0.0.1:5000/v3',
username='default',
project_name='default',
password='this is a secret')
@ -34,6 +51,7 @@ class OpenstackTest(unit.TobikoUnitTest):
self.patch(config.CONF.tobiko, 'keystone',
self.default_keystone_credentials)
self.patch(discover, 'Discover', KeystoneDiscoverMock)
def patch_get_heat_client(self, *args, **kwargs):
from heatclient import client

71
tobiko/tests/unit/openstack/keystone/test_client.py Normal file
View File

@ -0,0 +1,71 @@
# Copyright 2019 Red Hat
#
# 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 __future__ import absolute_import
from keystoneclient.v2_0 import client as client_v2
from keystoneclient.v3 import client as client_v3
from tobiko.openstack import keystone
from tobiko.tests.unit import openstack
from tobiko.tests.unit.openstack import test_client
KEYSTONE_CLIENTS = client_v2.Client, client_v3.Client
class KeystoneClientFixtureTest(test_client.OpenstackClientFixtureTest):
def create_client(self, session=None):
return keystone.KeystoneClientFixture(session=session)
class GetKeystoneClientTest(openstack.OpenstackTest):
def test_get_keystone_client(self, session=None, shared=True):
client1 = keystone.get_keystone_client(session=session, shared=shared)
client2 = keystone.get_keystone_client(session=session, shared=shared)
if shared:
self.assertIs(client1, client2)
else:
self.assertIsNot(client1, client2)
self.assertIsInstance(client1, KEYSTONE_CLIENTS)
self.assertIsInstance(client2, KEYSTONE_CLIENTS)
def test_get_keystone_client_with_not_shared(self):
self.test_get_keystone_client(shared=False)
def test_get_keystone_client_with_session(self):
session = keystone.get_keystone_session()
self.test_get_keystone_client(session=session)
class KeystoneClientTest(openstack.OpenstackTest):
def test_keystone_client_with_none(self):
default_client = keystone.get_keystone_client()
client = keystone.keystone_client(None)
self.assertIsInstance(client, KEYSTONE_CLIENTS)
self.assertIs(default_client, client)
def test_keystone_client_with_client(self):
default_client = keystone.get_keystone_client()
client = keystone.keystone_client(default_client)
self.assertIsInstance(client, KEYSTONE_CLIENTS)
self.assertIs(default_client, client)
def test_keystone_client_with_fixture(self):
fixture = keystone.KeystoneClientFixture()
client = keystone.keystone_client(fixture)
self.assertIsInstance(client, KEYSTONE_CLIENTS)
self.assertIs(client, fixture.client)