Keystone REST API for angular front end.
This change implements the REST API for keystone required for the identity angular front end. Changes in this patch: - fix PUT so it actually works - move to using new test util module - address TODO in tests Partially Implemenents: blueprint angularize-identity-tables Change-Id: I5b8cdc44250fcc5afdb41a5a33fb34dc8ea7d6c0
This commit is contained in:
Richard Jones
parent
04b692cc30
commit
fd99a0bd7f
@ -22,3 +22,4 @@ in https://wiki.openstack.org/wiki/APIChangeGuidelines.
|
||||
"""
|
||||
|
||||
# import REST API modules here
|
||||
import keystone #flake8: noqa
|
||||
|
||||
511
openstack_dashboard/api/rest/keystone.py
Normal file
511
openstack_dashboard/api/rest/keystone.py
Normal file
@ -0,0 +1,511 @@
|
||||
# Copyright 2014, Rackspace, US, Inc.
|
||||
#
|
||||
# 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.
|
||||
"""API over the keystone service.
|
||||
"""
|
||||
|
||||
import django.http
|
||||
from django.views import generic
|
||||
|
||||
from openstack_dashboard import api
|
||||
from openstack_dashboard.api.rest import utils as rest_utils
|
||||
|
||||
from openstack_dashboard.api.rest import urls
|
||||
|
||||
|
||||
@urls.register
|
||||
class Users(generic.View):
|
||||
"""API for keystone users.
|
||||
"""
|
||||
url_regex = r'keystone/users/$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request):
|
||||
"""Get a list of users.
|
||||
|
||||
By default, a listing of all users for the current domain are
|
||||
returned. You may specify GET parameters for project_id, group_id and
|
||||
domain_id to change that listing's context.
|
||||
|
||||
The listing result is an object with property "items".
|
||||
"""
|
||||
domain_context = request.session.get('domain_context')
|
||||
result = api.keystone.user_list(
|
||||
request,
|
||||
project=request.GET.get('project_id'),
|
||||
domain=request.GET.get('domain_id', domain_context),
|
||||
group=request.GET.get('group_id')
|
||||
)
|
||||
return {'items': [u.to_dict() for u in result]}
|
||||
|
||||
@rest_utils.ajax(method='POST')
|
||||
def post(self, request):
|
||||
"""Perform some action on the collection of users.
|
||||
|
||||
The POST data should be an application/json object with two
|
||||
parameters: "action" and "data".
|
||||
|
||||
action = "delete"
|
||||
This action deletes multiple users in one call, using the list of
|
||||
ids (strings) passed in as data.
|
||||
|
||||
This action returns HTTP 204 (no content) on success.
|
||||
|
||||
action = "create"
|
||||
This action creates a user using the parameters supplied in
|
||||
"data". The base parameters are name (string), email (string,
|
||||
optional), password (string, optional), project_id (string,
|
||||
optional), enabled (boolean, defaults to true). The user will be
|
||||
created in the default domain.
|
||||
|
||||
This action returns the new user object on success.
|
||||
|
||||
This action returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
action = request.DATA['action']
|
||||
data = request.DATA['data']
|
||||
|
||||
if action == 'delete':
|
||||
for user_id in data:
|
||||
if user_id != request.user.id:
|
||||
api.keystone.user_delete(request, user_id)
|
||||
elif action == 'create':
|
||||
# not sure why email is forced to None, but other code does it
|
||||
domain = api.keystone.get_default_domain(request)
|
||||
new_user = api.keystone.user_create(
|
||||
request,
|
||||
name=data['name'],
|
||||
email=data.get('email') or None,
|
||||
password=data.get('password'),
|
||||
project=data.get('project_id'),
|
||||
enabled=True,
|
||||
domain=domain.id
|
||||
)
|
||||
return rest_utils.CreatedResponse(
|
||||
'/api/keystone/users/%s' % new_user.id,
|
||||
new_user.to_dict()
|
||||
)
|
||||
else:
|
||||
raise rest_utils.AjaxError(400, 'invalid action')
|
||||
|
||||
|
||||
@urls.register
|
||||
class User(generic.View):
|
||||
"""API for a single keystone user.
|
||||
"""
|
||||
url_regex = r'keystone/users/(?P<id>[0-9a-f]+|current)$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request, id):
|
||||
"""Get a specific user by id.
|
||||
|
||||
If the id supplied is 'current' then the current logged-in user
|
||||
will be returned, otherwise the user specified by the id.
|
||||
"""
|
||||
if id == 'current':
|
||||
id = request.user.id
|
||||
return api.keystone.user_get(request, id).to_dict()
|
||||
|
||||
@rest_utils.ajax()
|
||||
def delete(self, request, id):
|
||||
"""Delete a single user by id.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
if id == 'current':
|
||||
raise django.http.HttpResponseNotFound('current')
|
||||
api.keystone.user_delete(request, id)
|
||||
|
||||
@rest_utils.ajax(method='PUT')
|
||||
def put(self, request, id):
|
||||
"""Update a single user.
|
||||
|
||||
The PUT data should be an application/json object with attributes to
|
||||
set to new values: password (string), project_id (string),
|
||||
enabled (boolean). A PUT may contain any one of those attributes, but
|
||||
if it contains more than one it must contain the project_id, even
|
||||
if it is not being altered.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
keys = tuple(request.DATA)
|
||||
if keys == ('password', ):
|
||||
api.keystone.user_update_password(request, id, **request.DATA)
|
||||
elif keys == ('enabled', ):
|
||||
api.keystone.user_update_enabled(request, id, **request.DATA)
|
||||
elif keys == ('project_id', ):
|
||||
api.keystone.user_update_tenant(request, id,
|
||||
project=request.DATA['project_id'])
|
||||
else:
|
||||
# update mutiple things, and hope the caller has supplied
|
||||
# everything
|
||||
request.DATA['project'] = request.DATA.pop('project_id', None)
|
||||
request.DATA.setdefault('password', None)
|
||||
api.keystone.user_update(request, id, **request.DATA)
|
||||
|
||||
|
||||
@urls.register
|
||||
class Roles(generic.View):
|
||||
"""API over all roles.
|
||||
"""
|
||||
url_regex = r'keystone/roles/$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request):
|
||||
"""Get a list of roles.
|
||||
|
||||
By default a listing of all roles are returned.
|
||||
|
||||
If the GET parameters project_id and user_id are specified then that
|
||||
user's roles for that project are returned. If user_id is 'current'
|
||||
then the current user's roles for that project are returned.
|
||||
|
||||
The listing result is an object with property "items".
|
||||
"""
|
||||
project_id = request.GET.get('project_id')
|
||||
user_id = request.GET.get('user_id')
|
||||
if project_id and user_id:
|
||||
if user_id == 'current':
|
||||
user_id = request.user.id
|
||||
roles = api.keystone.roles_for_user(request, user_id,
|
||||
project_id) or []
|
||||
items = [r.to_dict() for r in roles]
|
||||
else:
|
||||
items = [r.to_dict() for r in api.keystone.role_list(request)]
|
||||
return {'items': items}
|
||||
|
||||
@rest_utils.ajax(method='POST')
|
||||
def post(self, request):
|
||||
"""Perform some action on the collection of roles.
|
||||
|
||||
The POST data should be an application/json object with two
|
||||
parameters: "action" and "data".
|
||||
|
||||
action = "delete"
|
||||
This action deletes multiple roles in one call, using the list of
|
||||
ids (strings) passed in as data.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
|
||||
action = "create"
|
||||
This action creates a role using the "name" (string) parameter
|
||||
supplied in the "data" object.
|
||||
|
||||
This method returns the new role object on success.
|
||||
|
||||
action = "grant"
|
||||
This action adds a role to a user using the parameters
|
||||
"user_id" (string), "project_id" (string) and "role_id" (string).
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
action = request.DATA['action']
|
||||
data = request.DATA['data']
|
||||
|
||||
if action == 'delete':
|
||||
for role_id in data:
|
||||
api.keystone.role_delete(request, role_id)
|
||||
elif action == 'create':
|
||||
new_role = api.keystone.role_create(request, data['name'])
|
||||
return rest_utils.CreatedResponse(
|
||||
'/api/keystone/roles/%s' % new_role.id,
|
||||
new_role.to_dict()
|
||||
)
|
||||
elif action == 'grant':
|
||||
api.keystone.add_tenant_user_role(
|
||||
request,
|
||||
data['project_id'],
|
||||
data['user_id'],
|
||||
data['role_id']
|
||||
)
|
||||
else:
|
||||
raise rest_utils.AjaxError(400, 'invalid (unrecognised) action')
|
||||
|
||||
|
||||
@urls.register
|
||||
class Role(generic.View):
|
||||
"""API for a single role.
|
||||
"""
|
||||
url_regex = r'keystone/roles/(?P<id>[0-9a-f]+|default)$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request, id):
|
||||
"""Get a specific role by id.
|
||||
|
||||
If the id supplied is 'default' then the default role will be
|
||||
returned, otherwise the role specified by the id.
|
||||
"""
|
||||
if id == 'default':
|
||||
return api.keystone.get_default_role(request).to_dict()
|
||||
return api.keystone.role_get(request, id).to_dict()
|
||||
|
||||
@rest_utils.ajax()
|
||||
def delete(self, request, id):
|
||||
"""Delete a single role by id.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
if id == 'default':
|
||||
raise django.http.HttpResponseNotFound('default')
|
||||
api.keystone.role_delete(request, id)
|
||||
|
||||
@rest_utils.ajax(method='PUT')
|
||||
def put(self, request, id):
|
||||
"""Update a single role.
|
||||
|
||||
The PUT data should be an application/json object with the "name"
|
||||
attribute to update
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
api.keystone.role_update(request, id, request.DATA['name'])
|
||||
|
||||
|
||||
@urls.register
|
||||
class Domains(generic.View):
|
||||
"""API over all domains.
|
||||
"""
|
||||
url_regex = r'keystone/domains/$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request):
|
||||
"""Get a list of domains.
|
||||
|
||||
A listing of all domains are returned.
|
||||
|
||||
The listing result is an object with property "items".
|
||||
"""
|
||||
items = [d.to_dict() for d in api.keystone.domain_list(request)]
|
||||
return {'items': items}
|
||||
|
||||
@rest_utils.ajax(method='POST')
|
||||
def post(self, request):
|
||||
"""Perform some action on the collection of domains.
|
||||
|
||||
The POST data should be an application/json object with two
|
||||
parameters: "action" and "data".
|
||||
|
||||
action = "delete"
|
||||
This action deletes multiple domains in one call, using the list of
|
||||
ids (strings) passed in as data.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
|
||||
action = "create"
|
||||
This action creates a domain using parameters supplied in the
|
||||
"data" object. The "name" (string) parameter is required, others
|
||||
are optional: "description" (string) and "enabled" (boolean,
|
||||
defaults to true).
|
||||
|
||||
This method returns the new domain object on success.
|
||||
"""
|
||||
action = request.DATA['action']
|
||||
data = request.DATA['data']
|
||||
|
||||
if action == 'delete':
|
||||
for domain_id in data:
|
||||
api.keystone.domain_delete(request, domain_id)
|
||||
elif action == 'create':
|
||||
new_domain = api.keystone.domain_create(
|
||||
request,
|
||||
data['name'],
|
||||
description=data.get('description'),
|
||||
enabled=data.get('enabled', True),
|
||||
)
|
||||
return rest_utils.CreatedResponse(
|
||||
'/api/keystone/domains/%s' % new_domain.id,
|
||||
new_domain.to_dict()
|
||||
)
|
||||
else:
|
||||
raise rest_utils.AjaxError(400, 'invalid action')
|
||||
|
||||
|
||||
@urls.register
|
||||
class Domain(generic.View):
|
||||
"""API over a single domains.
|
||||
"""
|
||||
url_regex = r'keystone/domains/(?P<id>[0-9a-f]+|default)$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request, id):
|
||||
"""Get a specific domain by id.
|
||||
|
||||
If the id supplied is 'default' then the default domain will be
|
||||
returned, otherwise the domain specified by the id.
|
||||
"""
|
||||
if id == 'default':
|
||||
return api.keystone.get_default_domain(request).to_dict()
|
||||
return api.keystone.domain_get(request, id).to_dict()
|
||||
|
||||
@rest_utils.ajax()
|
||||
def delete(self, request, id):
|
||||
"""Delete a single domain by id.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
if id == 'default':
|
||||
raise django.http.HttpResponseNotFound('default')
|
||||
api.keystone.domain_delete(request, id)
|
||||
|
||||
@rest_utils.ajax()
|
||||
def put(self, request, id):
|
||||
"""Update a single domain.
|
||||
|
||||
The PUT data should be an application/json object with the attributes
|
||||
to set to new values: "name" (string), "description" (string) and
|
||||
"enabled" (boolean).
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
api.keystone.domain_update(
|
||||
request,
|
||||
id,
|
||||
description=request.DATA.get('description'),
|
||||
enabled=request.DATA.get('enabled'),
|
||||
name=request.DATA.get('name')
|
||||
)
|
||||
|
||||
|
||||
def _tenant_kwargs_from_DATA(data, enabled=True):
|
||||
# tenant_create takes arbitrary keyword arguments with only a small
|
||||
# restriction (the default args)
|
||||
kwargs = {'name': None, 'description': None, 'enabled': enabled,
|
||||
'domain': data.pop('domain_id', None)}
|
||||
kwargs.update(data)
|
||||
return kwargs
|
||||
|
||||
|
||||
@urls.register
|
||||
class Projects(generic.View):
|
||||
"""API over all projects.
|
||||
|
||||
Note that in the following "project" is used exclusively where in the
|
||||
underlying keystone API the terms "project" and "tenant" are used
|
||||
interchangeably.
|
||||
"""
|
||||
url_regex = r'keystone/projects/$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request):
|
||||
"""Get a list of projects.
|
||||
|
||||
By default a listing of all projects for the current domain are
|
||||
returned.
|
||||
|
||||
You may specify GET parameters for project_id (string), user_id
|
||||
(string) and admin (boolean) to change that listing's context.
|
||||
Additionally, paginate (boolean) and marker may be used to get
|
||||
paginated listings.
|
||||
|
||||
The listing result is an object with properties:
|
||||
|
||||
items
|
||||
The list of project objects.
|
||||
has_more
|
||||
Boolean indicating there are more results when pagination is used.
|
||||
"""
|
||||
result, has_more = api.keystone.tenant_list(
|
||||
request,
|
||||
paginate=request.GET.get('paginate', False),
|
||||
marker=request.GET.get('marker'),
|
||||
domain=request.GET.get('domain_id'),
|
||||
user=request.GET.get('user_id'),
|
||||
admin=request.GET.get('admin', True)
|
||||
)
|
||||
# return (list of results, has_more_data)
|
||||
return dict(has_more=has_more, items=[d.to_dict() for d in result])
|
||||
|
||||
@rest_utils.ajax(method='POST')
|
||||
def post(self, request):
|
||||
"""Perform some action on the collection of projects (tenants).
|
||||
|
||||
The POST data should be an application/json object with two
|
||||
parameters: "action" and "data".
|
||||
|
||||
action = "delete"
|
||||
This action deletes multiple projects in one call, using the list
|
||||
of ids (strings) passed in as data.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
|
||||
action = "create"
|
||||
This action creates a project using parameters supplied in the
|
||||
"data" object. The "name" (string) parameter is required, others
|
||||
are optional: "description" (string), "domain_id" (string) and
|
||||
"enabled" (boolean, defaults to true). Additional, undefined
|
||||
parameters may also be provided, but you'll have to look deep into
|
||||
keystone to figure out what they might be.
|
||||
|
||||
This method returns the new project object on success.
|
||||
"""
|
||||
action = request.DATA['action']
|
||||
data = request.DATA['data']
|
||||
|
||||
if action == 'delete':
|
||||
for id in data:
|
||||
api.keystone.tenant_delete(request, id)
|
||||
elif action == 'create':
|
||||
kwargs = _tenant_kwargs_from_DATA(data)
|
||||
if not kwargs['name']:
|
||||
raise rest_utils.AjaxError(400, '"name" is required')
|
||||
new_project = api.keystone.tenant_create(
|
||||
request,
|
||||
kwargs.pop('name'),
|
||||
**kwargs
|
||||
)
|
||||
return rest_utils.CreatedResponse(
|
||||
'/api/keystone/projects/%s' % new_project.id,
|
||||
new_project.to_dict()
|
||||
)
|
||||
else:
|
||||
raise rest_utils.AjaxError(400, 'invalid action')
|
||||
|
||||
|
||||
@urls.register
|
||||
class Project(generic.View):
|
||||
"""API over a single project.
|
||||
|
||||
Note that in the following "project" is used exclusively where in the
|
||||
underlying keystone API the terms "project" and "tenant" are used
|
||||
interchangeably.
|
||||
"""
|
||||
url_regex = r'keystone/projects/$'
|
||||
|
||||
@rest_utils.ajax()
|
||||
def get(self, request, id):
|
||||
"""Get a specific project by id.
|
||||
"""
|
||||
return api.keystone.tenant_get(request, id).to_dict()
|
||||
|
||||
@rest_utils.ajax()
|
||||
def delete(self, request, id):
|
||||
"""Delete a single project by id.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
api.keystone.tenant_delete(request, id)
|
||||
|
||||
@rest_utils.ajax()
|
||||
def put(self, request, id):
|
||||
"""Update a single project.
|
||||
|
||||
The PUT data should be an application/json object with the attributes
|
||||
to set to new values: "name" (string), "description" (string),
|
||||
"domain_id" (string) and "enabled" (boolean). Additional, undefined
|
||||
parameters may also be provided, but you'll have to look deep into
|
||||
keystone to figure out what they might be.
|
||||
|
||||
This method returns HTTP 204 (no content) on success.
|
||||
"""
|
||||
kwargs = _tenant_kwargs_from_DATA(request.DATA, enabled=None)
|
||||
api.keystone.tenant_update(request, id, **kwargs)
|
||||
542
openstack_dashboard/test/api_tests/keystone_rest_tests.py
Normal file
542
openstack_dashboard/test/api_tests/keystone_rest_tests.py
Normal file
@ -0,0 +1,542 @@
|
||||
# Copyright 2014, Rackspace, US, Inc.
|
||||
#
|
||||
# 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 unittest2
|
||||
|
||||
from django.conf import settings
|
||||
|
||||
from openstack_dashboard.api.rest import keystone
|
||||
|
||||
from rest_test_utils import construct_request # noqa
|
||||
|
||||
|
||||
class KeystoneRestTestCase(unittest2.TestCase):
|
||||
def assertStatusCode(self, response, expected_code):
|
||||
if response.status_code == expected_code:
|
||||
return
|
||||
self.fail('status code %r != %r: %s' % (response.status_code,
|
||||
expected_code,
|
||||
response.content))
|
||||
|
||||
#
|
||||
# Users
|
||||
#
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_get(self, kc):
|
||||
request = construct_request()
|
||||
kc.user_get.return_value.to_dict.return_value = {'name': 'Ni!'}
|
||||
response = keystone.User().get(request, 'the_id')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.user_get.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_get_current(self, kc):
|
||||
request = construct_request(**{'user.id': 'current_id'})
|
||||
kc.user_get.return_value.to_dict.return_value = {'name': 'Ni!'}
|
||||
response = keystone.User().get(request, 'current')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.user_get.assert_called_once_with(request, 'current_id')
|
||||
kc.user_get.assert_not_called()
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_get_list(self, kc):
|
||||
request = construct_request(**{
|
||||
'session.get': mock.Mock(return_value='the_domain'),
|
||||
'GET': {},
|
||||
})
|
||||
kc.user_list.return_value = [
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})
|
||||
]
|
||||
response = keystone.Users().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content,
|
||||
'{"items": [{"name": "Ni!"}, {"name": "Ptang!"}]}')
|
||||
kc.user_list.assert_called_once_with(request, project=None,
|
||||
domain='the_domain', group=None)
|
||||
|
||||
def test_user_create_full(self):
|
||||
self._test_user_create(
|
||||
'{"action": "create", "data": {"name": "bob", '
|
||||
'"password": "sekrit", "project_id": "project123", '
|
||||
'"email": "spam@company.example"}}',
|
||||
{
|
||||
'name': 'bob',
|
||||
'password': 'sekrit',
|
||||
'email': 'spam@company.example',
|
||||
'project': 'project123',
|
||||
'domain': 'the_domain',
|
||||
'enabled': True
|
||||
}
|
||||
)
|
||||
|
||||
def test_user_create_existing_role(self):
|
||||
self._test_user_create(
|
||||
'{"action": "create", "data": {"name": "bob", '
|
||||
'"password": "sekrit", "project_id": "project123", '
|
||||
'"email": "spam@company.example"}}',
|
||||
{
|
||||
'name': 'bob',
|
||||
'password': 'sekrit',
|
||||
'email': 'spam@company.example',
|
||||
'project': 'project123',
|
||||
'domain': 'the_domain',
|
||||
'enabled': True
|
||||
}
|
||||
)
|
||||
|
||||
def test_user_create_partial(self):
|
||||
self._test_user_create(
|
||||
'{"action": "create", "data": {"name": "bob"}}',
|
||||
{
|
||||
'name': 'bob',
|
||||
'password': None,
|
||||
'email': None,
|
||||
'project': None,
|
||||
'domain': 'the_domain',
|
||||
'enabled': True
|
||||
}
|
||||
)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def _test_user_create(self, supplied_body, expected_call, kc):
|
||||
request = construct_request(body=supplied_body)
|
||||
kc.get_default_domain.return_value = mock.Mock(**{'id': 'the_domain'})
|
||||
kc.user_create.return_value.id = 'user123'
|
||||
kc.user_create.return_value = mock.Mock(**{
|
||||
'id': 'user123',
|
||||
'to_dict.return_value': {'id': 'user123', 'name': 'bob'}
|
||||
})
|
||||
|
||||
response = keystone.Users().post(request)
|
||||
self.assertStatusCode(response, 201)
|
||||
self.assertEqual(response['location'],
|
||||
'/api/keystone/users/user123')
|
||||
self.assertEqual(response.content, '{"id": "user123", '
|
||||
'"name": "bob"}')
|
||||
kc.user_create.assert_called_once_with(request, **expected_call)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_delete_many(self, kc):
|
||||
request = construct_request(body='''
|
||||
{
|
||||
"action": "delete",
|
||||
"data": ["id1", "id2", "id3"]
|
||||
}
|
||||
''')
|
||||
|
||||
response = keystone.Users().post(request)
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.user_delete.assert_has_mock.calls([
|
||||
mock.call(request, 'id1'),
|
||||
mock.call(request, 'id2'),
|
||||
mock.call(request, 'id3'),
|
||||
])
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_delete(self, kc):
|
||||
request = construct_request()
|
||||
response = keystone.User().delete(request, 'the_id')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.user_delete.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_put_password(self, kc):
|
||||
request = construct_request(body='''
|
||||
{"password": "sekrit"}
|
||||
''')
|
||||
response = keystone.User().put(request, 'user123')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.user_update_password.assert_called_once_with(request,
|
||||
'user123',
|
||||
password='sekrit')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_put_enabled(self, kc):
|
||||
request = construct_request(body='''
|
||||
{"enabled": false}
|
||||
''')
|
||||
response = keystone.User().put(request, 'user123')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.user_update_enabled.assert_called_once_with(request,
|
||||
'user123',
|
||||
enabled=False)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_put_project(self, kc):
|
||||
request = construct_request(body='''
|
||||
{"project_id": "other123"}
|
||||
''')
|
||||
response = keystone.User().put(request, 'user123')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.user_update_tenant.assert_called_once_with(request,
|
||||
'user123',
|
||||
project='other123')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_user_put_multiple(self, kc):
|
||||
request = construct_request(body='''
|
||||
{"project_id": "other123", "enabled": false}
|
||||
''')
|
||||
response = keystone.User().put(request, 'user123')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.user_update.assert_called_once_with(request,
|
||||
'user123',
|
||||
enabled=False,
|
||||
password=None,
|
||||
project='other123')
|
||||
|
||||
#
|
||||
# Roles
|
||||
#
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_get(self, kc):
|
||||
request = construct_request()
|
||||
kc.role_get.return_value.to_dict.return_value = {'name': 'Ni!'}
|
||||
response = keystone.Role().get(request, 'the_id')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.role_get.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_get_default(self, kc):
|
||||
request = construct_request()
|
||||
kc.get_default_role.return_value.to_dict.return_value = {'name': 'Ni!'}
|
||||
response = keystone.Role().get(request, 'default')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.get_default_role.assert_called_once_with(request)
|
||||
kc.role_get.assert_not_called()
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_get_list(self, kc):
|
||||
request = construct_request(**{'GET': {}})
|
||||
kc.role_list.return_value = [
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})
|
||||
]
|
||||
response = keystone.Roles().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content,
|
||||
'{"items": [{"name": "Ni!"}, {"name": "Ptang!"}]}')
|
||||
kc.role_list.assert_called_once_with(request)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_get_for_user(self, kc):
|
||||
request = construct_request(**{'GET': {'user_id': 'user123',
|
||||
'project_id': 'project123'}})
|
||||
kc.roles_for_user.return_value = [
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})
|
||||
]
|
||||
response = keystone.Roles().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content,
|
||||
'{"items": [{"name": "Ni!"}, {"name": "Ptang!"}]}')
|
||||
kc.roles_for_user.assert_called_once_with(request, 'user123',
|
||||
'project123')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_create(self, kc):
|
||||
request = construct_request(body='''
|
||||
{"action": "create", "data": {"name": "bob"}}
|
||||
''')
|
||||
kc.role_create.return_value.id = 'role123'
|
||||
kc.role_create.return_value.to_dict.return_value = {
|
||||
'id': 'role123', 'name': 'bob'
|
||||
}
|
||||
|
||||
response = keystone.Roles().post(request)
|
||||
self.assertStatusCode(response, 201)
|
||||
self.assertEqual(response['location'],
|
||||
'/api/keystone/roles/role123')
|
||||
self.assertEqual(response.content, '{"id": "role123", "name": "bob"}')
|
||||
kc.role_create.assert_called_once_with(request, 'bob')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_grant(self, kc):
|
||||
request = construct_request(body='''
|
||||
{"action": "grant", "data": {"user_id": "user123",
|
||||
"role_id": "role123", "project_id": "project123"}}
|
||||
''')
|
||||
response = keystone.Roles().post(request)
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.add_tenant_user_role.assert_called_once_with(request, 'project123',
|
||||
'user123', 'role123')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_delete_many(self, kc):
|
||||
request = construct_request(body='''
|
||||
{
|
||||
"action": "delete",
|
||||
"data": ["id1", "id2", "id3"]
|
||||
}
|
||||
''')
|
||||
|
||||
response = keystone.Roles().post(request)
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.role_delete.assert_has_mock.calls([
|
||||
mock.call(request, 'id1'),
|
||||
mock.call(request, 'id2'),
|
||||
mock.call(request, 'id3'),
|
||||
])
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_delete(self, kc):
|
||||
request = construct_request()
|
||||
response = keystone.Role().delete(request, 'the_id')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.role_delete.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_role_put(self, kc):
|
||||
request = construct_request(body='{"name": "spam"}')
|
||||
response = keystone.Role().put(request, 'the_id')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.role_update.assert_called_once_with(request,
|
||||
'the_id',
|
||||
'spam')
|
||||
|
||||
#
|
||||
# Domains
|
||||
#
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_domain_get(self, kc):
|
||||
request = construct_request()
|
||||
kc.domain_get.return_value.to_dict.return_value = {'name': 'Ni!'}
|
||||
response = keystone.Domain().get(request, 'the_id')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.domain_get.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_domain_get_default(self, kc):
|
||||
request = construct_request()
|
||||
kc.get_default_domain.return_value.to_dict.return_value = {
|
||||
'name': 'Ni!'
|
||||
}
|
||||
response = keystone.Domain().get(request, 'default')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.get_default_domain.assert_called_once_with(request)
|
||||
kc.domain_get.assert_not_called()
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_domain_get_list(self, kc):
|
||||
request = construct_request()
|
||||
kc.domain_list.return_value = [
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})
|
||||
]
|
||||
response = keystone.Domains().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content,
|
||||
'{"items": [{"name": "Ni!"}, {"name": "Ptang!"}]}')
|
||||
kc.domain_list.assert_called_once_with(request)
|
||||
|
||||
def test_domain_create_full(self):
|
||||
self._test_domain_create(
|
||||
'{"action": "create", "data": {"name": "bob", '
|
||||
'"description": "sekrit", "enabled": false}}',
|
||||
{
|
||||
'description': 'sekrit',
|
||||
'enabled': False
|
||||
}
|
||||
)
|
||||
|
||||
def test_domain_create_partial(self):
|
||||
self._test_domain_create(
|
||||
'{"action": "create", "data": {"name": "bob"}}',
|
||||
{
|
||||
'description': None,
|
||||
'enabled': True
|
||||
}
|
||||
)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def _test_domain_create(self, supplied_body, expected_call, kc):
|
||||
request = construct_request(body=supplied_body)
|
||||
kc.domain_create.return_value.id = 'domain123'
|
||||
kc.domain_create.return_value.to_dict.return_value = {
|
||||
'id': 'domain123', 'name': 'bob'
|
||||
}
|
||||
|
||||
response = keystone.Domains().post(request)
|
||||
self.assertStatusCode(response, 201)
|
||||
self.assertEqual(response['location'],
|
||||
'/api/keystone/domains/domain123')
|
||||
self.assertEqual(response.content, '{"id": "domain123", '
|
||||
'"name": "bob"}')
|
||||
kc.domain_create.assert_called_once_with(request, 'bob',
|
||||
**expected_call)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_domain_delete_many(self, kc):
|
||||
request = construct_request(body='''
|
||||
{
|
||||
"action": "delete",
|
||||
"data": ["id1", "id2", "id3"]
|
||||
}
|
||||
''')
|
||||
|
||||
response = keystone.Domains().post(request)
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.domain_delete.assert_has_mock.calls([
|
||||
mock.call(request, 'id1'),
|
||||
mock.call(request, 'id2'),
|
||||
mock.call(request, 'id3'),
|
||||
])
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_domain_delete(self, kc):
|
||||
request = construct_request()
|
||||
response = keystone.Domain().delete(request, 'the_id')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.domain_delete.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_domain_put(self, kc):
|
||||
request = construct_request(body='{"name": "spam"}')
|
||||
response = keystone.Domain().put(request, 'the_id')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.domain_update.assert_called_once_with(request,
|
||||
'the_id',
|
||||
name='spam',
|
||||
description=None,
|
||||
enabled=None)
|
||||
|
||||
#
|
||||
# Projects
|
||||
#
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_project_get(self, kc):
|
||||
request = construct_request()
|
||||
kc.tenant_get.return_value.to_dict.return_value = {'name': 'Ni!'}
|
||||
response = keystone.Project().get(request, 'the_id')
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"name": "Ni!"}')
|
||||
kc.tenant_get.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_project_get_list(self, kc):
|
||||
request = construct_request(**{'GET': {}})
|
||||
kc.tenant_list.return_value = ([
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ni!'}}),
|
||||
mock.Mock(**{'to_dict.return_value': {'name': 'Ptang!'}})
|
||||
], False)
|
||||
with mock.patch.object(settings, 'DEBUG', True):
|
||||
response = keystone.Projects().get(request)
|
||||
self.assertStatusCode(response, 200)
|
||||
self.assertEqual(response.content, '{"has_more": false, '
|
||||
'"items": [{"name": "Ni!"}, {"name": "Ptang!"}]}')
|
||||
kc.tenant_list.assert_called_once_with(request, paginate=False,
|
||||
marker=None, domain=None,
|
||||
user=None, admin=True)
|
||||
|
||||
def test_project_create_full(self):
|
||||
self._test_project_create(
|
||||
'{"action": "create", "data": {"name": "bob", '
|
||||
'"domain_id": "domain123", "description": "sekrit", '
|
||||
'"enabled": false}}',
|
||||
{
|
||||
'description': 'sekrit',
|
||||
'domain': 'domain123',
|
||||
'enabled': False
|
||||
}
|
||||
)
|
||||
|
||||
def test_project_create_partial(self):
|
||||
self._test_project_create(
|
||||
'{"action": "create", "data": {"name": "bob"}}',
|
||||
{
|
||||
'description': None,
|
||||
'domain': None,
|
||||
'enabled': True
|
||||
}
|
||||
)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def _test_project_create(self, supplied_body, expected_call, kc):
|
||||
request = construct_request(body=supplied_body)
|
||||
kc.tenant_create.return_value.id = 'project123'
|
||||
kc.tenant_create.return_value.to_dict.return_value = {
|
||||
'id': 'project123', 'name': 'bob'
|
||||
}
|
||||
|
||||
response = keystone.Projects().post(request)
|
||||
self.assertStatusCode(response, 201)
|
||||
self.assertEqual(response['location'],
|
||||
'/api/keystone/projects/project123')
|
||||
self.assertEqual(response.content, '{"id": "project123", '
|
||||
'"name": "bob"}')
|
||||
kc.tenant_create.assert_called_once_with(request, 'bob',
|
||||
**expected_call)
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_project_delete_many(self, kc):
|
||||
request = construct_request(body='''
|
||||
{
|
||||
"action": "delete",
|
||||
"data": ["id1", "id2", "id3"]
|
||||
}
|
||||
''')
|
||||
|
||||
response = keystone.Projects().post(request)
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.tenant_delete.assert_has_mock.calls([
|
||||
mock.call(request, 'id1'),
|
||||
mock.call(request, 'id2'),
|
||||
mock.call(request, 'id3'),
|
||||
])
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_project_delete(self, kc):
|
||||
request = construct_request()
|
||||
response = keystone.Project().delete(request, 'the_id')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.tenant_delete.assert_called_once_with(request, 'the_id')
|
||||
|
||||
@mock.patch.object(keystone.api, 'keystone')
|
||||
def test_project_put(self, kc):
|
||||
# nothing in the Horizon code documents what additional parameters are
|
||||
# allowed, so we'll just assume GIGO
|
||||
request = construct_request(body='''
|
||||
{"name": "spam", "domain_id": "domain123", "foo": "bar"}
|
||||
''')
|
||||
response = keystone.Project().put(request, 'spam123')
|
||||
self.assertStatusCode(response, 204)
|
||||
self.assertEqual(response.content, '')
|
||||
kc.tenant_update.assert_called_once_with(request,
|
||||
'spam123',
|
||||
name='spam', foo='bar',
|
||||
description=None,
|
||||
domain='domain123',
|
||||
enabled=None)
|
||||
Loading…
Reference in New Issue
Block a user