openstack/python-troveclient
Code Issues Proposed changes

Adds support for Keystone V3 API

Browse Source 
Updated trove client to support Keystone V3 API. The
Keystoneclient session object is used for authentication, retrieving
the service catalog and HTTP connection/session management.

Added additional CLI parameters for Keystone V3.

Added tests using requests_mock instead of mocking completely the
session object like the previous failed patch:

https://review.openstack.org/#/c/102315

Change-Id: Ib804c0a45b3b4d3f8670ec642b2c2fd1be6cd656
This commit is contained in:
Bob Thyne 2014-11-12 23:21:15 -08:00
parent efb0f37f8d
commit 4f1fb53ecb
6 changed files with 622 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
requirements.txt
View File

@ -6,5 +6,6 @@ argparse
PrettyTable>=0.7,<0.8
requests>=2.2.0,!=2.4.0
simplejson>=2.2.0
python-keystoneclient>=0.11.1
Babel>=1.3
six>=1.7.0

1
test-requirements.txt
View File

@ -4,6 +4,7 @@
hacking>=0.8.0,<0.9
discover
oslosphinx>=2.2.0 # Apache-2.0
requests-mock>=0.5.1 # Apache-2.0
sphinx>=1.1.2,!=1.2.0,!=1.3b1,<1.3
testrepository>=0.0.18
testtools>=0.9.36

126
troveclient/client.py
View File

@ -25,6 +25,7 @@ import logging
import os
import requests
from keystoneclient import adapter
from troveclient.openstack.common.apiclient import client
from troveclient.openstack.common.apiclient import exceptions
from troveclient import service_catalog
@ -50,7 +51,21 @@ if not hasattr(urlparse, 'parse_qsl'):
urlparse.parse_qsl = cgi.parse_qsl
class HTTPClient(object):
class TroveClientMixin(object):
def get_database_api_version_from_endpoint(self):
magic_tuple = urlparse.urlsplit(self.management_url)
scheme, netloc, path, query, frag = magic_tuple
v = path.split("/")[1]
valid_versions = ['v1.0']
if v not in valid_versions:
msg = "Invalid client version '%s'. must be one of: %s" % (
(v, ', '.join(valid_versions)))
raise exceptions.UnsupportedVersion(msg)
return v[1:]
class HTTPClient(TroveClientMixin):
USER_AGENT = 'python-troveclient'
@ -59,7 +74,17 @@ class HTTPClient(object):
proxy_token=None, region_name=None,
endpoint_type='publicURL', service_type=None,
service_name=None, database_service_name=None, retries=None,
http_log_debug=False, cacert=None, bypass_url=None):
http_log_debug=False, cacert=None, bypass_url=None,
auth_system='keystone', auth_plugin=None):
if auth_system != 'keystone' and not auth_plugin:
raise exceptions.AuthSystemNotFound(auth_system)
if not auth_url and auth_system and auth_system != 'keystone':
auth_url = auth_plugin.get_auth_url()
if not auth_url:
raise exceptions.EndpointNotFound()
self.user = user
self.password = password
self.projectid = projectid
@ -89,6 +114,9 @@ class HTTPClient(object):
else:
self.verify_cert = True
self.auth_system = auth_system
self.auth_plugin = auth_plugin
self._logger = logging.getLogger(__name__)
if self.http_log_debug and not self._logger.handlers:
ch = logging.StreamHandler()
@ -251,7 +279,6 @@ class HTTPClient(object):
except exceptions.EndpointNotFound:
print("Could not find any suitable endpoint. Correct region?")
raise
elif resp.status_code == 305:
return resp['location']
else:
@ -313,6 +340,7 @@ class HTTPClient(object):
# with the endpoints any more, we need to replace
# our service account token with the user token.
self.auth_token = self.proxy_token
else:
try:
while auth_url:
@ -388,16 +416,88 @@ class HTTPClient(object):
return self._extract_service_catalog(url, resp, body)
def get_database_api_version_from_endpoint(self):
magic_tuple = urlparse.urlsplit(self.management_url)
scheme, netloc, path, query, frag = magic_tuple
v = path.split("/")[1]
valid_versions = ['v1.0']
if v not in valid_versions:
msg = "Invalid client version '%s'. must be one of: %s" % (
(v, ', '.join(valid_versions)))
raise exceptions.UnsupportedVersion(msg)
return v[1:]
class SessionClient(adapter.LegacyJsonAdapter, TroveClientMixin):
def __init__(self, session, auth, service_type=None, service_name=None,
region_name=None, endpoint_type='publicURL',
database_service_name=None, endpoint_override=None):
self.endpoint_type = endpoint_type
self.database_service_name = database_service_name
self.endpoint_override = endpoint_override
super(SessionClient, self).__init__(session=session,
auth=auth,
interface=endpoint_type,
service_type=service_type,
service_name=service_name,
region_name=region_name)
self.management_url = self._get_endpoint_url()
def request(self, url, method, **kwargs):
raise_exc = kwargs.pop('raise_exc', True)
resp, body = super(SessionClient, self).request(url,
method,
raise_exc=False,
**kwargs)
if raise_exc and resp.status_code >= 400:
raise exceptions.from_response(resp, body, url)
return resp, body
def _get_endpoint_url(self):
endpoint_url = self.session.get_endpoint(
self.auth, interface=self.endpoint_type,
service_type=self.service_type)
if not endpoint_url:
raise exceptions.EndpointNotFound
return endpoint_url.rstrip('/')
def _construct_http_client(username=None, password=None, project_id=None,
auth_url=None, insecure=False, timeout=None,
proxy_tenant_id=None, proxy_token=None,
region_name=None, endpoint_type='publicURL',
service_type='database',
service_name=None, database_service_name=None,
retries=None,
http_log_debug=False,
auth_system='keystone', auth_plugin=None,
cacert=None, bypass_url=None, tenant_id=None,
session=None,
auth=None):
if session:
return SessionClient(session=session,
auth=auth,
service_type=service_type,
service_name=service_name,
region_name=region_name,
endpoint_type=endpoint_type,
database_service_name=database_service_name)
return HTTPClient(username,
password,
projectid=project_id,
auth_url=auth_url,
insecure=insecure,
timeout=timeout,
tenant_id=tenant_id,
proxy_token=proxy_token,
proxy_tenant_id=proxy_tenant_id,
region_name=region_name,
endpoint_type=endpoint_type,
service_type=service_type,
service_name=service_name,
database_service_name=database_service_name,
retries=retries,
http_log_debug=http_log_debug,
cacert=cacert,
bypass_url=bypass_url,
auth_system=auth_system,
auth_plugin=auth_plugin,
)
def get_version_map():

