Merge "Refactor oauth1 code for updates"
This commit is contained in:
Jenkins
Gerrit Code Review
commit
aba1fb2268
@ -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