openstack/python-magnumclient
Code Issues Proposed changes

Merge "Use os-client-config in shell"

Browse Source
This commit is contained in:
Jenkins 2016-07-25 18:00:34 +00:00 committed by Gerrit Code Review
commit 84bdb36551
7 changed files with 501 additions and 625 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

238
magnumclient/common/apiclient/auth.py
View File

@ -1,238 +0,0 @@
# Copyright 2013 OpenStack Foundation
# Copyright 2013 Spanish National Research Council.
# All Rights Reserved.
#
# 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.
# E0202: An attribute inherited from %s hide this method
# pylint: disable=E0202
########################################################################
#
# THIS MODULE IS DEPRECATED
#
# Please refer to
# https://etherpad.openstack.org/p/kilo-magnumclient-library-proposals for
# the discussion leading to this deprecation.
#
# We recommend checking out the python-openstacksdk project
# (https://launchpad.net/python-openstacksdk) instead.
#
########################################################################
import abc
import argparse
import os
import six
from stevedore import extension
from magnumclient.common.apiclient import exceptions
_discovered_plugins = {}
def discover_auth_systems():
"""Discover the available auth-systems.
This won't take into account the old style auth-systems.
"""
global _discovered_plugins
_discovered_plugins = {}
def add_plugin(ext):
_discovered_plugins[ext.name] = ext.plugin
ep_namespace = "magnumclient.common.apiclient.auth"
mgr = extension.ExtensionManager(ep_namespace)
mgr.map(add_plugin)
def load_auth_system_opts(parser):
"""Load options needed by the available auth-systems into a parser.
This function will try to populate the parser with options from the
available plugins.
"""
group = parser.add_argument_group("Common auth options")
BaseAuthPlugin.add_common_opts(group)
for name, auth_plugin in _discovered_plugins.items():
group = parser.add_argument_group(
"Auth-system '%s' options" % name,
conflict_handler="resolve")
auth_plugin.add_opts(group)
def load_plugin(auth_system):
try:
plugin_class = _discovered_plugins[auth_system]
except KeyError:
raise exceptions.AuthSystemNotFound(auth_system)
return plugin_class(auth_system=auth_system)
def load_plugin_from_args(args):
"""Load required plugin and populate it with options.
Try to guess auth system if it is not specified. Systems are tried in
alphabetical order.
:type args: argparse.Namespace
:raises: AuthPluginOptionsMissing
"""
auth_system = args.os_auth_system
if auth_system:
plugin = load_plugin(auth_system)
plugin.parse_opts(args)
plugin.sufficient_options()
return plugin
for plugin_auth_system in sorted(six.iterkeys(_discovered_plugins)):
plugin_class = _discovered_plugins[plugin_auth_system]
plugin = plugin_class()
plugin.parse_opts(args)
try:
plugin.sufficient_options()
except exceptions.AuthPluginOptionsMissing:
continue
return plugin
raise exceptions.AuthPluginOptionsMissing(["auth_system"])
@six.add_metaclass(abc.ABCMeta)
class BaseAuthPlugin(object):
"""Base class for authentication plugins.
An authentication plugin needs to override at least the authenticate
method to be a valid plugin.
"""
auth_system = None
opt_names = []
common_opt_names = [
"auth_system",
"username",
"password",
"tenant_id",
"tenant_name",
"project_id",
"project_name",
"user_domain_id",
"user_domain_name",
"project_domain_id",
"project_domain_name",
"token",
"auth_url",
]
def __init__(self, auth_system=None, **kwargs):
self.auth_system = auth_system or self.auth_system
self.opts = dict((name, kwargs.get(name))
for name in self.opt_names)
@staticmethod
def _parser_add_opt(parser, opt):
"""Add an option to parser in two variants.
:param opt: option name (with underscores)
"""
dashed_opt = opt.replace("_", "-")
env_var = "OS_%s" % opt.upper()
arg_default = os.environ.get(env_var, "")
arg_help = "Defaults to env[%s]." % env_var
parser.add_argument(
"--os-%s" % dashed_opt,
metavar="<%s>" % dashed_opt,
default=arg_default,
help=arg_help)
parser.add_argument(
"--os_%s" % opt,
metavar="<%s>" % dashed_opt,
help=argparse.SUPPRESS)
@classmethod
def add_opts(cls, parser):
"""Populate the parser with the options for this plugin."""
for opt in cls.opt_names:
# use `BaseAuthPlugin.common_opt_names` since it is never
# changed in child classes
if opt not in BaseAuthPlugin.common_opt_names:
cls._parser_add_opt(parser, opt)
@classmethod
def add_common_opts(cls, parser):
"""Add options that are common for several plugins."""
for opt in cls.common_opt_names:
cls._parser_add_opt(parser, opt)
@staticmethod
def get_opt(opt_name, args):
"""Return option name and value.
:param opt_name: name of the option, e.g., "username"
:param args: parsed arguments
"""
return (opt_name, getattr(args, "os_%s" % opt_name, None))
def parse_opts(self, args):
"""Parse the actual auth-system options if any.
This method is expected to populate the attribute `self.opts` with a
dict containing the options and values needed to make authentication.
"""
self.opts.update(dict(self.get_opt(opt_name, args)
for opt_name in self.opt_names))
def authenticate(self, http_client):
"""Authenticate using plugin defined method.
The method usually analyses `self.opts` and performs
a request to authentication server.
:param http_client: client object that needs authentication
:type http_client: HTTPClient
:raises: AuthorizationFailure
"""
self.sufficient_options()
self._do_authenticate(http_client)
@abc.abstractmethod
def _do_authenticate(self, http_client):
"""Protected method for authentication."""
def sufficient_options(self):
"""Check if all required options are present.
:raises: AuthPluginOptionsMissing
"""
missing = [opt
for opt in self.opt_names
if not self.opts.get(opt)]
if missing:
raise exceptions.AuthPluginOptionsMissing(missing)
@abc.abstractmethod
def token_and_endpoint(self, endpoint_type, service_type):
"""Return token and endpoint.
:param service_type: Service type of the endpoint
:type service_type: string
:param endpoint_type: Type of endpoint.
Possible values: public or publicURL,
internal or internalURL,
admin or adminURL
:type endpoint_type: string
:returns: tuple of token and endpoint strings
:raises: EndpointException
"""

