Merge "Add loading mock fixtures"
This commit is contained in:
commit
b36d47d6a8
|
@ -21,6 +21,7 @@ from this module on libraries that are only available in testing.
|
|||
|
||||
from keystoneauth1.fixture.discovery import * # noqa
|
||||
from keystoneauth1.fixture import exception
|
||||
from keystoneauth1.fixture.plugin import * # noqa
|
||||
from keystoneauth1.fixture import v2
|
||||
from keystoneauth1.fixture import v3
|
||||
|
||||
|
@ -32,6 +33,8 @@ V3FederationToken = v3.V3FederationToken
|
|||
|
||||
__all__ = ('DiscoveryList',
|
||||
'FixtureValidationError',
|
||||
'LoadingFixture',
|
||||
'TestPlugin',
|
||||
'V2Discovery',
|
||||
'V3Discovery',
|
||||
'V2Token',
|
||||
|
|
|
@ -0,0 +1,176 @@
|
|||
# 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 uuid
|
||||
|
||||
import fixtures
|
||||
|
||||
from keystoneauth1 import discover
|
||||
from keystoneauth1 import loading
|
||||
from keystoneauth1 import plugin
|
||||
|
||||
__all__ = (
|
||||
'LoadingFixture',
|
||||
'TestPlugin',
|
||||
)
|
||||
|
||||
|
||||
DEFAULT_TEST_ENDPOINT = 'https://openstack.example.com/%(service_type)s'
|
||||
|
||||
|
||||
def _format_endpoint(endpoint, **kwargs):
|
||||
# can't format AUTH_INTERFACE object so replace with string
|
||||
if kwargs.get('service_type') is plugin.AUTH_INTERFACE:
|
||||
kwargs['service_type'] = 'identity'
|
||||
|
||||
version = kwargs.get('version')
|
||||
if version:
|
||||
discover.normalize_version_number(version)
|
||||
kwargs['version'] = ".".join(str(v) for v in version)
|
||||
|
||||
return endpoint % kwargs # pass kwargs ok?
|
||||
|
||||
|
||||
class TestPlugin(plugin.BaseAuthPlugin):
|
||||
"""A simple plugin that returns what you gave it for testing.
|
||||
|
||||
When testing services that use authentication plugins you often want to
|
||||
stub out the authentication calls and focus on the important part of your
|
||||
service. This plugin acts like a real keystoneauth plugin and returns known
|
||||
standard values without having to stub out real keystone responses.
|
||||
|
||||
Note that this plugin is a BaseAuthPlugin and not a BaseIdentityPlugin.
|
||||
This means it implements the basic plugin interface that services should be
|
||||
using but does not implement get_auth_ref. get_auth_ref should not be
|
||||
relied upon by services because a user could always configure the service
|
||||
to use a non-keystone auth.
|
||||
|
||||
:param str token: The token to include in authenticated requests.
|
||||
:param str endpoint: The endpoint to respond to service lookups with.
|
||||
:param str user_id: The user_id to report for the authenticated user.
|
||||
:param str project_id: The project_id to report for the authenticated user.
|
||||
"""
|
||||
|
||||
auth_type = 'test_plugin'
|
||||
|
||||
def __init__(self,
|
||||
token=None,
|
||||
endpoint=None,
|
||||
user_id=None,
|
||||
project_id=None):
|
||||
super(TestPlugin, self).__init__()
|
||||
|
||||
self.token = token or uuid.uuid4().hex
|
||||
self.endpoint = endpoint or DEFAULT_TEST_ENDPOINT
|
||||
self.user_id = user_id or uuid.uuid4().hex
|
||||
self.project_id = project_id or uuid.uuid4().hex
|
||||
|
||||
def get_endpoint(self, session, **kwargs):
|
||||
return _format_endpoint(self.endpoint, **kwargs)
|
||||
|
||||
def get_token(self, session, **kwargs):
|
||||
return self.token
|
||||
|
||||
def get_user_id(self, session, **kwargs):
|
||||
return self.user_id
|
||||
|
||||
def get_project_id(self, session, **kwargs):
|
||||
return self.project_id
|
||||
|
||||
def invalidate(self):
|
||||
self.token = uuid.uuid4().hex
|
||||
return True
|
||||
|
||||
# NOTE(jamielennox): You'll notice there's no get_access/get_auth_ref
|
||||
# function here. These functions are only part of identity plugins, which
|
||||
# whilst the most common are not the only way you can authenticate. You're
|
||||
# application should really only rely on the presence of the above
|
||||
# functions, everything else is on a best effort basis.
|
||||
|
||||
|
||||
class _TestPluginLoader(loading.BaseLoader):
|
||||
|
||||
def __init__(self, plugin):
|
||||
super(_TestPluginLoader, self).__init__()
|
||||
self._plugin = plugin
|
||||
|
||||
def create_plugin(self, **kwargs):
|
||||
return self._plugin
|
||||
|
||||
def get_options(self):
|
||||
return []
|
||||
|
||||
|
||||
class LoadingFixture(fixtures.Fixture):
|
||||
"""A fixture that will stub out all plugin loading calls.
|
||||
|
||||
When using keystoneauth plugins loaded from config, CLI or elsewhere it is
|
||||
often difficult to handle the plugin parts in tests because we don't have a
|
||||
reasonable default.
|
||||
|
||||
This fixture will create a :py:class:`TestPlugin` that will be
|
||||
returned for all calls to plugin loading so you can simply bypass the
|
||||
authentication steps and return something well known.
|
||||
|
||||
:param str token: The token to include in authenticated requests.
|
||||
:param str endpoint: The endpoint to respond to service lookups with.
|
||||
:param str user_id: The user_id to report for the authenticated user.
|
||||
:param str project_id: The project_id to report for the authenticated user.
|
||||
"""
|
||||
|
||||
MOCK_POINT = 'keystoneauth1.loading.base.get_plugin_loader'
|
||||
|
||||
def __init__(self,
|
||||
token=None,
|
||||
endpoint=None,
|
||||
user_id=None,
|
||||
project_id=None):
|
||||
super(LoadingFixture, self).__init__()
|
||||
|
||||
# these are created and saved here so that a test could use them
|
||||
self.token = token or uuid.uuid4().hex
|
||||
self.endpoint = endpoint or DEFAULT_TEST_ENDPOINT
|
||||
self.user_id = user_id or uuid.uuid4().hex
|
||||
self.project_id = project_id or uuid.uuid4().hex
|
||||
|
||||
def setUp(self):
|
||||
super(LoadingFixture, self).setUp()
|
||||
|
||||
self.useFixture(fixtures.MonkeyPatch(self.MOCK_POINT,
|
||||
self.get_plugin_loader))
|
||||
|
||||
def create_plugin(self):
|
||||
return TestPlugin(token=self.token,
|
||||
endpoint=self.endpoint,
|
||||
user_id=self.user_id,
|
||||
project_id=self.project_id)
|
||||
|
||||
def get_plugin_loader(self, auth_type):
|
||||
plugin = self.create_plugin()
|
||||
plugin.auth_type = auth_type
|
||||
return _TestPluginLoader(plugin)
|
||||
|
||||
def get_endpoint(self, path=None, **kwargs):
|
||||
"""Utility function to get the endpoint the plugin would return.
|
||||
|
||||
This function is provided as a convenience so you can do comparisons in
|
||||
your tests. Overriding it will not affect the endpoint returned by the
|
||||
plugin.
|
||||
|
||||
:param str path: The path to append to the plugin endpoint.
|
||||
"""
|
||||
endpoint = _format_endpoint(self.endpoint, **kwargs)
|
||||
|
||||
if path:
|
||||
endpoint = "%s/%s" % (endpoint.rstrip('/'), path.lstrip('/'))
|
||||
|
||||
return endpoint
|
|
@ -0,0 +1,88 @@
|
|||
# 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 uuid
|
||||
|
||||
from oslo_config import fixture as config
|
||||
|
||||
from keystoneauth1 import fixture
|
||||
from keystoneauth1 import loading
|
||||
from keystoneauth1 import session
|
||||
from keystoneauth1.tests.unit import utils
|
||||
|
||||
|
||||
class FixturesTests(utils.TestCase):
|
||||
|
||||
GROUP = uuid.uuid4().hex
|
||||
AUTH_TYPE = uuid.uuid4().hex
|
||||
|
||||
def setUp(self):
|
||||
super(FixturesTests, self).setUp()
|
||||
self.conf_fixture = self.useFixture(config.Config())
|
||||
|
||||
# conf loading will still try to read the auth_type from the config
|
||||
# object and pass that to the get_plugin_loader method. This value will
|
||||
# typically be ignored and the fake plugin returned regardless of name
|
||||
# but it could be a useful differentiator and it also ensures that the
|
||||
# application has called register_auth_conf_options before simply
|
||||
# returning a fake plugin.
|
||||
loading.register_auth_conf_options(self.conf_fixture.conf,
|
||||
group=self.GROUP)
|
||||
|
||||
self.conf_fixture.config(auth_type=self.AUTH_TYPE, group=self.GROUP)
|
||||
|
||||
def useLoadingFixture(self, **kwargs):
|
||||
return self.useFixture(fixture.LoadingFixture(**kwargs))
|
||||
|
||||
def test_endpoint_resolve(self):
|
||||
endpoint = "http://%(service_type)s/%(version)s/%(interface)s"
|
||||
loader = self.useLoadingFixture(endpoint=endpoint)
|
||||
|
||||
endpoint_filter = {'service_type': 'compute',
|
||||
'service_name': 'nova',
|
||||
'version': (2, 1),
|
||||
'interface': 'public'}
|
||||
|
||||
auth = loading.load_auth_from_conf_options(self.conf_fixture.conf,
|
||||
self.GROUP)
|
||||
sess = session.Session(auth=auth)
|
||||
|
||||
loader_endpoint = loader.get_endpoint(**endpoint_filter)
|
||||
plugin_endpoint = sess.get_endpoint(**endpoint_filter)
|
||||
|
||||
self.assertEqual("http://compute/2.1/public", loader_endpoint)
|
||||
self.assertEqual(loader_endpoint, plugin_endpoint)
|
||||
|
||||
def test_conf_loaded(self):
|
||||
token = uuid.uuid4().hex
|
||||
endpoint_filter = {'service_type': 'compute',
|
||||
'service_name': 'nova',
|
||||
'version': (2, 1)}
|
||||
|
||||
loader = self.useLoadingFixture(token=token)
|
||||
|
||||
url = loader.get_endpoint('/path', **endpoint_filter)
|
||||
|
||||
m = self.requests_mock.get(url)
|
||||
|
||||
auth = loading.load_auth_from_conf_options(self.conf_fixture.conf,
|
||||
self.GROUP)
|
||||
sess = session.Session(auth=auth)
|
||||
self.assertEqual(self.AUTH_TYPE, auth.auth_type)
|
||||
|
||||
sess.get('/path', endpoint_filter=endpoint_filter)
|
||||
|
||||
self.assertTrue(m.called_once)
|
||||
|
||||
self.assertTrue(token, m.last_request.headers['X-Auth-Token'])
|
||||
self.assertEqual(loader.project_id, sess.get_project_id())
|
||||
self.assertEqual(loader.user_id, sess.get_user_id())
|
Loading…
Reference in New Issue