203
troveclient/shell.py
View File

@ -32,16 +32,23 @@ import sys
import pkg_resources
import six
from keystoneclient.auth.identity.generic import password
from keystoneclient.auth.identity.generic import token
from keystoneclient.auth.identity import v3 as identity
from keystoneclient import session as ks_session
import troveclient
import troveclient.extension
from troveclient import client
from troveclient.openstack.common.apiclient import exceptions as exc
from troveclient.openstack.common import gettextutils as gtu
from troveclient.openstack.common.gettextutils import _ # noqa
from troveclient.openstack.common import strutils
from troveclient import utils
from troveclient.v1 import shell as shell_v1
DEFAULT_OS_DATABASE_API_VERSION = "1.0"
DEFAULT_TROVE_ENDPOINT_TYPE = 'publicURL'
DEFAULT_TROVE_SERVICE_TYPE = 'database'
@ -98,53 +105,9 @@ class OpenStackTroveShell(object):
default=False),
help="Print debugging output.")
parser.add_argument('--os-username',
metavar='<auth-user-name>',
default=utils.env('OS_USERNAME',
'TROVE_USERNAME'),
help='Defaults to env[OS_USERNAME].')
parser.add_argument('--os_username',
help=argparse.SUPPRESS)
parser.add_argument('--os-password',
metavar='<auth-password>',
default=utils.env('OS_PASSWORD',
'TROVE_PASSWORD'),
help='Defaults to env[OS_PASSWORD].')
parser.add_argument('--os_password',
help=argparse.SUPPRESS)
parser.add_argument('--os-tenant-name',
metavar='<auth-tenant-name>',
default=utils.env('OS_TENANT_NAME',
'TROVE_PROJECT_ID'),
help='Defaults to env[OS_TENANT_NAME].')
parser.add_argument('--os_tenant_name',
help=argparse.SUPPRESS)
parser.add_argument('--os-tenant-id',
metavar='<auth-tenant-id>',
default=utils.env('OS_TENANT_ID',
'TROVE_TENANT_ID'),
help='Defaults to env[OS_TENANT_ID].')
parser.add_argument('--os_tenant_id',
help=argparse.SUPPRESS)
parser.add_argument('--os-auth-url',
metavar='<auth-url>',
default=utils.env('OS_AUTH_URL',
'TROVE_URL'),
help='Defaults to env[OS_AUTH_URL].')
parser.add_argument('--os_auth_url',
help=argparse.SUPPRESS)
parser.add_argument('--os-region-name',
metavar='<region-name>',
default=utils.env('OS_REGION_NAME',
'TROVE_REGION_NAME'),
help='Defaults to env[OS_REGION_NAME].')
parser.add_argument('--os_region_name',
help=argparse.SUPPRESS)
parser.add_argument('--os-auth-system',
metavar='<auth-system>',
default=utils.env('OS_AUTH_SYSTEM'))
parser.add_argument('--service-type',
metavar='<service-type>',
@ -196,19 +159,6 @@ class OpenStackTroveShell(object):
parser.add_argument('--os_database_api_version',
help=argparse.SUPPRESS)
parser.add_argument('--os-cacert',
metavar='<ca-certificate>',
default=utils.env('OS_CACERT', default=None),
help='Specify a CA bundle file to use in '
'verifying a TLS (https) server certificate. '
'Defaults to env[OS_CACERT].')
parser.add_argument('--insecure',
default=utils.env('TROVECLIENT_INSECURE',
default=False),
action='store_true',
help=argparse.SUPPRESS)
parser.add_argument('--retries',
metavar='<retries>',
type=int,
@ -223,8 +173,45 @@ class OpenStackTroveShell(object):
help='Output JSON instead of prettyprint. '
'Defaults to env[OS_JSON_OUTPUT].')
self._append_global_identity_args(parser)
return parser
def _append_global_identity_args(self, parser):
# Register CLI identity related arguments
# Use Keystoneclient API to register common V3 CLI arguments
ks_session.Session.register_cli_options(parser)
identity.Password.register_argparse_arguments(parser)
parser.add_argument('--os-tenant-name',
metavar='<auth-tenant-name>',
default=utils.env('OS_TENANT_NAME'),
help='Tenant to request authorization on. '
'Defaults to env[OS_TENANT_NAME].')
parser.add_argument('--os_tenant_name',
help=argparse.SUPPRESS)
parser.add_argument('--os-tenant-id',
metavar='<tenant-id>',
default=utils.env('OS_TENANT_ID'),
help='Tenant to request authorization on. '
'Defaults to env[OS_TENANT_ID].')
parser.add_argument('--os_tenant_id',
help=argparse.SUPPRESS)
parser.add_argument('--os-auth-token',
default=utils.env('OS_AUTH_TOKEN'),
help='Defaults to env[OS_AUTH_TOKEN]')
parser.add_argument('--os-region-name',
metavar='<region-name>',
default=utils.env('OS_REGION_NAME'),
help='Specify the region to use. '
'Defaults to env[OS_REGION_NAME].')
parser.add_argument('--os_region_name',
help=argparse.SUPPRESS)
def get_subcommand_parser(self, version):
parser = self.get_base_parser()
@ -334,17 +321,16 @@ class OpenStackTroveShell(object):
if not debug:
return
streamhandler = logging.StreamHandler()
streamformat = "%(levelname)s (%(module)s:%(lineno)d) %(message)s"
streamhandler.setFormatter(logging.Formatter(streamformat))
logger.setLevel(logging.DEBUG)
logger.addHandler(streamhandler)
logging.basicConfig(level=logging.DEBUG,
format=streamformat)
def main(self, argv):
# Parse args once to find version and debug settings
parser = self.get_base_parser()
(options, args) = parser.parse_known_args(argv)
self.setup_debugging(options.debug)
self.options = options
# build available subcommands based on version
self.extensions = self._discover_extensions(
@ -373,14 +359,19 @@ class OpenStackTroveShell(object):
(os_username, os_password, os_tenant_name, os_auth_url,
os_region_name, os_tenant_id, endpoint_type, insecure,
service_type, service_name, database_service_name,
cacert, bypass_url) = (
cacert, bypass_url, os_auth_system) = (
args.os_username, args.os_password,
args.os_tenant_name, args.os_auth_url,
args.os_region_name, args.os_tenant_id,
args.endpoint_type, args.insecure,
args.service_type, args.service_name,
args.database_service_name,
args.os_cacert, args.bypass_url)
args.os_cacert, args.bypass_url, args.os_auth_system)
if os_auth_system and os_auth_system != "keystone":
auth_plugin = troveclient.auth_plugin.load_plugin(os_auth_system)
else:
auth_plugin = None
if not endpoint_type:
endpoint_type = DEFAULT_TROVE_ENDPOINT_TYPE
@ -403,26 +394,64 @@ class OpenStackTroveShell(object):
"via either --os-password or via "
"env[OS_PASSWORD]")
if not (os_tenant_name or os_tenant_id):
raise exc.CommandError("You must provide a tenant_id "
"via either --os-tenant-id or "
"env[OS_TENANT_ID]")
if not os_auth_url:
raise exc.CommandError(
"You must provide an auth url "
"via either --os-auth-url or env[OS_AUTH_URL]")
if not (os_tenant_name or os_tenant_id):
# V3 stuff
project_info_provided = (self.options.os_tenant_name or
self.options.os_tenant_id or
(self.options.os_project_name and
(self.options.os_project_domain_name or
self.options.os_project_domain_id)) or
self.options.os_project_id)
if (not project_info_provided):
raise exc.CommandError(
"You must provide a tenant_id "
"via either --os-tenant-id or env[OS_TENANT_ID]")
_("You must provide a tenant_name, tenant_id, "
"project_id or project_name (with "
"project_domain_name or project_domain_id) via "
" --os-tenant-name (env[OS_TENANT_NAME]),"
" --os-tenant-id (env[OS_TENANT_ID]),"
" --os-project-id (env[OS_PROJECT_ID])"
" --os-project-name (env[OS_PROJECT_NAME]),"
" --os-project-domain-id "
"(env[OS_PROJECT_DOMAIN_ID])"
" --os-project-domain-name "
"(env[OS_PROJECT_DOMAIN_NAME])"))
if not os_auth_url:
raise exc.CommandError(
"You must provide an auth url "
"via either --os-auth-url or env[OS_AUTH_URL]")
use_session = True
if auth_plugin or bypass_url:
use_session = False
keystone_session = None
keystone_auth = None
if use_session:
project_id = args.os_project_id or args.os_tenant_id
project_name = args.os_project_name or args.os_tenant_name
keystone_session = (ks_session.Session.
load_from_cli_options(args))
keystone_auth = self._get_keystone_auth(
keystone_session,
args.os_auth_url,
username=args.os_username,
user_id=args.os_user_id,
user_domain_id=args.os_user_domain_id,
user_domain_name=args.os_user_domain_name,
password=args.os_password,
auth_token=args.os_auth_token,
project_id=project_id,
project_name=project_name,
project_domain_id=args.os_project_domain_id,
project_domain_name=args.os_project_domain_name)
self.cs = client.Client(options.os_database_api_version, os_username,
os_password, os_tenant_name, os_auth_url,
insecure, region_name=os_region_name,
@ -435,17 +464,25 @@ class OpenStackTroveShell(object):
retries=options.retries,
http_log_debug=args.debug,
cacert=cacert,
bypass_url=bypass_url)
bypass_url=bypass_url,
auth_system=os_auth_system,
auth_plugin=auth_plugin,
session=keystone_session,
auth=keystone_auth)
try:
if not utils.isunauthenticated(args.func):
self.cs.authenticate()
# If Keystone is used, authentication is handled as
# part of session.
if not use_session:
self.cs.authenticate()
except exc.Unauthorized:
raise exc.CommandError("Invalid OpenStack Trove credentials.")
except exc.AuthorizationFailure:
raise exc.CommandError("Unable to authorize user")
endpoint_api_version = self.cs.get_database_api_version_from_endpoint()
if endpoint_api_version != options.os_database_api_version:
msg = (("Database API version is set to %s "
"but you are accessing a %s endpoint. "
@ -498,6 +535,20 @@ class OpenStackTroveShell(object):
else:
self.parser.print_help()
def _get_keystone_auth(self, session, auth_url, **kwargs):
auth_token = kwargs.pop('auth_token', None)
if auth_token:
return token.Token(auth_url, auth_token, **kwargs)
else:
return password.Password(
auth_url,
username=kwargs.pop('username'),
user_id=kwargs.pop('user_id'),
password=kwargs.pop('password'),
user_domain_id=kwargs.pop('user_domain_id'),
user_domain_name=kwargs.pop('user_domain_name'),
**kwargs)
# I'm picky about my shell help.
class OpenStackHelpFormatter(argparse.HelpFormatter):

366
troveclient/tests/test_shell.py Normal file
View File

@ -0,0 +1,366 @@
#
# 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 re
import sys
import fixtures
from keystoneclient import fixture
import mock
import requests_mock
import six
import testtools
import uuid
import troveclient.client
from troveclient import exceptions
import troveclient.shell
try:
import json
except ImportError:
import simplejson as json
V2_URL = "http://no.where/v2.0"
V3_URL = "http://no.where/v3"
FAKE_V2_ENV = {'OS_USERNAME': uuid.uuid4().hex,
'OS_PASSWORD': uuid.uuid4().hex,
'OS_TENANT_ID': uuid.uuid4().hex,
'OS_AUTH_URL': V2_URL}
FAKE_V3_ENV = {'OS_USERNAME': uuid.uuid4().hex,
'OS_PASSWORD': uuid.uuid4().hex,
'OS_PROJECT_ID': uuid.uuid4().hex,
'OS_USER_DOMAIN_NAME': uuid.uuid4().hex,
'OS_AUTH_URL': V3_URL}
UPDATED = '2013-03-06T00:00:00Z'
TEST_SERVICE_CATALOG = [{
"endpoints": [{
"adminURL": "http://no.where/admin",
"region": "RegionOne",
"internalURL": "http://no.where/internal",
"publicURL": "http://no.where/v1.0"
}],
"type": "database",
"name": "trove"
}]
def _create_ver_list(versions):
return {'versions': {'values': versions}}
class ShellTest(testtools.TestCase):
version_id = u'v2.0'
links = [{u'href': u'http://no.where/v2.0', u'rel': u'self'}]
v2_version = fixture.V2Discovery(V2_URL)
v2_version.updated_str = UPDATED
v2_auth_response = json.dumps({
"access": {
"token": {
"expires_at": "2020-01-01T00:00:10.000123Z",
"id": 'fakeToken',
"tenant": {
"id": uuid.uuid4().hex
},
},
"user": {
"id": uuid.uuid4().hex
},
"serviceCatalog": TEST_SERVICE_CATALOG,
},
})
def make_env(self, exclude=None, fake_env=FAKE_V2_ENV):
env = dict((k, v) for k, v in fake_env.items() if k != exclude)
self.useFixture(fixtures.MonkeyPatch('os.environ', env))
def setUp(self):
super(ShellTest, self).setUp()
self.useFixture(fixtures.MonkeyPatch(
'troveclient.client.get_client_class',
mock.MagicMock))
def shell(self, argstr, exitcodes=(0,)):
orig = sys.stdout
orig_stderr = sys.stderr
try:
sys.stdout = six.StringIO()
sys.stderr = six.StringIO()
_shell = troveclient.shell.OpenStackTroveShell()
_shell.main(argstr.split())
except SystemExit:
exc_type, exc_value, exc_traceback = sys.exc_info()
self.assertIn(exc_value.code, exitcodes)
finally:
stdout = sys.stdout.getvalue()
sys.stdout.close()
sys.stdout = orig
stderr = sys.stderr.getvalue()
sys.stderr.close()
sys.stderr = orig_stderr
return (stdout, stderr)
def register_keystone_discovery_fixture(self, mreq):
mreq.register_uri('GET', V2_URL,
json=_create_ver_list([self.v2_version]),
status_code=200)
def test_help_unknown_command(self):
self.assertRaises(exceptions.CommandError, self.shell, 'help foofoo')
def test_help(self):
required = [
'.*?^usage: ',
'.*?^See "trove help COMMAND" for help on a specific command',
]
stdout, stderr = self.shell('help')
for r in required:
self.assertThat(
(stdout + stderr),
testtools.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]')
self.make_env(exclude='OS_USERNAME')
try:
self.shell('list')
except exceptions.CommandError as message:
self.assertEqual(required, message.args[0])
else:
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]',)
self.make_env(exclude='OS_AUTH_URL')
try:
self.shell('list')
except exceptions.CommandError as message:
self.assertEqual(required, message.args)
else:
self.fail('CommandError not raised')
@mock.patch('keystoneclient._discover.get_version_data',
return_value=[{u'status': u'stable', u'id': version_id,
u'links': links}])
@mock.patch('troveclient.v1.datastores.DatastoreVersions.list')
@requests_mock.Mocker()
def test_datastore_version_list(self, mock_discover,
mock_list, mock_requests):
expected = '\n'.join([
'+----+------+',
'| ID | Name |',
'+----+------+',
'+----+------+',
''
])
self.make_env()
self.register_keystone_discovery_fixture(mock_requests)
mock_requests.register_uri('POST', "http://no.where/v2.0/tokens",
text=self.v2_auth_response)
stdout, stderr = self.shell('datastore-version-list XXX')
self.assertEqual(expected, (stdout + stderr))
@mock.patch('keystoneclient._discover.get_version_data',
return_value=[{u'status': u'stable', u'id': version_id,
u'links': links}])
@mock.patch('troveclient.v1.datastores.Datastores.list')
@requests_mock.Mocker()
def test_get_datastore_list(self, mock_discover,
mock_list, mock_requests):
expected = '\n'.join([
'+----+------+',
'| ID | Name |',
'+----+------+',
'+----+------+',
''
])
self.make_env()
self.register_keystone_discovery_fixture(mock_requests)
mock_requests.register_uri('POST', "http://no.where/v2.0/tokens",
text=self.v2_auth_response)
stdout, stderr = self.shell('datastore-list')
self.assertEqual(expected, (stdout + stderr))
class ShellTestKeystoneV3(ShellTest):
version_id = u'v3'
links = [{u'href': u'http://no.where/v3', u'rel': u'self'}]
v3_version = fixture.V3Discovery(V3_URL)
v3_version.updated_str = UPDATED
test_service_catalog = [{
"endpoints": [{
"url": "http://no.where/v1.0/",
"region": "RegionOne",
"interface": "public"
}, {
"url": "http://no.where/v1.0",
"region": "RegionOne",
"interface": "internal"
}, {
"url": "http://no.where/v1.0",
"region": "RegionOne",
"interface": "admin"
}],
"type": "database",
"name": "trove"
}]
service_catalog2 = [{
"endpoints": [{
"url": "http://no.where/vXYZ",
"region": "RegionOne",
"interface": "public"
}],
"type": "database",
"name": "trove"
}]
v3_auth_response = json.dumps({
"token": {
"methods": [
"token",
"password"
],
"expires_at": "2020-01-01T00:00:10.000123Z",
"project": {
"domain": {
"id": uuid.uuid4().hex,
"name": uuid.uuid4().hex
},
"id": uuid.uuid4().hex,
"name": uuid.uuid4().hex
},
"user": {
"domain": {
"id": uuid.uuid4().hex,
"name": uuid.uuid4().hex
},
"id": uuid.uuid4().hex,
"name": uuid.uuid4().hex
},
"issued_at": "2013-05-29T16:55:21.468960Z",
"catalog": test_service_catalog
},
})
def make_env(self, exclude=None, fake_env=FAKE_V3_ENV):
if 'OS_AUTH_URL' in fake_env:
fake_env.update({'OS_AUTH_URL': 'http://no.where/v3'})
env = dict((k, v) for k, v in fake_env.items() if k != exclude)
self.useFixture(fixtures.MonkeyPatch('os.environ', env))
def register_keystone_discovery_fixture(self, mreq):
v3_url = "http://no.where/v3"
v3_version = fixture.V3Discovery(v3_url)
mreq.register_uri('GET', v3_url, json=_create_ver_list([v3_version]),
status_code=200)
def test_no_project_id(self):
required = (
u'You must provide a tenant_name, tenant_id, '
u'project_id or project_name (with '
u'project_domain_name or project_domain_id) via '
u' --os-tenant-name (env[OS_TENANT_NAME]),'
u' --os-tenant-id (env[OS_TENANT_ID]),'
u' --os-project-id (env[OS_PROJECT_ID])'
u' --os-project-name (env[OS_PROJECT_NAME]),'
u' --os-project-domain-id '
u'(env[OS_PROJECT_DOMAIN_ID])'
u' --os-project-domain-name '
u'(env[OS_PROJECT_DOMAIN_NAME])'
)
self.make_env(exclude='OS_PROJECT_ID')
try:
self.shell('list')
except exceptions.CommandError as message:
self.assertEqual(required, message.args[0])
else:
self.fail('CommandError not raised')
@mock.patch('keystoneclient._discover.get_version_data',
return_value=[{u'status': u'stable', u'id': version_id,
u'links': links}])
@mock.patch('troveclient.v1.datastores.DatastoreVersions.list')
@requests_mock.Mocker()
def test_datastore_version_list(self, mock_discover,
mock_list, mock_requests):
expected = '\n'.join([
'+----+------+',
'| ID | Name |',
'+----+------+',
'+----+------+',
''
])
self.make_env()
self.register_keystone_discovery_fixture(mock_requests)
mock_requests.register_uri('POST', "http://no.where/v3/auth/tokens",
headers={'X-Subject-Token': 'fakeToken'},
text=self.v3_auth_response)
stdout, stderr = self.shell('datastore-version-list XXX')
self.assertEqual(expected, (stdout + stderr))
@mock.patch('keystoneclient._discover.get_version_data',
return_value=[{u'status': u'stable', u'id': version_id,
u'links': links}])
@mock.patch('troveclient.v1.datastores.Datastores.list')
@requests_mock.Mocker()
def test_get_datastore_list(self, mock_discover,
mock_list, mock_requests):
expected = '\n'.join([
'+----+------+',
'| ID | Name |',
'+----+------+',
'+----+------+',
''
])
self.make_env()
self.register_keystone_discovery_fixture(mock_requests)
mock_requests.register_uri('POST', "http://no.where/v3/auth/tokens",
headers={'X-Subject-Token': 'fakeToken'},
text=self.v3_auth_response)
stdout, stderr = self.shell('datastore-list')
self.assertEqual(expected, (stdout + stderr))
@mock.patch('keystoneclient._discover.get_version_data',
return_value=[{u'status': u'stable', u'id': version_id,
u'links': links}])
@requests_mock.Mocker()
def test_invalid_client_version(self, mock_discover,
mock_requests):
response = json.loads(self.v3_auth_response)
response['token']['catalog'] = self.service_catalog2
self.make_env()
self.register_keystone_discovery_fixture(mock_requests)
mock_requests.register_uri('POST', "http://no.where/v3/auth/tokens",
headers={'X-Subject-Token': 'fakeToken'},
text=json.dumps(response))
try:
self.shell('datastore-list')
except exceptions.UnsupportedVersion:
pass
else:
self.fail('UnsupportedVersion not raised')

22
troveclient/v1/client.py
View File

@ -47,11 +47,13 @@ class Client(object):
def __init__(self, username, password, project_id=None, auth_url='',
insecure=False, timeout=None, tenant_id=None,
proxy_tenant_id=None, proxy_token=None, region_name=None,
endpoint_type='publicURL', extensions=None,
endpoint_type=None, extensions=None,
service_type='database', service_name=None,
database_service_name=None, retries=None,
http_log_debug=False,
cacert=None, bypass_url=None):
cacert=None, bypass_url=None,
auth_system='keystone', auth_plugin=None, session=None,
auth=None):
# self.limits = limits.LimitsManager(self)
# extensions
@ -91,11 +93,11 @@ class Client(object):
setattr(self, extension.name,
extension.manager_class(self))
self.client = trove_client.HTTPClient(
username,
password,
project_id,
auth_url,
self.client = trove_client._construct_http_client(
username=username,
password=password,
project_id=project_id,
auth_url=auth_url,
insecure=insecure,
timeout=timeout,
tenant_id=tenant_id,
@ -109,7 +111,11 @@ class Client(object):
retries=retries,
http_log_debug=http_log_debug,
cacert=cacert,
bypass_url=bypass_url)
bypass_url=bypass_url,
auth_system=auth_system,
auth_plugin=auth_plugin,
session=session,
auth=auth)
def authenticate(self):
"""Authenticate against the server.