Refactor oauth1 code for updates
The keystoneclient code for oauth1 support has changed. As such, we should remove the delete, list and authenticate functions, since they are not in keystoneclient. Also, we must now pass in the project id when creating a request token. Additionally we must now pass in roles when authorizing a request token. Added functional tests to ensure output and input args are the same. bp add-oauth-support Change-Id: I559c18a73ad95a0c8b7a6a95f463b78334186f61
This commit is contained in:
Steve Martinelli
parent
0da5bfe428
commit
d5aaba9d82
@ -43,7 +43,7 @@ class CreateConsumer(show.ShowOne):
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
identity_client = self.app.client_manager.identity
|
||||
consumer = identity_client.consumers.create_consumer(
|
||||
consumer = identity_client.oauth1.consumers.create(
|
||||
parsed_args.description
|
||||
)
|
||||
info = {}
|
||||
@ -69,8 +69,8 @@ class DeleteConsumer(command.Command):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
identity_client = self.app.client_manager.identity
|
||||
consumer = utils.find_resource(
|
||||
identity_client.consumers, parsed_args.consumer)
|
||||
identity_client.consumers.delete_consumer(consumer.id)
|
||||
identity_client.oauth1.consumers, parsed_args.consumer)
|
||||
identity_client.oauth1.consumers.delete(consumer.id)
|
||||
return
|
||||
|
||||
|
||||
@ -82,7 +82,7 @@ class ListConsumer(lister.Lister):
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
columns = ('ID', 'Description')
|
||||
data = self.app.client_manager.identity.consumers.list_consumers()
|
||||
data = self.app.client_manager.identity.oauth1.consumers.list()
|
||||
return (columns,
|
||||
(utils.get_item_properties(
|
||||
s, columns,
|
||||
@ -113,7 +113,7 @@ class SetConsumer(command.Command):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
identity_client = self.app.client_manager.identity
|
||||
consumer = utils.find_resource(
|
||||
identity_client.consumers, parsed_args.consumer)
|
||||
identity_client.oauth1.consumers, parsed_args.consumer)
|
||||
kwargs = {}
|
||||
if parsed_args.description:
|
||||
kwargs['description'] = parsed_args.description
|
||||
@ -122,14 +122,9 @@ class SetConsumer(command.Command):
|
||||
sys.stdout.write("Consumer not updated, no arguments present")
|
||||
return
|
||||
|
||||
consumer = identity_client.consumers.update_consumer(
|
||||
consumer.id,
|
||||
**kwargs
|
||||
)
|
||||
|
||||
info = {}
|
||||
info.update(consumer._info)
|
||||
return zip(*sorted(six.iteritems(info)))
|
||||
consumer = identity_client.oauth1.consumers.update(
|
||||
consumer.id, **kwargs)
|
||||
return
|
||||
|
||||
|
||||
class ShowConsumer(show.ShowOne):
|
||||
@ -150,7 +145,7 @@ class ShowConsumer(show.ShowOne):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
identity_client = self.app.client_manager.identity
|
||||
consumer = utils.find_resource(
|
||||
identity_client.consumers, parsed_args.consumer)
|
||||
identity_client.oauth1.consumers, parsed_args.consumer)
|
||||
|
||||
info = {}
|
||||
info.update(consumer._info)
|
||||
|
||||
@ -18,55 +18,8 @@
|
||||
import logging
|
||||
import six
|
||||
|
||||
from cliff import command
|
||||
from cliff import lister
|
||||
from cliff import show
|
||||
|
||||
from openstackclient.common import utils
|
||||
|
||||
|
||||
class AuthenticateAccessToken(show.ShowOne):
|
||||
"""Authenticate access token to receive keystone token"""
|
||||
|
||||
api = 'identity'
|
||||
log = logging.getLogger(__name__ + '.AuthenticateAccessToken')
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(AuthenticateAccessToken, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'--consumer-key',
|
||||
metavar='<consumer-key>',
|
||||
help='Consumer key',
|
||||
required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'--consumer-secret',
|
||||
metavar='<consumer-secret>',
|
||||
help='Consumer secret',
|
||||
required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'--access-key',
|
||||
metavar='<access-key>',
|
||||
help='Access token key',
|
||||
required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'--access-secret',
|
||||
metavar='<access-secret>',
|
||||
help='Access token secret',
|
||||
required=True
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
token_client = self.app.client_manager.identity.tokens
|
||||
keystone_token = token_client.authenticate_access_token(
|
||||
parsed_args.consumer_key, parsed_args.consumer_secret,
|
||||
parsed_args.access_key, parsed_args.access_secret)
|
||||
return zip(*sorted(six.iteritems(keystone_token)))
|
||||
|
||||
|
||||
class AuthorizeRequestToken(show.ShowOne):
|
||||
"""Authorize request token command"""
|
||||
@ -78,17 +31,28 @@ class AuthorizeRequestToken(show.ShowOne):
|
||||
parser.add_argument(
|
||||
'--request-key',
|
||||
metavar='<request-key>',
|
||||
help='Consumer key',
|
||||
help='Request token key',
|
||||
required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'--role-ids',
|
||||
metavar='<role-ids>',
|
||||
help='Requested role IDs',
|
||||
required=True
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
token_client = self.app.client_manager.identity.tokens
|
||||
self.log.debug('take_action(%s)' % parsed_args)
|
||||
identity_client = self.app.client_manager.identity
|
||||
|
||||
verifier_pin = token_client.authorize_request_token(
|
||||
parsed_args.request_key)
|
||||
roles = []
|
||||
for r_id in parsed_args.role_ids.split():
|
||||
roles.append(r_id)
|
||||
|
||||
verifier_pin = identity_client.oauth1.request_tokens.authorize(
|
||||
parsed_args.request_key,
|
||||
roles)
|
||||
info = {}
|
||||
info.update(verifier_pin._info)
|
||||
return zip(*sorted(six.iteritems(info)))
|
||||
@ -134,13 +98,15 @@ class CreateAccessToken(show.ShowOne):
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
token_client = self.app.client_manager.identity.tokens
|
||||
access_token = token_client.create_access_token(
|
||||
self.log.debug('take_action(%s)' % parsed_args)
|
||||
token_client = self.app.client_manager.identity.oauth1.access_tokens
|
||||
access_token = token_client.create(
|
||||
parsed_args.consumer_key, parsed_args.consumer_secret,
|
||||
parsed_args.request_key, parsed_args.request_secret,
|
||||
parsed_args.verifier)
|
||||
return zip(*sorted(six.iteritems(access_token)))
|
||||
info = {}
|
||||
info.update(access_token._info)
|
||||
return zip(*sorted(six.iteritems(info)))
|
||||
|
||||
|
||||
class CreateRequestToken(show.ShowOne):
|
||||
@ -162,27 +128,24 @@ class CreateRequestToken(show.ShowOne):
|
||||
help='Consumer secret',
|
||||
required=True
|
||||
)
|
||||
parser.add_argument(
|
||||
'--role-ids',
|
||||
metavar='<role-ids>',
|
||||
help='Requested role IDs',
|
||||
)
|
||||
parser.add_argument(
|
||||
'--project-id',
|
||||
metavar='<project-id>',
|
||||
help='Requested project ID',
|
||||
required=True
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
token_client = self.app.client_manager.identity.tokens
|
||||
request_token = token_client.create_request_token(
|
||||
self.log.debug('take_action(%s)' % parsed_args)
|
||||
token_client = self.app.client_manager.identity.oauth1.request_tokens
|
||||
request_token = token_client.create(
|
||||
parsed_args.consumer_key,
|
||||
parsed_args.consumer_secret,
|
||||
parsed_args.role_ids,
|
||||
parsed_args.project_id)
|
||||
return zip(*sorted(six.iteritems(request_token)))
|
||||
info = {}
|
||||
info.update(request_token._info)
|
||||
return zip(*sorted(six.iteritems(info)))
|
||||
|
||||
|
||||
class CreateToken(show.ShowOne):
|
||||
@ -201,64 +164,3 @@ class CreateToken(show.ShowOne):
|
||||
if 'tenant_id' in token:
|
||||
token['project_id'] = token.pop('tenant_id')
|
||||
return zip(*sorted(six.iteritems(token)))
|
||||
|
||||
|
||||
class DeleteAccessToken(command.Command):
|
||||
"""Delete access token command"""
|
||||
|
||||
log = logging.getLogger(__name__ + '.DeleteAccessToken')
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(DeleteAccessToken, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'user',
|
||||
metavar='<user>',
|
||||
help='Name or ID of user',
|
||||
)
|
||||
parser.add_argument(
|
||||
'access_key',
|
||||
metavar='<access-key>',
|
||||
help='Access token to be deleted',
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
|
||||
identity_client = self.app.client_manager.identity
|
||||
user = utils.find_resource(
|
||||
identity_client.users, parsed_args.user).id
|
||||
identity_client.tokens.delete_access_token(user,
|
||||
parsed_args.access_key)
|
||||
return
|
||||
|
||||
|
||||
class ListAccessToken(lister.Lister):
|
||||
"""List access tokens command"""
|
||||
|
||||
log = logging.getLogger(__name__ + '.ListAccessToken')
|
||||
|
||||
def get_parser(self, prog_name):
|
||||
parser = super(ListAccessToken, self).get_parser(prog_name)
|
||||
parser.add_argument(
|
||||
'user',
|
||||
metavar='<user>',
|
||||
help='Name or ID of user',
|
||||
)
|
||||
return parser
|
||||
|
||||
def take_action(self, parsed_args):
|
||||
self.log.debug('take_action(%s)', parsed_args)
|
||||
|
||||
identity_client = self.app.client_manager.identity
|
||||
user = utils.find_resource(
|
||||
identity_client.users, parsed_args.user).id
|
||||
|
||||
columns = ('ID', 'Consumer ID', 'Expires At',
|
||||
'Project Id', 'Authorizing User Id')
|
||||
data = identity_client.tokens.list_access_tokens(user)
|
||||
return (columns,
|
||||
(utils.get_item_properties(
|
||||
s, columns,
|
||||
formatters={},
|
||||
) for s in data))
|
||||
|
||||
@ -139,6 +139,43 @@ ASSIGNMENT_WITH_DOMAIN_ID_AND_GROUP_ID = {
|
||||
'role': {'id': role_id},
|
||||
}
|
||||
|
||||
consumer_id = 'test consumer id'
|
||||
consumer_description = 'someone we trust'
|
||||
consumer_secret = 'test consumer secret'
|
||||
|
||||
OAUTH_CONSUMER = {
|
||||
'id': consumer_id,
|
||||
'secret': consumer_secret,
|
||||
'description': consumer_description
|
||||
}
|
||||
|
||||
access_token_id = 'test access token id'
|
||||
access_token_secret = 'test access token secret'
|
||||
access_token_expires = '2014-05-18T03:13:18.152071Z'
|
||||
|
||||
OAUTH_ACCESS_TOKEN = {
|
||||
'id': access_token_id,
|
||||
'expires': access_token_expires,
|
||||
'key': access_token_id,
|
||||
'secret': access_token_secret
|
||||
}
|
||||
|
||||
request_token_id = 'test request token id'
|
||||
request_token_secret = 'test request token secret'
|
||||
request_token_expires = '2014-05-17T11:10:51.511336Z'
|
||||
|
||||
OAUTH_REQUEST_TOKEN = {
|
||||
'id': request_token_id,
|
||||
'expires': request_token_expires,
|
||||
'key': request_token_id,
|
||||
'secret': request_token_secret
|
||||
}
|
||||
|
||||
oauth_verifier_pin = '6d74XaDS'
|
||||
OAUTH_VERIFIER = {
|
||||
'oauth_verifier': oauth_verifier_pin
|
||||
}
|
||||
|
||||
|
||||
class FakeIdentityv3Client(object):
|
||||
def __init__(self, **kwargs):
|
||||
@ -146,6 +183,8 @@ class FakeIdentityv3Client(object):
|
||||
self.domains.resource_class = fakes.FakeResource(None, {})
|
||||
self.groups = mock.Mock()
|
||||
self.groups.resource_class = fakes.FakeResource(None, {})
|
||||
self.oauth1 = mock.Mock()
|
||||
self.oauth1.resource_class = fakes.FakeResource(None, {})
|
||||
self.projects = mock.Mock()
|
||||
self.projects.resource_class = fakes.FakeResource(None, {})
|
||||
self.roles = mock.Mock()
|
||||
@ -169,6 +208,18 @@ class FakeFederatedClient(FakeIdentityv3Client):
|
||||
self.identity_providers.resource_class = fakes.FakeResource(None, {})
|
||||
|
||||
|
||||
class FakeOAuth1Client(FakeIdentityv3Client):
|
||||
def __init__(self, **kwargs):
|
||||
super(FakeOAuth1Client, self).__init__(**kwargs)
|
||||
|
||||
self.access_tokens = mock.Mock()
|
||||
self.access_tokens.resource_class = fakes.FakeResource(None, {})
|
||||
self.consumers = mock.Mock()
|
||||
self.consumers.resource_class = fakes.FakeResource(None, {})
|
||||
self.request_tokens = mock.Mock()
|
||||
self.request_tokens.resource_class = fakes.FakeResource(None, {})
|
||||
|
||||
|
||||
class TestIdentityv3(utils.TestCommand):
|
||||
def setUp(self):
|
||||
super(TestIdentityv3, self).setUp()
|
||||
@ -187,3 +238,13 @@ class TestFederatedIdentity(utils.TestCommand):
|
||||
endpoint=fakes.AUTH_URL,
|
||||
token=fakes.AUTH_TOKEN
|
||||
)
|
||||
|
||||
|
||||
class TestOAuth1(utils.TestCommand):
|
||||
def setUp(self):
|
||||
super(TestOAuth1, self).setUp()
|
||||
|
||||
self.app.client_manager.identity = FakeOAuth1Client(
|
||||
endpoint=fakes.AUTH_URL,
|
||||
token=fakes.AUTH_TOKEN
|
||||
)
|
||||
|
||||
200
openstackclient/tests/identity/v3/test_consumer.py
Normal file
200
openstackclient/tests/identity/v3/test_consumer.py
Normal file
@ -0,0 +1,200 @@
|
||||
# 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 copy
|
||||
|
||||
from openstackclient.identity.v3 import consumer
|
||||
from openstackclient.tests import fakes
|
||||
from openstackclient.tests.identity.v3 import fakes as identity_fakes
|
||||
|
||||
|
||||
class TestOAuth1(identity_fakes.TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestOAuth1, self).setUp()
|
||||
identity_client = self.app.client_manager.identity
|
||||
self.consumers_mock = identity_client.oauth1.consumers
|
||||
self.consumers_mock.reset_mock()
|
||||
|
||||
|
||||
class TestConsumerCreate(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestConsumerCreate, self).setUp()
|
||||
|
||||
self.consumers_mock.create.return_value = fakes.FakeResource(
|
||||
None, copy.deepcopy(identity_fakes.OAUTH_CONSUMER),
|
||||
loaded=True)
|
||||
|
||||
self.cmd = consumer.CreateConsumer(self.app, None)
|
||||
|
||||
def test_create_consumer(self):
|
||||
arglist = [
|
||||
'--description', identity_fakes.consumer_description
|
||||
]
|
||||
verifylist = [
|
||||
('description', identity_fakes.consumer_description)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.consumers_mock.create.assert_called_with(
|
||||
identity_fakes.consumer_description)
|
||||
|
||||
collist = ('description', 'id', 'secret')
|
||||
self.assertEqual(columns, collist)
|
||||
datalist = (
|
||||
identity_fakes.consumer_description,
|
||||
identity_fakes.consumer_id,
|
||||
identity_fakes.consumer_secret
|
||||
)
|
||||
self.assertEqual(data, datalist)
|
||||
|
||||
|
||||
class TestConsumerDelete(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestConsumerDelete, self).setUp()
|
||||
|
||||
# This is the return value for utils.find_resource()
|
||||
self.consumers_mock.get.return_value = fakes.FakeResource(
|
||||
None,
|
||||
copy.deepcopy(identity_fakes.OAUTH_CONSUMER),
|
||||
loaded=True)
|
||||
|
||||
self.consumers_mock.delete.return_value = None
|
||||
self.cmd = consumer.DeleteConsumer(self.app, None)
|
||||
|
||||
def test_delete_consumer(self):
|
||||
arglist = [
|
||||
identity_fakes.consumer_id
|
||||
]
|
||||
verifylist = [
|
||||
('consumer', identity_fakes.consumer_id)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
result = self.cmd.run(parsed_args)
|
||||
self.assertEqual(result, 0)
|
||||
|
||||
self.consumers_mock.delete.assert_called_with(
|
||||
identity_fakes.consumer_id,
|
||||
)
|
||||
|
||||
|
||||
class TestConsumerList(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestConsumerList, self).setUp()
|
||||
|
||||
self.consumers_mock.get.return_value = fakes.FakeResource(
|
||||
None,
|
||||
copy.deepcopy(identity_fakes.OAUTH_CONSUMER),
|
||||
loaded=True,
|
||||
)
|
||||
self.consumers_mock.list.return_value = [
|
||||
fakes.FakeResource(
|
||||
None,
|
||||
copy.deepcopy(identity_fakes.OAUTH_CONSUMER),
|
||||
loaded=True,
|
||||
),
|
||||
]
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = consumer.ListConsumer(self.app, None)
|
||||
|
||||
def test_consumer_list(self):
|
||||
arglist = []
|
||||
verifylist = []
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
|
||||
# DisplayCommandBase.take_action() returns two tuples
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
self.consumers_mock.list.assert_called_with()
|
||||
|
||||
collist = ('ID', 'Description')
|
||||
self.assertEqual(columns, collist)
|
||||
datalist = ((
|
||||
identity_fakes.consumer_id,
|
||||
identity_fakes.consumer_description
|
||||
), )
|
||||
self.assertEqual(tuple(data), datalist)
|
||||
|
||||
|
||||
class TestConsumerShow(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestConsumerShow, self).setUp()
|
||||
|
||||
consumer_no_secret = copy.deepcopy(identity_fakes.OAUTH_CONSUMER)
|
||||
del consumer_no_secret['secret']
|
||||
self.consumers_mock.get.return_value = fakes.FakeResource(
|
||||
None, consumer_no_secret, loaded=True)
|
||||
|
||||
# Get the command object to test
|
||||
self.cmd = consumer.ShowConsumer(self.app, None)
|
||||
|
||||
def test_consumer_show(self):
|
||||
arglist = [
|
||||
identity_fakes.consumer_id
|
||||
]
|
||||
verifylist = [
|
||||
('consumer', identity_fakes.consumer_id)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.consumers_mock.get.assert_called_with(
|
||||
identity_fakes.consumer_id)
|
||||
|
||||
collist = ('description', 'id' )
|
||||
self.assertEqual(columns, collist)
|
||||
datalist = (
|
||||
identity_fakes.consumer_description,
|
||||
identity_fakes.consumer_id
|
||||
)
|
||||
self.assertEqual(data, datalist)
|
||||
|
||||
|
||||
class TestConsumerSet(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestConsumerSet, self).setUp()
|
||||
|
||||
self.consumers_mock.get.return_value = fakes.FakeResource(
|
||||
None, copy.deepcopy(identity_fakes.OAUTH_CONSUMER), loaded=True)
|
||||
|
||||
consumer_updated = copy.deepcopy(identity_fakes.OAUTH_CONSUMER)
|
||||
consumer_updated['description'] = "consumer new description"
|
||||
self.consumers_mock.update.return_value = fakes.FakeResource(
|
||||
None, consumer_updated, loaded=True)
|
||||
|
||||
self.cmd = consumer.SetConsumer(self.app, None)
|
||||
|
||||
def test_consumer_update(self):
|
||||
new_description = "consumer new description"
|
||||
|
||||
arglist = [
|
||||
'--description', new_description,
|
||||
identity_fakes.consumer_id
|
||||
]
|
||||
verifylist = [
|
||||
('description', new_description),
|
||||
('consumer', identity_fakes.consumer_id)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
result = self.cmd.run(parsed_args)
|
||||
self.assertEqual(result, 0)
|
||||
|
||||
kwargs = {'description': new_description}
|
||||
self.consumers_mock.update.assert_called_with(
|
||||
identity_fakes.consumer_id, **kwargs)
|
||||
152
openstackclient/tests/identity/v3/test_oauth.py
Normal file
152
openstackclient/tests/identity/v3/test_oauth.py
Normal file
@ -0,0 +1,152 @@
|
||||
# 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 copy
|
||||
|
||||
from openstackclient.identity.v3 import token
|
||||
from openstackclient.tests import fakes
|
||||
from openstackclient.tests.identity.v3 import fakes as identity_fakes
|
||||
|
||||
|
||||
class TestOAuth1(identity_fakes.TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestOAuth1, self).setUp()
|
||||
identity_client = self.app.client_manager.identity
|
||||
self.access_tokens_mock = identity_client.oauth1.access_tokens
|
||||
self.access_tokens_mock.reset_mock()
|
||||
self.request_tokens_mock = identity_client.oauth1.request_tokens
|
||||
self.request_tokens_mock.reset_mock()
|
||||
|
||||
|
||||
class TestRequestTokenCreate(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestRequestTokenCreate, self).setUp()
|
||||
|
||||
self.request_tokens_mock.create.return_value = fakes.FakeResource(
|
||||
None, copy.deepcopy(identity_fakes.OAUTH_REQUEST_TOKEN),
|
||||
loaded=True)
|
||||
|
||||
self.cmd = token.CreateRequestToken(self.app, None)
|
||||
|
||||
def test_create_request_tokens(self):
|
||||
arglist = [
|
||||
'--consumer-key', identity_fakes.consumer_id,
|
||||
'--consumer-secret', identity_fakes.consumer_secret,
|
||||
'--project-id', identity_fakes.project_id
|
||||
]
|
||||
verifylist = [
|
||||
('consumer_key', identity_fakes.consumer_id),
|
||||
('consumer_secret', identity_fakes.consumer_secret),
|
||||
('project_id', identity_fakes.project_id)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.request_tokens_mock.create.assert_called_with(
|
||||
identity_fakes.consumer_id,
|
||||
identity_fakes.consumer_secret,
|
||||
identity_fakes.project_id)
|
||||
|
||||
collist = ('expires', 'id', 'key', 'secret')
|
||||
self.assertEqual(columns, collist)
|
||||
datalist = (
|
||||
identity_fakes.request_token_expires,
|
||||
identity_fakes.request_token_id,
|
||||
identity_fakes.request_token_id,
|
||||
identity_fakes.request_token_secret
|
||||
)
|
||||
self.assertEqual(data, datalist)
|
||||
|
||||
|
||||
class TestRequestTokenAuthorize(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestRequestTokenAuthorize, self).setUp()
|
||||
|
||||
self.request_tokens_mock.authorize.return_value = \
|
||||
fakes.FakeResource(
|
||||
None, copy.deepcopy(identity_fakes.OAUTH_VERIFIER),
|
||||
loaded=True)
|
||||
|
||||
self.cmd = token.AuthorizeRequestToken(self.app, None)
|
||||
|
||||
def test_authorize_request_tokens(self):
|
||||
arglist = [
|
||||
'--request-key', identity_fakes.request_token_id,
|
||||
'--role-ids', identity_fakes.role_id
|
||||
]
|
||||
verifylist = [
|
||||
('request_key', identity_fakes.request_token_id),
|
||||
('role_ids', identity_fakes.role_id)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.request_tokens_mock.authorize.assert_called_with(
|
||||
identity_fakes.request_token_id,
|
||||
[identity_fakes.role_id])
|
||||
|
||||
collist = ('oauth_verifier',)
|
||||
self.assertEqual(columns, collist)
|
||||
datalist = (
|
||||
identity_fakes.oauth_verifier_pin,
|
||||
)
|
||||
self.assertEqual(data, datalist)
|
||||
|
||||
|
||||
class TestAccessTokenCreate(TestOAuth1):
|
||||
|
||||
def setUp(self):
|
||||
super(TestAccessTokenCreate, self).setUp()
|
||||
|
||||
self.access_tokens_mock.create.return_value = fakes.FakeResource(
|
||||
None, copy.deepcopy(identity_fakes.OAUTH_ACCESS_TOKEN),
|
||||
loaded=True)
|
||||
|
||||
self.cmd = token.CreateAccessToken(self.app, None)
|
||||
|
||||
def test_create_access_tokens(self):
|
||||
arglist = [
|
||||
'--consumer-key', identity_fakes.consumer_id,
|
||||
'--consumer-secret', identity_fakes.consumer_secret,
|
||||
'--request-key', identity_fakes.request_token_id,
|
||||
'--request-secret', identity_fakes.request_token_secret,
|
||||
'--verifier', identity_fakes.oauth_verifier_pin
|
||||
]
|
||||
verifylist = [
|
||||
('consumer_key', identity_fakes.consumer_id),
|
||||
('consumer_secret', identity_fakes.consumer_secret),
|
||||
('request_key', identity_fakes.request_token_id),
|
||||
('request_secret', identity_fakes.request_token_secret),
|
||||
('verifier', identity_fakes.oauth_verifier_pin)
|
||||
]
|
||||
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
|
||||
columns, data = self.cmd.take_action(parsed_args)
|
||||
|
||||
self.access_tokens_mock.create.assert_called_with(
|
||||
identity_fakes.consumer_id,
|
||||
identity_fakes.consumer_secret,
|
||||
identity_fakes.request_token_id,
|
||||
identity_fakes.request_token_secret,
|
||||
identity_fakes.oauth_verifier_pin)
|
||||
|
||||
collist = ('expires', 'id', 'key', 'secret')
|
||||
self.assertEqual(columns, collist)
|
||||
datalist = (
|
||||
identity_fakes.access_token_expires,
|
||||
identity_fakes.access_token_id,
|
||||
identity_fakes.access_token_id,
|
||||
identity_fakes.access_token_secret
|
||||
)
|
||||
self.assertEqual(data, datalist)
|
||||
@ -163,10 +163,7 @@ openstack.identity.v2_0 =
|
||||
user_show = openstackclient.identity.v2_0.user:ShowUser
|
||||
|
||||
openstack.identity.v3 =
|
||||
access_token_authenticate = openstackclient.identity.v3.token:AuthenticateAccessToken
|
||||
access_token_create = openstackclient.identity.v3.token:CreateAccessToken
|
||||
access_token_delete = openstackclient.identity.v3.token:DeleteAccessToken
|
||||
access_token_list = openstackclient.identity.v3.token:ListAccessToken
|
||||
|
||||
consumer_create = openstackclient.identity.v3.consumer:CreateConsumer
|
||||
consumer_delete = openstackclient.identity.v3.consumer:DeleteConsumer
|
||||
|
||||
Loading…
Reference in New Issue
Block a user