4
magnumclient/common/httpclient.py
View File

@ -44,6 +44,10 @@ def _extract_error_json(body):
if 'error_message' in body_json:
raw_msg = body_json['error_message']
error_json = json.loads(raw_msg)
elif 'error' in body_json:
error_body = body_json['error']
error_json = {'faultstring': error_body['title'],
'debuginfo': error_body['message']}
else:
error_body = body_json['errors'][0]
raw_msg = error_body['title']

270
magnumclient/shell.py
View File

@ -49,7 +49,6 @@ try:
except ImportError:
pass
from magnumclient.common.apiclient import auth
from magnumclient.common import cliutils
from magnumclient import exceptions as exc
from magnumclient.v1 import client as client_v1
@ -57,7 +56,7 @@ from magnumclient.v1 import shell as shell_v1
from magnumclient import version
DEFAULT_API_VERSION = '1'
DEFAULT_ENDPOINT_TYPE = 'publicURL'
DEFAULT_INTERFACE = 'public'
DEFAULT_SERVICE_TYPE = 'container-infra'
logger = logging.getLogger(__name__)
@ -271,20 +270,111 @@ class OpenStackMagnumShell(object):
# type=positive_non_zero_float,
# help="Set HTTP call timeout (in seconds)")
parser.add_argument('--os-auth-url',
metavar='<auth-auth-url>',
default=cliutils.env('OS_AUTH_URL', default=None),
help='Defaults to env[OS_AUTH_URL].')
parser.add_argument('--os-user-id',
metavar='<auth-user-id>',
default=cliutils.env('OS_USER_ID', default=None),
help='Defaults to env[OS_USER_ID].')
parser.add_argument('--os-username',
metavar='<auth-username>',
default=cliutils.env('OS_USERNAME', default=None),
help='Defaults to env[OS_USERNAME].')
parser.add_argument('--os-user-domain-id',
metavar='<auth-user-domain-id>',
default=cliutils.env('OS_USER_DOMAIN_ID',
default=None),
help='Defaults to env[OS_USER_DOMAIN_ID].')
parser.add_argument('--os-user-domain-name',
metavar='<auth-user-domain-name>',
default=cliutils.env('OS_USER_DOMAIN_NAME',
default=None),
help='Defaults to env[OS_USER_DOMAIN_NAME].')
parser.add_argument('--os-project-id',
metavar='<auth-project-id>',
default=cliutils.env('OS_PROJECT_ID',
default=None),
help='Defaults to env[OS_PROJECT_ID].')
parser.add_argument('--os-project-name',
metavar='<auth-project-name>',
default=cliutils.env('OS_PROJECT_NAME',
default=None),
help='Defaults to env[OS_PROJECT_NAME].')
parser.add_argument('--os-tenant-id',
metavar='<auth-tenant-id>',
default=cliutils.env('OS_TENANT_ID',
default=None),
help=argparse.SUPPRESS)
parser.add_argument('--os-tenant-name',
metavar='<auth-tenant-name>',
default=cliutils.env('OS_TENANT_NAME',
default=None),
help=argparse.SUPPRESS)
parser.add_argument('--os-project-domain-id',
metavar='<auth-project-domain-id>',
default=cliutils.env('OS_PROJECT_DOMAIN_ID',
default=None),
help='Defaults to env[OS_PROJECT_DOMAIN_ID].')
parser.add_argument('--os-project-domain-name',
metavar='<auth-project-domain-name>',
default=cliutils.env('OS_PROJECT_DOMAIN_NAME',
default=None),
help='Defaults to env[OS_PROJECT_DOMAIN_NAME].')
parser.add_argument('--os-token',
metavar='<auth-token>',
default=cliutils.env('OS_TOKEN', default=None),
help='Defaults to env[OS_TOKEN].')
parser.add_argument('--os-password',
metavar='<auth-password>',
default=cliutils.env('OS_PASSWORD',
default=None),
help='Defaults to env[OS_PASSWORD].')
parser.add_argument('--service-type',
metavar='<service-type>',
help='Defaults to container for all '
help='Defaults to container-infra for all '
'actions.')
parser.add_argument('--service_type',
help=argparse.SUPPRESS)
parser.add_argument('--endpoint-type',
metavar='<endpoint-type>',
default=cliutils.env('OS_ENDPOINT_TYPE',
default=None),
help=argparse.SUPPRESS)
parser.add_argument('--os-endpoint-type',
metavar='<os-endpoint-type>',
default=cliutils.env('OS_ENDPOINT_TYPE',
default=None),
help='Defaults to env[OS_ENDPOINT_TYPE]')
parser.add_argument('--os-interface',
metavar='<os-interface>',
default=cliutils.env(
'OS_ENDPOINT_TYPE',
default=DEFAULT_ENDPOINT_TYPE),
help='Defaults to env[OS_ENDPOINT_TYPE] or '
+ DEFAULT_ENDPOINT_TYPE + '.')
'OS_INTERFACE',
default=DEFAULT_INTERFACE),
help=argparse.SUPPRESS)
parser.add_argument('--os-cloud',
metavar='<auth-cloud>',
default=cliutils.env('OS_CLOUD', default=None),
help='Defaults to env[OS_CLOUD].')
# NOTE(dtroyer): We can't add --endpoint_type here due to argparse
# thinking usage-list --end is ambiguous; but it
# works fine with only --endpoint-type present
@ -309,12 +399,17 @@ class OpenStackMagnumShell(object):
'verifying a TLS (https) server certificate. '
'Defaults to env[OS_CACERT].')
parser.add_argument('--os-endpoint-override',
metavar='<endpoint-override>',
default=cliutils.env('OS_ENDPOINT_OVERRIDE',
default=None),
help="Use this API endpoint instead of the "
"Service Catalog.")
parser.add_argument('--bypass-url',
metavar='<bypass-url>',
default=cliutils.env('BYPASS_URL', default=None),
dest='bypass_url',
help="Use this API endpoint instead of the "
"Service Catalog.")
help=argparse.SUPPRESS)
parser.add_argument('--bypass_url',
help=argparse.SUPPRESS)
@ -324,9 +419,6 @@ class OpenStackMagnumShell(object):
action='store_true',
help="Do not verify https connections")
# The auth-system-plugins might require some extra options
auth.load_auth_system_opts(parser)
return parser
def get_subcommand_parser(self, version):
@ -434,95 +526,43 @@ class OpenStackMagnumShell(object):
self.do_bash_completion(args)
return 0
(os_username, os_project_name, os_project_id,
os_tenant_id, os_tenant_name,
os_user_domain_id, os_user_domain_name,
os_project_domain_id, os_project_domain_name,
os_auth_url, os_auth_system, endpoint_type,
service_type, bypass_url, insecure) = (
(args.os_username, args.os_project_name, args.os_project_id,
args.os_tenant_id, args.os_tenant_name,
args.os_user_domain_id, args.os_user_domain_name,
args.os_project_domain_id, args.os_project_domain_name,
args.os_auth_url, args.os_auth_system, args.endpoint_type,
args.service_type, args.bypass_url, args.insecure)
)
if not args.service_type:
args.service_type = DEFAULT_SERVICE_TYPE
os_project_id = (os_project_id or os_tenant_id)
os_project_name = (os_project_name or os_tenant_name)
if args.bypass_url:
args.os_endpoint_override = args.bypass_url
if os_auth_system and os_auth_system != "keystone":
auth_plugin = auth.load_plugin(os_auth_system)
else:
auth_plugin = None
args.os_project_id = (args.os_project_id or args.os_tenant_id)
args.os_project_name = (args.os_project_name or args.os_tenant_name)
# Fetched and set later as needed
os_password = None
if not endpoint_type:
endpoint_type = DEFAULT_ENDPOINT_TYPE
if not service_type:
service_type = DEFAULT_SERVICE_TYPE
# NA - there is only one service this CLI accesses
# service_type = utils.get_service_type(args.func) or service_type
# FIXME(usrleon): Here should be restrict for project id same as
# for os_username or os_password but for compatibility it is not.
if not cliutils.isunauthenticated(args.func):
if auth_plugin:
auth_plugin.parse_opts(args)
if (not (args.os_token and
(args.os_auth_url or args.os_endpoint_override)) and
not args.os_cloud
):
if not auth_plugin or not auth_plugin.opts:
if not os_username:
raise exc.CommandError("You must provide a username "
"via either --os-username or "
"env[OS_USERNAME]")
if not os_project_name and not os_project_id:
raise exc.CommandError("You must provide a project name "
"or project id via --os-project-name, "
"--os-project-id, env[OS_PROJECT_NAME] "
"or env[OS_PROJECT_ID]")
if not os_auth_url:
if os_auth_system and os_auth_system != 'keystone':
os_auth_url = auth_plugin.get_auth_url()
if not os_auth_url:
raise exc.CommandError("You must provide an auth url "
"via either --os-auth-url or "
"env[OS_AUTH_URL] or specify an "
"auth_system which defines a "
"default url with --os-auth-system "
"or env[OS_AUTH_SYSTEM]")
# NOTE: The Magnum client authenticates when you create it. So instead of
# creating here and authenticating later, which is what the novaclient
# does, we just create the client later.
# Now check for the password/token of which pieces of the
# identifying keyring key can come from the underlying client
if not cliutils.isunauthenticated(args.func):
# NA - Client can't be used with SecretsHelper
if (auth_plugin and auth_plugin.opts and
"os_password" not in auth_plugin.opts):
use_pw = False
else:
use_pw = True
if use_pw:
# Auth using token must have failed or not happened
# at all, so now switch to password mode and save
# the token when its gotten... using our keyring
# saver
os_password = args.os_password
if not os_password:
if not (args.os_username or args.os_user_id):
raise exc.CommandError(
'Expecting a password provided via either '
'--os-password, env[OS_PASSWORD], or '
'prompted response')
"You must provide a username via either --os-username "
"or via env[OS_USERNAME]"
)
if not args.os_password:
raise exc.CommandError(
"You must provide a password via either "
"--os-password, env[OS_PASSWORD], or prompted "
"response"
)
if (not args.os_project_name and not args.os_project_id):
raise exc.CommandError(
"You must provide a project name or project id via "
"--os-project-name, --os-project-id, "
"env[OS_PROJECT_NAME] or env[OS_PROJECT_ID]"
)
if not args.os_auth_url:
raise exc.CommandError(
"You must provide an auth url via either "
"--os-auth-url or via env[OS_AUTH_URL]"
)
try:
client = {
'1': client_v1,
@ -530,20 +570,32 @@ class OpenStackMagnumShell(object):
except KeyError:
client = client_v1
self.cs = client.Client(username=os_username,
api_key=os_password,
project_id=os_project_id,
project_name=os_project_name,
user_domain_id=os_user_domain_id,
user_domain_name=os_user_domain_name,
project_domain_id=os_project_domain_id,
project_domain_name=os_project_domain_name,
auth_url=os_auth_url,
service_type=service_type,
region_name=args.os_region_name,
magnum_url=bypass_url,
endpoint_type=endpoint_type,
insecure=insecure)
args.os_endpoint_type = (args.os_endpoint_type or args.endpoint_type)
if args.os_endpoint_type:
args.os_interface = args.os_endpoint_type
if args.os_interface.endswith('URL'):
args.os_interface = args.os_interface[:-3]
self.cs = client.Client(
cloud=args.os_cloud,
user_id=args.os_user_id,
username=args.os_username,
password=args.os_password,
auth_token=args.os_token,
project_id=args.os_project_id,
project_name=args.os_project_name,
user_domain_id=args.os_user_domain_id,
user_domain_name=args.os_user_domain_name,
project_domain_id=args.os_project_domain_id,
project_domain_name=args.os_project_domain_name,
auth_url=args.os_auth_url,
service_type=args.service_type,
region_name=args.os_region_name,
magnum_url=args.os_endpoint_override,
interface=args.os_interface,
insecure=args.insecure,
)
args.func(self.cs, args)

4
magnumclient/tests/test_client.py
View File

@ -22,9 +22,9 @@ class ClientTest(testtools.TestCase):
@mock.patch('magnumclient.v1.client.Client')
def test_no_version_argument(self, mock_magnum_client):
client.Client(input_auth_token='mytoken', magnum_url='http://myurl/')
client.Client(auth_token='mytoken', magnum_url='http://myurl/')
mock_magnum_client.assert_called_with(
input_auth_token='mytoken', magnum_url='http://myurl/')
auth_token='mytoken', magnum_url='http://myurl/')
@mock.patch('magnumclient.v1.client.Client')
def test_valid_version_argument(self, mock_magnum_client):

