124 lines
4.9 KiB
Python
124 lines
4.9 KiB
Python
# 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 mock
|
|
import testtools
|
|
|
|
from openstack.auth.identity import v2
|
|
from openstack import exceptions
|
|
from openstack.tests.auth import common
|
|
|
|
TEST_URL = 'http://127.0.0.1:5000/v2.0'
|
|
|
|
TEST_SERVICE_CATALOG = common.TEST_SERVICE_CATALOG_V2
|
|
TEST_RESPONSE_DICT = common.TEST_RESPONSE_DICT_V2
|
|
|
|
|
|
class TestV2Auth(testtools.TestCase):
|
|
def test_password(self):
|
|
kargs = {'trust_id': common.TEST_TRUST_ID,
|
|
'tenant_id': common.TEST_TENANT_ID,
|
|
'tenant_name': common.TEST_TENANT_NAME}
|
|
|
|
sot = v2.Password(TEST_URL, common.TEST_USER, common.TEST_PASS,
|
|
**kargs)
|
|
|
|
self.assertEqual(common.TEST_USER, sot.username)
|
|
self.assertEqual(common.TEST_PASS, sot.password)
|
|
self.assertEqual(common.TEST_TRUST_ID, sot.trust_id)
|
|
self.assertEqual(common.TEST_TENANT_ID, sot.tenant_id)
|
|
self.assertEqual(common.TEST_TENANT_NAME, sot.tenant_name)
|
|
expected = {'passwordCredentials': {'password': common.TEST_PASS,
|
|
'username': common.TEST_USER}}
|
|
self.assertEqual(expected, sot.get_auth_data())
|
|
|
|
def test_token(self):
|
|
kargs = {'trust_id': common.TEST_TRUST_ID,
|
|
'tenant_id': common.TEST_TENANT_ID,
|
|
'tenant_name': common.TEST_TENANT_NAME}
|
|
|
|
sot = v2.Token(TEST_URL, common.TEST_TOKEN, **kargs)
|
|
|
|
self.assertEqual(common.TEST_TOKEN, sot.token)
|
|
self.assertEqual(common.TEST_TRUST_ID, sot.trust_id)
|
|
self.assertEqual(common.TEST_TENANT_ID, sot.tenant_id)
|
|
self.assertEqual(common.TEST_TENANT_NAME, sot.tenant_name)
|
|
expected = {'token': {'id': common.TEST_TOKEN}}
|
|
self.assertEqual(expected, sot.get_auth_data())
|
|
|
|
def create_mock_transport(self, xresp):
|
|
transport = mock.Mock()
|
|
transport.post = mock.Mock()
|
|
response = mock.Mock()
|
|
response.json = mock.Mock()
|
|
response.json.return_value = xresp
|
|
transport.post.return_value = response
|
|
return transport
|
|
|
|
def test_authorize_tenant_id(self):
|
|
kargs = {'trust_id': common.TEST_TRUST_ID,
|
|
'tenant_id': common.TEST_TENANT_ID,
|
|
'tenant_name': common.TEST_TENANT_NAME}
|
|
sot = v2.Token(TEST_URL, common.TEST_TOKEN, **kargs)
|
|
xport = self.create_mock_transport(TEST_RESPONSE_DICT)
|
|
|
|
resp = sot.authorize(xport)
|
|
|
|
eurl = TEST_URL + '/tokens'
|
|
eheaders = {'Accept': 'application/json',
|
|
'X-Auth-Token': common.TEST_TOKEN}
|
|
ejson = {'auth': {'token': {'id': common.TEST_TOKEN},
|
|
'trust_id': common.TEST_TRUST_ID,
|
|
'tenantId': common.TEST_TENANT_ID}}
|
|
xport.post.assert_called_with(eurl, headers=eheaders, json=ejson)
|
|
ecatalog = TEST_RESPONSE_DICT['access'].copy()
|
|
ecatalog['version'] = 'v2.0'
|
|
self.assertEqual(ecatalog, resp._info)
|
|
|
|
def test_authorize_tenant_name(self):
|
|
kargs = {'tenant_name': common.TEST_TENANT_NAME}
|
|
sot = v2.Token(TEST_URL, common.TEST_TOKEN, **kargs)
|
|
xport = self.create_mock_transport(TEST_RESPONSE_DICT)
|
|
|
|
resp = sot.authorize(xport)
|
|
|
|
eurl = TEST_URL + '/tokens'
|
|
eheaders = {'Accept': 'application/json',
|
|
'X-Auth-Token': common.TEST_TOKEN}
|
|
ejson = {'auth': {'token': {'id': common.TEST_TOKEN},
|
|
'tenantName': common.TEST_TENANT_NAME}}
|
|
xport.post.assert_called_with(eurl, headers=eheaders, json=ejson)
|
|
ecatalog = TEST_RESPONSE_DICT['access'].copy()
|
|
ecatalog['version'] = 'v2.0'
|
|
self.assertEqual(ecatalog, resp._info)
|
|
|
|
def test_authorize_token_only(self):
|
|
sot = v2.Token(TEST_URL, common.TEST_TOKEN)
|
|
xport = self.create_mock_transport(TEST_RESPONSE_DICT)
|
|
|
|
resp = sot.authorize(xport)
|
|
|
|
eurl = TEST_URL + '/tokens'
|
|
eheaders = {'Accept': 'application/json',
|
|
'X-Auth-Token': common.TEST_TOKEN}
|
|
ejson = {'auth': {'token': {'id': common.TEST_TOKEN}}}
|
|
xport.post.assert_called_with(eurl, headers=eheaders, json=ejson)
|
|
ecatalog = TEST_RESPONSE_DICT['access'].copy()
|
|
ecatalog['version'] = 'v2.0'
|
|
self.assertEqual(ecatalog, resp._info)
|
|
|
|
def test_authorize_bad_response(self):
|
|
sot = v2.Token(TEST_URL, common.TEST_TOKEN)
|
|
xport = self.create_mock_transport({})
|
|
|
|
self.assertRaises(exceptions.InvalidResponse, sot.authorize, xport)
|