114
magnumclient/tests/test_shell.py
View File

@ -128,8 +128,8 @@ class ShellTest(utils.TestCase):
matchers.MatchesRegex(r, re.DOTALL | re.MULTILINE))
def test_no_username(self):
required = ('You must provide a username via either'
' --os-username or env[OS_USERNAME]')
required = ('You must provide a username via'
' either --os-username or via env[OS_USERNAME]')
self.make_env(exclude='OS_USERNAME')
try:
self.shell('bay-list')
@ -140,7 +140,7 @@ class ShellTest(utils.TestCase):
def test_no_user_id(self):
required = ('You must provide a username via'
' either --os-username or env[OS_USERNAME]')
' either --os-username or via env[OS_USERNAME]')
self.make_env(exclude='OS_USER_ID', fake_env=FAKE_ENV2)
try:
self.shell('bay-list')
@ -170,15 +170,13 @@ class ShellTest(utils.TestCase):
self.fail('CommandError not raised')
def test_no_auth_url(self):
required = ('You must provide an auth url'
' via either --os-auth-url or env[OS_AUTH_URL] or'
' specify an auth_system which defines a default url'
' with --os-auth-system or env[OS_AUTH_SYSTEM]',)
required = ("You must provide an auth url via either "
"--os-auth-url or via env[OS_AUTH_URL]")
self.make_env(exclude='OS_AUTH_URL')
try:
self.shell('bay-list')
except exceptions.CommandError as message:
self.assertEqual(required, message.args)
self.assertEqual(required, message.args[0])
else:
self.fail('CommandError not raised')
@ -200,20 +198,6 @@ class ShellTest(utils.TestCase):
_, session_kwargs = mock_session.Session.call_args_list[0]
self.assertEqual(False, session_kwargs['verify'])
@mock.patch('sys.stdin', side_effect=mock.MagicMock)
@mock.patch('getpass.getpass', side_effect=EOFError)
def test_no_password(self, mock_getpass, mock_stdin):
required = ('Expecting a password provided'
' via either --os-password, env[OS_PASSWORD],'
' or prompted response',)
self.make_env(exclude='OS_PASSWORD')
try:
self.shell('bay-list')
except exceptions.CommandError as message:
self.assertEqual(required, message.args)
else:
self.fail('CommandError not raised')
@mock.patch('sys.argv', ['magnum'])
@mock.patch('sys.stdout', six.StringIO())
@mock.patch('sys.stderr', six.StringIO())
@ -228,17 +212,25 @@ class ShellTest(utils.TestCase):
self.assertIn('Command-line interface to the OpenStack Magnum API',
sys.stdout.getvalue())
def _expected_client_kwargs(self):
return {
'password': 'password', 'auth_token': None,
'auth_url': self.AUTH_URL,
'cloud': None, 'interface': 'public',
'insecure': False, 'magnum_url': None,
'project_id': None, 'project_name': 'project_name',
'project_domain_id': None, 'project_domain_name': None,
'region_name': None, 'service_type': 'container-infra',
'user_id': None, 'username': 'username',
'user_domain_id': None, 'user_domain_name': None
}
@mock.patch('magnumclient.v1.client.Client')
def _test_main_region(self, command, expected_region_name, mock_client):
self.shell(command)
mock_client.assert_called_once_with(
username='username', api_key='password',
endpoint_type='publicURL', project_id='',
project_name='project_name', auth_url=self.AUTH_URL,
service_type='container-infra', region_name=expected_region_name,
project_domain_id='', project_domain_name='',
user_domain_id='', user_domain_name='',
magnum_url=None, insecure=False)
expected_args = self._expected_client_kwargs()
expected_args['region_name'] = expected_region_name
mock_client.assert_called_once_with(**expected_args)
def test_main_option_region(self):
self.make_env()
@ -258,27 +250,42 @@ class ShellTest(utils.TestCase):
def test_main_endpoint_public(self, mock_client):
self.make_env()
self.shell('--endpoint-type publicURL bay-list')
mock_client.assert_called_once_with(
username='username', api_key='password',
endpoint_type='publicURL', project_id='',
project_name='project_name', auth_url=self.AUTH_URL,
service_type='container-infra', region_name=None,
project_domain_id='', project_domain_name='',
user_domain_id='', user_domain_name='',
magnum_url=None, insecure=False)
expected_args = self._expected_client_kwargs()
expected_args['interface'] = 'public'
mock_client.assert_called_once_with(**expected_args)
@mock.patch('magnumclient.v1.client.Client')
def test_main_endpoint_internal(self, mock_client):
self.make_env()
self.shell('--endpoint-type internalURL bay-list')
mock_client.assert_called_once_with(
username='username', api_key='password',
endpoint_type='internalURL', project_id='',
project_name='project_name', auth_url=self.AUTH_URL,
service_type='container-infra', region_name=None,
project_domain_id='', project_domain_name='',
user_domain_id='', user_domain_name='',
magnum_url=None, insecure=False)
expected_args = self._expected_client_kwargs()
expected_args['interface'] = 'internal'
mock_client.assert_called_once_with(**expected_args)
@mock.patch('magnumclient.v1.client.Client')
def test_main_os_cloud(self, mock_client):
expected_cloud = 'default'
self.shell('--os-cloud %s bay-list' % expected_cloud)
expected_args = self._expected_client_kwargs()
expected_args['cloud'] = expected_cloud
expected_args['username'] = None
expected_args['password'] = None
expected_args['project_name'] = None
expected_args['auth_url'] = None
mock_client.assert_called_once_with(**expected_args)
@mock.patch('magnumclient.v1.client.Client')
def test_main_env_os_cloud(self, mock_client):
expected_cloud = 'default'
self.make_env(fake_env={'OS_CLOUD': expected_cloud})
self.shell('bay-list')
expected_args = self._expected_client_kwargs()
expected_args['cloud'] = expected_cloud
expected_args['username'] = None
expected_args['password'] = None
expected_args['project_name'] = None
expected_args['auth_url'] = None
mock_client.assert_called_once_with(**expected_args)
class ShellTestKeystoneV3(ShellTest):
@ -301,11 +308,10 @@ class ShellTestKeystoneV3(ShellTest):
def test_main_endpoint_public(self, mock_client):
self.make_env(fake_env=FAKE_ENV4)
self.shell('--endpoint-type publicURL bay-list')
mock_client.assert_called_once_with(
username='username', api_key='password',
endpoint_type='publicURL', project_id='project_id',
project_name='', auth_url=self.AUTH_URL,
service_type='container-infra', region_name=None,
project_domain_id='', project_domain_name='Default',
user_domain_id='', user_domain_name='Default',
magnum_url=None, insecure=False)
expected_args = self._expected_client_kwargs()
expected_args['interface'] = 'public'
expected_args['project_id'] = 'project_id'
expected_args['project_name'] = None
expected_args['project_domain_name'] = 'Default'
expected_args['user_domain_name'] = 'Default'
mock_client.assert_called_once_with(**expected_args)

352
magnumclient/tests/v1/test_client.py
View File

@ -20,174 +20,206 @@ from keystoneauth1.exceptions import catalog
from magnumclient.v1 import client
class ClientTest(testtools.TestCase):
class ClientInitializeTest(testtools.TestCase):
def _load_session_kwargs(self):
return {
'username': None,
'project_id': None,
'project_name': None,
'auth_url': None,
'password': None,
'auth_type': 'password',
'insecure': False,
'user_domain_id': None,
'user_domain_name': None,
'project_domain_id': None,
'project_domain_name': None,
'auth_token': None,
'timeout': 600,
}
def _load_service_type_kwargs(self):
return {
'interface': 'public',
'region_name': None,
'service_name': None,
'service_type': 'container-infra',
}
def _session_client_kwargs(self, session):
kwargs = self._load_service_type_kwargs()
kwargs['endpoint_override'] = None
kwargs['session'] = session
return kwargs
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_session(self, mock_session, http_client):
@mock.patch('magnumclient.v1.client._load_session')
@mock.patch('magnumclient.v1.client._load_service_type',
return_value='container-infra')
def test_init_with_session(self,
mock_load_service_type,
mock_load_session,
mock_http_client):
session = mock.Mock()
client.Client(session=session)
mock_session.assert_not_called()
http_client.assert_called_once_with(
interface='public',
region_name=None,
service_name=None,
service_type='container-infra',
session=session)
mock_load_session.assert_not_called()
mock_load_service_type.assert_called_once_with(
session,
**self._load_service_type_kwargs()
)
mock_http_client.assert_called_once_with(
**self._session_client_kwargs(session)
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.token_endpoint.Token')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_token_and_url(
self, mock_session, mock_token, http_client):
mock_auth_plugin = mock.Mock()
mock_token.return_value = mock_auth_plugin
def _test_init_with_secret(self,
init_func,
mock_load_service_type,
mock_load_session,
mock_http_client,):
expected_password = 'expected_password'
session = mock.Mock()
mock_session.return_value = session
client.Client(input_auth_token='mytoken', magnum_url='http://myurl/')
mock_session.assert_called_once_with(
auth=mock_auth_plugin, verify=True)
http_client.assert_called_once_with(
endpoint_override='http://myurl/',
interface='public',
region_name=None,
service_name=None,
service_type='container-infra',
session=session)
mock_load_session.return_value = session
init_func(expected_password)
load_session_args = self._load_session_kwargs()
load_session_args['password'] = expected_password
mock_load_session.assert_called_once_with(
**load_session_args
)
mock_load_service_type.assert_called_once_with(
session,
**self._load_service_type_kwargs()
)
mock_http_client.assert_called_once_with(
**self._session_client_kwargs(session)
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.loading.get_plugin_loader')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_token(
self, mock_session, mock_loader, http_client):
mock_plugin = mock.Mock()
mock_loader.return_value = mock_plugin
client.Client(input_auth_token='mytoken', auth_url='authurl')
mock_loader.assert_called_once_with('token')
mock_plugin.load_from_options.assert_called_once_with(
auth_url='authurl',
project_id=None,
project_name=None,
project_domain_id=None,
project_domain_name=None,
user_domain_id=None,
user_domain_name=None,
token='mytoken')
http_client.assert_called_once_with(
interface='public',
region_name=None,
service_name=None,
service_type='container-infra',
session=mock.ANY)
@mock.patch('magnumclient.v1.client._load_session')
@mock.patch('magnumclient.v1.client._load_service_type',
return_value='container-infra')
def test_init_with_password(self,
mock_load_service_type,
mock_load_session,
mock_http_client):
self._test_init_with_secret(
lambda x: client.Client(password=x),
mock_load_service_type,
mock_load_session,
mock_http_client
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.loading.get_plugin_loader')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_user(
self, mock_session, mock_loader, http_client):
mock_plugin = mock.Mock()
mock_loader.return_value = mock_plugin
@mock.patch('magnumclient.v1.client._load_session')
@mock.patch('magnumclient.v1.client._load_service_type',
return_value='container-infra')
def test_init_with_api_key(self,
mock_load_service_type,
mock_load_session,
mock_http_client):
self._test_init_with_secret(
lambda x: client.Client(api_key=x),
mock_load_service_type,
mock_load_session,
mock_http_client
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('magnumclient.v1.client._load_session')
@mock.patch('magnumclient.v1.client._load_service_type',
return_value='container-infra')
def test_init_with_auth_token(self,
mock_load_service_type,
mock_load_session,
mock_http_client,):
expected_token = 'expected_password'
session = mock.Mock()
mock_load_session.return_value = session
client.Client(auth_token=expected_token)
load_session_args = self._load_session_kwargs()
load_session_args['auth_token'] = expected_token
load_session_args['auth_type'] = 'token'
mock_load_session.assert_called_once_with(
**load_session_args
)
mock_load_service_type.assert_called_once_with(
session,
**self._load_service_type_kwargs()
)
mock_http_client.assert_called_once_with(
**self._session_client_kwargs(session)
)
def _test_init_with_interface(self,
init_func,
mock_load_service_type,
mock_load_session,
mock_http_client):
expected_interface = 'admin'
session = mock.Mock()
mock_load_session.return_value = session
init_func(expected_interface)
mock_load_session.assert_called_once_with(
**self._load_session_kwargs()
)
expected_kwargs = self._load_service_type_kwargs()
expected_kwargs['interface'] = expected_interface
mock_load_service_type.assert_called_once_with(
session,
**expected_kwargs
)
expected_kwargs = self._session_client_kwargs(session)
expected_kwargs['interface'] = expected_interface
mock_http_client.assert_called_once_with(
**expected_kwargs
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('magnumclient.v1.client._load_session')
@mock.patch('magnumclient.v1.client._load_service_type',
return_value='container-infra')
def test_init_with_interface(self,
mock_load_service_type,
mock_load_session,
mock_http_client):
self._test_init_with_interface(
lambda x: client.Client(interface=x),
mock_load_service_type,
mock_load_session,
mock_http_client
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('magnumclient.v1.client._load_session')
@mock.patch('magnumclient.v1.client._load_service_type',
return_value='container-infra')
def test_init_with_endpoint_type(self,
mock_load_service_type,
mock_load_session,
mock_http_client):
self._test_init_with_interface(
lambda x: client.Client(interface='public',
endpoint_type=('%sURL' % x)),
mock_load_service_type,
mock_load_session,
mock_http_client
)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('magnumclient.v1.client._load_session')
def test_init_with_legacy_service_type(self,
mock_load_session,
mock_http_client):
session = mock.Mock()
mock_load_session.return_value = session
session.get_endpoint.side_effect = [
catalog.EndpointNotFound(),
mock.Mock()
]
client.Client(username='myuser', auth_url='authurl')
mock_loader.assert_called_once_with('password')
mock_plugin.load_from_options.assert_called_once_with(
auth_url='authurl',
username='myuser',
password=None,
project_domain_id=None,
project_domain_name=None,
user_domain_id=None,
user_domain_name=None,
project_id=None,
project_name=None)
http_client.assert_called_once_with(
interface='public',
region_name=None,
service_name=None,
service_type='container-infra',
session=mock.ANY)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.loading.get_plugin_loader')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_legacy_service_type(
self, mock_session, mock_loader, http_client):
mock_plugin = mock.Mock()
mock_loader.return_value = mock_plugin
mock_session_obj = mock.Mock()
mock_session.return_value = mock_session_obj
mock_session_obj.get_endpoint.side_effect = [
catalog.EndpointNotFound(), mock.Mock()]
client.Client(username='myuser', auth_url='authurl')
mock_loader.assert_called_once_with('password')
mock_plugin.load_from_options.assert_called_once_with(
auth_url='authurl',
username='myuser',
password=None,
project_domain_id=None,
project_domain_name=None,
user_domain_id=None,
user_domain_name=None,
project_id=None,
project_name=None)
http_client.assert_called_once_with(
interface='public',
region_name=None,
service_name=None,
service_type='container',
session=mock.ANY)
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.loading.get_plugin_loader')
@mock.patch('keystoneauth1.session.Session')
def test_init_unauthorized(
self, mock_session, mock_loader, http_client):
mock_plugin = mock.Mock()
mock_loader.return_value = mock_plugin
mock_session_obj = mock.Mock()
mock_session.return_value = mock_session_obj
mock_session_obj.get_endpoint.side_effect = Exception()
self.assertRaises(
RuntimeError,
client.Client, username='myuser', auth_url='authurl')
mock_loader.assert_called_once_with('password')
mock_plugin.load_from_options.assert_called_once_with(
auth_url='authurl',
username='myuser',
password=None,
project_domain_id=None,
project_domain_name=None,
user_domain_id=None,
user_domain_name=None,
project_id=None,
project_name=None)
http_client.assert_not_called()
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_endpoint_override(self, mock_session, http_client):
session = mock.Mock()
client.Client(session=session, endpoint_override='magnumurl')
mock_session.assert_not_called()
http_client.assert_called_once_with(
interface='public',
region_name=None,
service_name=None,
service_type='container-infra',
session=session,
endpoint_override='magnumurl')
@mock.patch('magnumclient.common.httpclient.SessionClient')
@mock.patch('keystoneauth1.session.Session')
def test_init_with_magnum_url_and_endpoint_override(self, mock_session,
http_client):
session = mock.Mock()
client.Client(session=session, magnum_url='magnumurl',
endpoint_override='magnumurl_override')
mock_session.assert_not_called()
http_client.assert_called_once_with(
interface='public',
region_name=None,
service_name=None,
service_type='container-infra',
session=session,
endpoint_override='magnumurl')
expected_kwargs = self._session_client_kwargs(session)
expected_kwargs['service_type'] = 'container'
mock_http_client.assert_called_once_with(
**expected_kwargs
)

144
magnumclient/v1/client.py
View File

@ -14,8 +14,8 @@
# limitations under the License.
from keystoneauth1.exceptions import catalog
from keystoneauth1 import loading
from keystoneauth1 import session as ksa_session
import os_client_config
from magnumclient.common import httpclient
from magnumclient.v1 import baymodels
@ -28,6 +28,49 @@ DEFAULT_SERVICE_TYPE = 'container-infra'
LEGACY_DEFAULT_SERVICE_TYPE = 'container'
def _load_session(cloud=None, insecure=False, timeout=None, **kwargs):
cloud_config = os_client_config.OpenStackConfig()
cloud_config = cloud_config.get_one_cloud(
cloud=cloud,
verify=not insecure,
**kwargs)
verify, cert = cloud_config.get_requests_verify_args()
auth = cloud_config.get_auth()
session = ksa_session.Session(
auth=auth, verify=verify, cert=cert,
timeout=timeout)
return session
def _load_service_type(session,
service_type=None, service_name=None,
interface=None, region_name=None):
try:
# Trigger an auth error so that we can throw the exception
# we always have
session.get_endpoint(
service_type=service_type,
service_name=service_name,
interface=interface,
region_name=region_name)
except catalog.EndpointNotFound:
service_type = LEGACY_DEFAULT_SERVICE_TYPE
try:
session.get_endpoint(
service_type=service_type,
service_name=service_name,
interface=interface,
region_name=region_name)
except Exception as e:
raise RuntimeError(str(e))
except Exception as e:
raise RuntimeError(str(e))
return service_type
class Client(object):
def __init__(self, username=None, api_key=None, project_id=None,
project_name=None, auth_url=None, magnum_url=None,
@ -35,87 +78,63 @@ class Client(object):
service_type=DEFAULT_SERVICE_TYPE,
region_name=None, input_auth_token=None,
session=None, password=None, auth_type='password',
interface='public', service_name=None, insecure=False,
interface=None, service_name=None, insecure=False,
user_domain_id=None, user_domain_name=None,
project_domain_id=None, project_domain_name=None):
project_domain_id=None, project_domain_name=None,
auth_token=None, timeout=600, **kwargs):
# We have to keep the api_key are for backwards compat, but let's
# remove it from the rest of our code since it's not a keystone
# concept
if not password:
password = api_key
# Backwards compat for people assing in input_auth_token
if input_auth_token:
auth_token = input_auth_token
# Backwards compat for people assing in endpoint_type
if endpoint_type:
interface = endpoint_type
# osc sometimes give 'None' value
if not interface:
interface = 'public'
if interface.endswith('URL'):
interface = interface[:-3]
# fix (yolanda): os-cloud-config is using endpoint_override
# instead of magnum_url
if endpoint_override and not magnum_url:
magnum_url = endpoint_override
if magnum_url and not endpoint_override:
endpoint_override = magnum_url
if magnum_url and input_auth_token:
auth_type = 'admin_token'
session = None
loader_kwargs = dict(
token=input_auth_token,
endpoint=magnum_url)
elif input_auth_token and not session:
auth_type = 'token'
loader_kwargs = dict(
token=input_auth_token,
auth_url=auth_url,
project_id=project_id,
project_name=project_name,
user_domain_id=user_domain_id,
user_domain_name=user_domain_name,
project_domain_id=project_domain_id,
project_domain_name=project_domain_name)
else:
loader_kwargs = dict(
if not session:
if auth_token:
auth_type = 'token'
session = _load_session(
username=username,
password=password,
auth_url=auth_url,
project_id=project_id,
project_name=project_name,
auth_url=auth_url,
password=password,
auth_type=auth_type,
insecure=insecure,
user_domain_id=user_domain_id,
user_domain_name=user_domain_name,
project_domain_id=project_domain_id,
project_domain_name=project_domain_name)
project_domain_name=project_domain_name,
auth_token=auth_token,
timeout=timeout,
**kwargs
)
# Backwards compatibility for people not passing in Session
if session is None:
loader = loading.get_plugin_loader(auth_type)
# This should be able to handle v2 and v3 Keystone Auth
auth_plugin = loader.load_from_options(**loader_kwargs)
session = ksa_session.Session(
auth=auth_plugin, verify=(not insecure))
client_kwargs = {}
if magnum_url:
client_kwargs['endpoint_override'] = magnum_url
if not magnum_url:
try:
# Trigger an auth error so that we can throw the exception
# we always have
session.get_endpoint(
service_type=service_type,
service_name=service_name,
interface=interface,
region_name=region_name)
except catalog.EndpointNotFound:
service_type = LEGACY_DEFAULT_SERVICE_TYPE
try:
session.get_endpoint(
service_type=service_type,
service_name=service_name,
interface=interface,
region_name=region_name)
except Exception as e:
raise RuntimeError(str(e))
except Exception as e:
raise RuntimeError(str(e))
if not endpoint_override:
service_type = _load_service_type(
session,
service_type=service_type,
service_name=service_name,
interface=interface,
region_name=region_name,
)
self.http_client = httpclient.SessionClient(
service_type=service_type,
@ -123,7 +142,8 @@ class Client(object):
interface=interface,
region_name=region_name,
session=session,
**client_kwargs)
endpoint_override=endpoint_override,
)
self.bays = bays.BayManager(self.http_client)
self.certificates = certificates.CertificateManager(self.http_client)
self.baymodels = baymodels.BayModelManager(self.http_client)