openstack
/
keystone
Code Issues Proposed changes

Changes to support CRUD on services/roles. 

Change-Id: I5fd653b351bed115942fbbc9845b8f413065adcd
This commit is contained in:
Yogeshwar Srikrishnan 2011-08-05 13:44:01 -05:00
parent 8b316b7130
commit cea7fca224
22 changed files with 1045 additions and 187 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
.pylintrc
View File

@ -1,4 +1,8 @@
# The format of this file isn't really documented; just use --generate-rcfile
[MASTER]
# Add <file or directory> to the black list. It should be a base name, not a
# path. You may set this option multiple times.
ignore=test
[Messages Control]
# NOTE(justinsb): We might want to have a 2nd strict pylintrc in future
@ -19,8 +23,8 @@ argument-rgx=[a-z_][a-z0-9_]{1,30}$
# and be lowecased with underscores
method-rgx=([a-z_][a-z0-9_]{2,50}|setUp|tearDown)$
# Module names matching nova-* are ok (files in bin/)
module-rgx=(([a-z_][a-z0-9_]*)|([A-Z][a-zA-Z0-9]+)|(nova-[a-z0-9_-]+))$
# Module names matching keystone-* are ok (files in bin/)
module-rgx=(([a-z_][a-z0-9_]*)|([A-Z][a-zA-Z0-9]+)|(keystone-[a-z0-9_-]+))$
# Don't require docstrings on tests.
no-docstring-rgx=((__.*__)|([tT]est.*)|setUp|tearDown)$

67
bin/keystone-manage
View File

@ -78,7 +78,7 @@ def Main():
parser.error('No object type specified for first argument')
object_type = args[0]
if object_type in ['user', 'tenant', 'role', 'endpointTemplates', 'token',
if object_type in ['user', 'tenant', 'role', 'service' , 'endpointTemplates', 'token',
'endpoint']:
pass
else:
@ -125,7 +125,7 @@ def Main():
if len(args) > 4:
tenant = args[4]
object.tenant_id = tenant
db_api.user.create(object)
db_api.USER.create(object)
print "SUCCESS: User %s created." % object.id
except Exception as exc:
raise Exception("Failed to create user %s" % (object_id,),
@ -133,11 +133,11 @@ def Main():
return
elif command == "disable":
try:
object = db_api.user.get(object_id)
object = db_api.USER.get(object_id)
if object == None:
raise IndexError("User %s not found" % object_id)
object.enabled = False
db_api.user.update(object_id, object)
db_api.USER.update(object_id, object)
print "SUCCESS: User %s disabled." % object.id
except Exception as exc:
raise Exception("Failed to disable user %s" % (object_id,),
@ -147,7 +147,7 @@ def Main():
try:
if len(args) > 2:
tenant = args[2]
objects = db_api.user.get_by_tenant(tenant)
objects = db_api.USER.get_by_tenant(tenant)
if objects == None:
raise IndexError("Users not found")
print 'id', 'enabled'
@ -155,7 +155,7 @@ def Main():
for row in objects:
print row.id, row.enabled
else:
objects = db_api.user.get_all()
objects = db_api.USER.get_all()
if objects == None:
raise IndexError("Users not found")
print 'id', 'enabled', 'tenant'
@ -171,7 +171,7 @@ def Main():
object = db_models.Tenant()
object.id = object_id
object.enabled = True
db_api.tenant.create(object)
db_api.TENANT.create(object)
print "SUCCESS: Tenant %s created." % object.id
return
except Exception as exc:
@ -179,7 +179,7 @@ def Main():
sys.exc_info())
elif command == "list":
try:
objects = db_api.tenant.get_all()
objects = db_api.TENANT.get_all()
if objects == None:
raise IndexError("Tenants not found")
print 'tenant', 'enabled'
@ -191,11 +191,11 @@ def Main():
return
elif command == "disable":
try:
object = db_api.tenant.get(object_id)
object = db_api.TENANT.get(object_id)
if object == None:
raise IndexError("Tenant %s not found" % object_id)
object.enabled = False
db_api.tenant.update(object_id, object)
db_api.TENANT.update(object_id, object)
print "SUCCESS: Tenant %s disabled." % object.id
except Exception as exc:
raise Exception("Failed to disable tenant %s" % (object_id,),
@ -206,7 +206,7 @@ def Main():
try:
object = db_models.Role()
object.id = object_id
db_api.role.create(object)
db_api.ROLE.create(object)
print "SUCCESS: Role %s created successfully." % object.id
return
except Exception as exc:
@ -216,7 +216,7 @@ def Main():
if len(args) == 3:
tenant = args[2]
try:
objects = db_api.tenant.get_role_assignments(tenant)
objects = db_api.TENANT.get_role_assignments(tenant)
if objects == None:
raise IndexError("Assignments not found")
print 'Role assignments for tenant %s' % tenant
@ -232,7 +232,7 @@ def Main():
else:
tenant = None
try:
objects = db_api.role.get_all()
objects = db_api.ROLE.get_all()
if objects == None:
raise IndexError("Roles not found")
print 'All roles'
@ -258,7 +258,7 @@ def Main():
object.user_id = user
if tenant != None:
object.tenant_id = tenant
db_api.user.user_role_add(object)
db_api.USER.user_role_add(object)
print "SUCCESS: Granted %s the %s role on %s." % \
(object.user_id, object.role_id, object.tenant_id)
except Exception as exc:
@ -288,7 +288,7 @@ def Main():
object.internal_url = internal_url
object.enabled = enabled
object.is_global = is_global
object = db_api.endpoint_template.create(object)
object = db_api.ENDPOINT_TEMPLATE.create(object)
print "SUCCESS: Created EndpointTemplates for %s pointing " \
"to %s." % (object.service, object.public_url)
return
@ -299,7 +299,7 @@ def Main():
if len(args) == 3:
tenant = args[2]
try:
objects = db_api.endpoint_template.endpoint_get_by_tenant(
objects = db_api.ENDPOINT_TEMPLATE.endpoint_get_by_tenant(
tenant)
if objects == None:
raise IndexError("URLs not found")
@ -315,7 +315,7 @@ def Main():
else:
tenant = None
try:
objects = db_api.endpoint_template.get_all()
objects = db_api.ENDPOINT_TEMPLATE.get_all()
if objects == None:
raise IndexError("URLs not found")
print 'All EndpointTemplates'
@ -339,7 +339,7 @@ def Main():
object = db_models.Endpoints()
object.tenant_id = tenant_id
object.endpoint_template_id = endpoint_template_id
object = db_api.endpoint_template.endpoint_add(object)
object = db_api.ENDPOINT_TEMPLATE.endpoint_add(object)
print "SUCCESS: Endpoint %s added to tenant %s." % \
(endpoint_template_id, tenant_id)
return
@ -359,7 +359,7 @@ def Main():
.replace("-", ""),
"%Y%m%dT%H:%M")
object.expires = tuple_time
db_api.token.create(object)
db_api.TOKEN.create(object)
print "SUCCESS: Token %s created." % object.id
return
except Exception as exc:
@ -367,7 +367,7 @@ def Main():
sys.exc_info())
elif command == "list":
try:
objects = db_api.token.get_all()
objects = db_api.TOKEN.get_all()
if objects == None:
raise IndexError("Tokens not found")
print 'token', 'user', 'expiration', 'tenant'
@ -379,16 +379,39 @@ def Main():
return
elif command == "delete":
try:
object = db_api.token.get(object_id)
object = db_api.TOKEN.get(object_id)
if object == None:
raise IndexError("Token %s not found" % object_id)
else:
db_api.token.delete(object_id)
db_api.TOKEN.delete(object_id)
print 'SUCCESS: Token %s deleted.' % object_id
except Exception, e:
raise Exception("Failed to delete token %s" % (object_id,),
sys.exc_info())
return
elif object_type == "service":
if command == "add":
try:
object = db_models.Service()
object.id = object_id
db_api.SERVICE.create(object)
print "SUCCESS: Service %s created successfully." % object.id
return
except Exception as exc:
raise Exception("Failed to create Service %s" % (object_id,), sys.exc_info())
elif command == "list":
try:
objects = db_api.SERVICE.get_all()
if objects == None:
raise IndexError("Services not found")
print objects
print 'All Services'
print 'Service'
print '-' * 20
for row in objects:
print row.id
except Exception, e:
raise Exception("Error getting all services", sys.exc_info())
# Command not handled
print ("ERROR: %s %s not yet supported" % (object_type, command))

7
bin/sampledata.sh
View File

@ -28,12 +28,15 @@
`dirname $0`/keystone-manage $* user add joeuser secrete 1234
`dirname $0`/keystone-manage $* user add joeadmin secrete 1234
`dirname $0`/keystone-manage $* user add admin secrete 1234
`dirname $0`/keystone-manage $* user add serviceadmin secrete 1234
`dirname $0`/keystone-manage $* user add disabled secrete 1234
`dirname $0`/keystone-manage $* user disable disabled
# Roles
`dirname $0`/keystone-manage $* role add Admin
`dirname $0`/keystone-manage $* role add KeystoneServiceAdmin
`dirname $0`/keystone-manage $* role grant Admin admin
`dirname $0`/keystone-manage $* role grant KeystoneServiceAdmin serviceadmin
`dirname $0`/keystone-manage $* role grant Admin joeadmin 1234
`dirname $0`/keystone-manage $* role grant Admin joeadmin ANOTHER:TENANT
@ -60,6 +63,7 @@
# Tokens
`dirname $0`/keystone-manage $* token add 887665443383838 joeuser 1234 2012-02-05T00:00
`dirname $0`/keystone-manage $* token add 999888777666 admin 1234 2015-02-05T00:00
`dirname $0`/keystone-manage $* token add 111222333444 serviceadmin 1234 2015-02-05T00:00
`dirname $0`/keystone-manage $* token add 000999 admin 1234 2010-02-05T00:00
`dirname $0`/keystone-manage $* token add 999888777 disabled 1234 2015-02-05T00:00
@ -69,3 +73,6 @@
`dirname $0`/keystone-manage $* endpoint add 1234 3
`dirname $0`/keystone-manage $* endpoint add 1234 4
`dirname $0`/keystone-manage $* endpoint add 1234 5
#Add Services
`dirname $0`/keystone-manage $* service add exampleservice

5
etc/keystone.conf
View File

@ -41,13 +41,16 @@ admin_port = 5001
#Role that allows to perform admin operations.
keystone-admin-role = Admin
#Role that allows to perform service admin operations.
keystone-service-admin-role = KeystoneServiceAdmin
[keystone.backends.sqlalchemy]
# SQLAlchemy connection string for the reference implementation registry
# server. Any valid SQLAlchemy connection string is fine.
# See: http://bit.ly/ideIpI
sql_connection = sqlite:///keystone.db
backend_entities = ['UserRoleAssociation', 'Endpoints',
'Role', 'Tenant', 'User', 'Credentials', 'EndpointTemplates', 'Token']
'Role', 'Tenant', 'User', 'Credentials', 'EndpointTemplates', 'Token','Service']
# Period in seconds after which SQLAlchemy should reestablish its connection
# to the database.

12
keystone/backends/__init__.py
View File

@ -17,13 +17,15 @@
import ast
import logging
import keystone.utils as utils
from keystone.backends import api, models
from keystone.backends import models as models
from keystone.backends import api as api
DEFAULT_BACKENDS = 'keystone.backends.sqlalchemy'
#Configs applicable to all backends.
#Reference to Admin Role.
KeyStoneAdminRole = None
KEYSTONEADMINROLE = None
KEYSTONESERVICEADMINROLE = None
def configure_backends(options):
@ -33,5 +35,7 @@ def configure_backends(options):
backend_module = utils.import_module(backend)
backend_module.configure_backend(options[backend])
#Initialize common configs general to all backends.
global KeyStoneAdminRole
KeyStoneAdminRole = options["keystone-admin-role"]
global KEYSTONEADMINROLE
KEYSTONEADMINROLE = options["keystone-admin-role"]
global KEYSTONESERVICEADMINROLE
KEYSTONESERVICEADMINROLE = options["keystone-service-admin-role"]

65
keystone/backends/api.py
View File

@ -183,6 +183,9 @@ class BaseRoleAPI(object):
def create(self, values):
raise NotImplementedError
def delete(self, id):
raise NotImplementedError
def get(self, id):
raise NotImplementedError
@ -275,37 +278,57 @@ class BaseEndpointTemplateAPI(object):
def endpoint_delete(self, id):
raise NotImplementedError
class BaseServiceAPI:
def create(self, values):
raise NotImplementedError
def get(self, id):
raise NotImplementedError
def get_all(self):
raise NotImplementedError
def get_page(self, marker, limit):
raise NotImplementedError
def get_page_markers(self, marker, limit):
raise NotImplementedError
#API
#TODO(Yogi) Refactor all API to separate classes specific to models.
endpoint_template = BaseEndpointTemplateAPI()
group = BaseGroupAPI()
role = BaseRoleAPI()
tenant_group = BaseTenantGroupAPI()
tenant = BaseTenantAPI()
token = BaseTokenAPI()
user = BaseUserAPI()
ENDPOINT_TEMPLATE = BaseEndpointTemplateAPI()
GROUP = BaseGroupAPI()
ROLE = BaseRoleAPI()
TENANT_GROUP = BaseTenantGroupAPI()
TENANT = BaseTenantAPI()
TOKEN = BaseTokenAPI()
USER = BaseUserAPI()
SERVICE = BaseServiceAPI()
# Function to dynamically set module references.
def set_value(variable_name, value):
if variable_name == 'endpoint_template':
global endpoint_template
endpoint_template = value
global ENDPOINT_TEMPLATE
ENDPOINT_TEMPLATE = value
elif variable_name == 'group':
global group
group = value
global GROUP
GROUP = value
elif variable_name == 'role':
global role
role = value
global ROLE
ROLE = value
elif variable_name == 'tenant_group':
global tenant_group
tenant_group = value
global TENANT_GROUP
TENANT_GROUP = value
elif variable_name == 'tenant':
global tenant
tenant = value
global TENANT
TENANT = value
elif variable_name == 'token':
global token
token = value
global TOKEN
TOKEN = value
elif variable_name == 'user':
global user
user = value
global USER
USER = value
elif variable_name == 'service':
global SERVICE
SERVICE = value

4
keystone/backends/models.py
View File

@ -23,6 +23,7 @@ User = None
Credentials = None
Token = None
EndpointTemplates = None
Service = None
# Function to dynamically set model references.
@ -51,3 +52,6 @@ def set_value(variable_name, value):
elif variable_name == 'EndpointTemplates':
global EndpointTemplates
EndpointTemplates = value
elif variable_name == 'Service':
global Service
Service = value

7
keystone/backends/sqlalchemy/api/role.py
View File

@ -26,6 +26,13 @@ class RoleAPI(BaseRoleAPI):
role_ref.save()
return role_ref
def delete(self, id, session=None):
if not session:
session = get_session()
with session.begin():
role_ref = self.get(id, session)
session.delete(role_ref)
def get(self, id, session=None):
if not session:
session = get_session()

103
keystone/backends/sqlalchemy/api/service.py Normal file
View File

@ -0,0 +1,103 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2010 OpenStack LLC.
# All Rights Reserved.
#
# 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.
from keystone.backends.sqlalchemy import get_session, models
from keystone.backends.api import BaseServiceAPI
class ServiceAPI(BaseServiceAPI):
def __init__(self):
pass
def create(self, values):
service_ref = models.Service()
service_ref.update(values)
service_ref.save()
return service_ref
def get(self, id, session=None):
if not session:
session = get_session()
result = session.query(models.Service).filter_by(id=id).first()
return result
def get_all(self, session=None):
print "Enter Get All Service"
if not session:
session = get_session()
return session.query(models.Service).all()
def get_page(self, marker, limit, session=None):
if not session:
session = get_session()
if marker:
return session.query(models.Service).filter("id>:marker").params(\
marker='%s' % marker).order_by(\
models.Service.id.desc()).limit(limit).all()
else:
return session.query(models.Service).order_by(\
models.Service.id.desc()).limit(limit).all()
def get_page_markers(self, marker, limit, session=None):
if not session:
session = get_session()
first = session.query(models.Service).order_by(\
models.Service.id).first()
last = session.query(models.Service).order_by(\
models.Service.id.desc()).first()
if first is None:
return (None, None)
if marker is None:
marker = first.id
next_page = session.query(models.Service).\
filter("id > :marker").params(\
marker='%s' % marker).order_by(\
models.Service.id).limit(limit).all()
prev_page = session.query(models.Service).\
filter("id < :marker").params(\
marker='%s' % marker).order_by(\
models.Service.id.desc()).limit(int(limit)).all()
if len(next_page) == 0:
next_page = last
else:
for t in next_page:
next_page = t
if len(prev_page) == 0:
prev_page = first
else:
for t in prev_page:
prev_page = t
if prev_page.id == marker:
prev_page = None
else:
prev_page = prev_page.id
if next_page.id == last.id:
next_page = None
else:
next_page = next_page.id
return (prev_page, next_page)
def delete(self, id, session=None):
if not session:
session = get_session()
with session.begin():
service_ref = self.get(id, session)
session.delete(service_ref)
def get():
return ServiceAPI()

10
keystone/backends/sqlalchemy/models.py
View File

@ -104,6 +104,16 @@ class Role(Base, KeystoneBase):
__api__ = 'role'
id = Column(String(255), primary_key=True, unique=True)
desc = Column(String(255))
service_id = Column(Integer, ForeignKey('services.id'))
__table_args__ = (
UniqueConstraint("id", "service_id"), {})
class Service(Base, KeystoneBase):
__tablename__ = 'services'
__api__ = 'service'
id = Column(String(255), primary_key=True, unique=True)
desc = Column(String(255))
class Tenant(Base, KeystoneBase):

5
keystone/controllers/roles.py
View File

@ -18,6 +18,11 @@ class RolesController(wsgi.Controller):
return utils.send_result(201, req,
config.SERVICE.create_role(utils.get_auth_token(req), role))
@utils.wrap_error
def delete_role(self, req, role_id):
rval = config.SERVICE.delete_role(utils.get_auth_token(req), role_id)
return utils.send_result(204, req, rval)
@utils.wrap_error
def get_roles(self, req):
marker, limit, url = get_marker_limit_and_url(req)

38
keystone/controllers/services.py Executable file
View File

@ -0,0 +1,38 @@
from keystone import utils
from keystone.common import wsgi
from keystone.logic.types.service import Service
import keystone.config as config
from . import get_marker_limit_and_url
class ServicesController(wsgi.Controller):
"""Controller for Service related operations"""
def __init__(self, options):
self.options = options
# Not exposed yet.
@utils.wrap_error
def create_service(self, req):
service = utils.get_normalized_request_content(Service, req)
return utils.send_result(201, req,
config.SERVICE.create_service(utils.get_auth_token(req), service))
@utils.wrap_error
def get_services(self, req):
marker, limit, url = get_marker_limit_and_url(req)
services = config.SERVICE.get_services(
utils.get_auth_token(req), marker, limit, url)
return utils.send_result(200, req, services)
@utils.wrap_error
def get_service(self, req, service_id):
service = config.SERVICE.get_service(
utils.get_auth_token(req), service_id)
return utils.send_result(200, req, service)
@utils.wrap_error
def delete_service(self, req, service_id):
rval = config.SERVICE.delete_service(utils.get_auth_token(req),
service_id)
return utils.send_result(204, req, rval)

228
keystone/logic/service.py
View File

@ -24,6 +24,7 @@ from keystone.logic.types import fault
from keystone.logic.types.tenant import \
Tenant, Tenants, User as TenantUser
from keystone.logic.types.role import Role, RoleRef, RoleRefs, Roles
from keystone.logic.types.service import Service, Services
from keystone.logic.types.user import User, User_Update, Users
from keystone.logic.types.endpoint import Endpoint, Endpoints, \
EndpointTemplate, EndpointTemplates
@ -42,11 +43,11 @@ class IdentityService(object):
raise fault.BadRequestFault("Expecting Password Credentials!")
if not credentials.tenant_id:
duser = api.user.get(credentials.username)
duser = api.USER.get(credentials.username)
if duser == None:
raise fault.UnauthorizedFault("Unauthorized")
else:
duser = api.user.get_by_tenant(credentials.username,
duser = api.USER.get_by_tenant(credentials.username,
credentials.tenant_id)
if duser == None:
raise fault.UnauthorizedFault("Unauthorized on this tenant")
@ -61,9 +62,9 @@ class IdentityService(object):
# TODO: Handle tenant/token search
#
if not credentials.tenant_id:
dtoken = api.token.get_for_user(duser.id)
dtoken = api.TOKEN.get_for_user(duser.id)
else:
dtoken = api.token.get_for_user_by_tenant(duser.id,
dtoken = api.TOKEN.get_for_user_by_tenant(duser.id,
credentials.tenant_id)
tenant_id = credentials.tenant_id or duser.tenant_id
@ -76,7 +77,7 @@ class IdentityService(object):
if credentials.tenant_id:
dtoken.tenant_id = credentials.tenant_id
dtoken.expires = datetime.now() + timedelta(days=1)
api.token.create(dtoken)
api.TOKEN.create(dtoken)
#if tenant_id is passed in the call that tenant_id is passed else
#user's default tenant_id is used.
return self.__get_auth_data(dtoken, tenant_id)
@ -84,7 +85,7 @@ class IdentityService(object):
def validate_token(self, admin_token, token_id, belongs_to=None):
self.__validate_admin_token(admin_token)
if not api.token.get(token_id):
if not api.TOKEN.get(token_id):
raise fault.UnauthorizedFault("Bad token, please reauthenticate")
(token, user) = self.__validate_token(token_id, belongs_to)
@ -94,11 +95,11 @@ class IdentityService(object):
def revoke_token(self, admin_token, token_id):
self.__validate_admin_token(admin_token)
dtoken = api.token.get(token_id)
dtoken = api.TOKEN.get(token_id)
if not dtoken:
raise fault.ItemNotFoundFault("Token not found")
api.token.delete(token_id)
api.TOKEN.delete(token_id)
#
# Tenant Operations
@ -113,7 +114,7 @@ class IdentityService(object):
if tenant.tenant_id == None:
raise fault.BadRequestFault("Expecting a unique Tenant Id")
if api.tenant.get(tenant.tenant_id) != None:
if api.TENANT.get(tenant.tenant_id) != None:
raise fault.TenantConflictFault(
"A tenant with that id already exists")
@ -122,7 +123,7 @@ class IdentityService(object):
dtenant.desc = tenant.description
dtenant.enabled = tenant.enabled
api.tenant.create(dtenant)
api.TENANT.create(dtenant)
return tenant
##
@ -133,11 +134,11 @@ class IdentityService(object):
(_token, user) = self.__validate_admin_token(admin_token)
# If Global admin return all
ts = []
dtenants = api.tenant.get_page(marker, limit)
dtenants = api.TENANT.get_page(marker, limit)
for dtenant in dtenants:
ts.append(Tenant(dtenant.id,
dtenant.desc, dtenant.enabled))
prev, next = api.tenant.get_page_markers(marker, limit)
prev, next = api.TENANT.get_page_markers(marker, limit)
links = []
if prev:
links.append(atom.Link('prev',
@ -150,12 +151,12 @@ class IdentityService(object):
#If not global admin ,return tenants specific to user.
(_token, user) = self.__validate_token(admin_token, False)
ts = []
dtenants = api.tenant.tenants_for_user_get_page(
dtenants = api.TENANT.tenants_for_user_get_page(
user, marker, limit)
for dtenant in dtenants:
ts.append(Tenant(dtenant.id,
dtenant.desc, dtenant.enabled))
prev, next = api.tenant.tenants_for_user_get_page_markers(
prev, next = api.TENANT.tenants_for_user_get_page_markers(
user, marker, limit)
links = []
if prev:
@ -169,7 +170,7 @@ class IdentityService(object):
def get_tenant(self, admin_token, tenant_id):
self.__validate_admin_token(admin_token)
dtenant = api.tenant.get(tenant_id)
dtenant = api.TENANT.get(tenant_id)
if not dtenant:
raise fault.ItemNotFoundFault("The tenant could not be found")
return Tenant(dtenant.id, dtenant.desc, dtenant.enabled)
@ -180,25 +181,25 @@ class IdentityService(object):
if not isinstance(tenant, Tenant):
raise fault.BadRequestFault("Expecting a Tenant")
dtenant = api.tenant.get(tenant_id)
dtenant = api.TENANT.get(tenant_id)
if dtenant == None:
raise fault.ItemNotFoundFault("The tenant could not be found")
values = {'desc': tenant.description, 'enabled': tenant.enabled}
api.tenant.update(tenant_id, values)
api.TENANT.update(tenant_id, values)
return Tenant(dtenant.id, tenant.description, tenant.enabled)
def delete_tenant(self, admin_token, tenant_id):
self.__validate_admin_token(admin_token)
dtenant = api.tenant.get(tenant_id)
dtenant = api.TENANT.get(tenant_id)
if dtenant == None:
raise fault.ItemNotFoundFault("The tenant could not be found")
if not api.tenant.is_empty(tenant_id):
if not api.TENANT.is_empty(tenant_id):
raise fault.ForbiddenFault("You may not delete a tenant that "
"contains get_users")
api.tenant.delete(dtenant.id)
api.TENANT.delete(dtenant.id)
return None
#
@ -210,9 +211,9 @@ class IdentityService(object):
token = None
user = None
if token_id:
token = api.token.get(token_id)
token = api.TOKEN.get(token_id)
if token:
user = api.user.get(token.user_id)
user = api.USER.get(token.user_id)
return (token, user)
#
@ -229,11 +230,11 @@ class IdentityService(object):
if user.user_id == None or len(user.user_id.strip()) == 0:
raise fault.BadRequestFault("Expecting a unique User Id")
if api.user.get(user.user_id) != None:
if api.USER.get(user.user_id) != None:
raise fault.UserConflictFault(
"An user with that id already exists")
if api.user.get_by_email(user.email) != None:
if api.USER.get_by_email(user.email) != None:
raise fault.EmailConflictFault("Email already exists")
duser = models.User()
@ -242,13 +243,13 @@ class IdentityService(object):
duser.email = user.email
duser.enabled = user.enabled
duser.tenant_id = user.tenant_id
api.user.create(duser)
api.USER.create(duser)
return user
def validate_and_fetch_user_tenant(self, tenant_id):
if tenant_id != None and len(tenant_id) > 0:
dtenant = api.tenant.get(tenant_id)
dtenant = api.TENANT.get(tenant_id)
if dtenant == None:
raise fault.ItemNotFoundFault("The tenant is not found")
elif not dtenant.enabled:
@ -263,20 +264,20 @@ class IdentityService(object):
if tenant_id == None:
raise fault.BadRequestFault("Expecting a Tenant Id")
dtenant = api.tenant.get(tenant_id)
dtenant = api.TENANT.get(tenant_id)
if dtenant is None:
raise fault.ItemNotFoundFault("The tenant not found")
if not dtenant.enabled:
raise fault.TenantDisabledFault("Your account has been disabled")
ts = []
dtenantusers = api.user.users_get_by_tenant_get_page(tenant_id, marker,
dtenantusers = api.USER.users_get_by_tenant_get_page(tenant_id, marker,
limit)
for dtenantuser in dtenantusers:
ts.append(User(None, dtenantuser.id, tenant_id,
dtenantuser.email, dtenantuser.enabled))
links = []
if ts.__len__():
prev, next = api.user.users_get_by_tenant_get_page_markers(
prev, next = api.USER.users_get_by_tenant_get_page_markers(
tenant_id, marker, limit)
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" %
@ -289,13 +290,13 @@ class IdentityService(object):
def get_users(self, admin_token, marker, limit, url):
self.__validate_admin_token(admin_token)
ts = []
dusers = api.user.users_get_page(marker, limit)
dusers = api.USER.users_get_page(marker, limit)
for duser in dusers:
ts.append(User(None, duser.id, duser.tenant_id,
duser.email, duser.enabled))
links = []
if ts.__len__():
prev, next = api.user.users_get_page_markers(marker, limit)
prev, next = api.USER.users_get_page_markers(marker, limit)
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" %
(url, prev, limit)))
@ -306,11 +307,11 @@ class IdentityService(object):
def get_user(self, admin_token, user_id):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
dtenant = api.tenant.get(duser.tenant_id)
dtenant = api.TENANT.get(duser.tenant_id)
ts = []
return User_Update(None, duser.id, duser.tenant_id,
@ -319,7 +320,7 @@ class IdentityService(object):
def update_user(self, admin_token, user_id, user):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
@ -328,92 +329,92 @@ class IdentityService(object):
raise fault.BadRequestFault("Expecting a User")
if user.email != duser.email and \
api.user.get_by_email(user.email) is not None:
api.USER.get_by_email(user.email) is not None:
raise fault.EmailConflictFault(
"Email already exists")
values = {'email': user.email}
api.user.update(user_id, values)
duser = api.user.user_get_update(user_id)
api.USER.update(user_id, values)
duser = api.USER.user_get_update(user_id)
return User(duser.password, duser.id, duser.tenant_id,
duser.email, duser.enabled)
def set_user_password(self, admin_token, user_id, user):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
if not isinstance(user, User):
raise fault.BadRequestFault("Expecting a User")
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if duser == None:
raise fault.ItemNotFoundFault("The user could not be found")
values = {'password': user.password}
api.user.update(user_id, values)
api.USER.update(user_id, values)
return User_Update(user.password,
None, None, None, None)
def enable_disable_user(self, admin_token, user_id, user):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
if not isinstance(user, User):
raise fault.BadRequestFault("Expecting a User")
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if duser == None:
raise fault.ItemNotFoundFault("The user could not be found")
values = {'enabled': user.enabled}
api.user.update(user_id, values)
api.USER.update(user_id, values)
return User_Update(None,
None, None, None, user.enabled)
def set_user_tenant(self, admin_token, user_id, user):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
if not isinstance(user, User):
raise fault.BadRequestFault("Expecting a User")
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if duser == None:
raise fault.ItemNotFoundFault("The user could not be found")
dtenant = self.validate_and_fetch_user_tenant(user.tenant_id)
values = {'tenant_id': user.tenant_id}
api.user.update(user_id, values)
api.USER.update(user_id, values)
return User_Update(None,
None, user.tenant_id, None, None)
def delete_user(self, admin_token, user_id):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
dtenant = api.tenant.get(duser.tenant_id)
dtenant = api.TENANT.get(duser.tenant_id)
if dtenant != None:
api.user.delete_tenant_user(user_id, dtenant.id)
api.USER.delete_tenant_user(user_id, dtenant.id)
else:
api.user.delete(user_id)
api.USER.delete(user_id)
return None
def __get_auth_data(self, dtoken, tenant_id):
"""return AuthData object for a token"""
endpoints = None
if tenant_id != None:
endpoints = api.tenant.get_all_endpoints(tenant_id)
endpoints = api.TENANT.get_all_endpoints(tenant_id)
token = auth.Token(dtoken.expires, dtoken.id, tenant_id)
return auth.AuthData(token, endpoints)
@ -423,12 +424,12 @@ class IdentityService(object):
token = auth.Token(dtoken.expires, dtoken.id, dtoken.tenant_id)
ts = []
if dtoken.tenant_id:
droleRefs = api.role.ref_get_all_tenant_roles(duser.id,
droleRefs = api.ROLE.ref_get_all_tenant_roles(duser.id,
dtoken.tenant_id)
for droleRef in droleRefs:
ts.append(RoleRef(droleRef.id, droleRef.role_id,
droleRef.tenant_id))
droleRefs = api.role.ref_get_all_global_roles(duser.id)
droleRefs = api.ROLE.ref_get_all_global_roles(duser.id)
for droleRef in droleRefs:
ts.append(RoleRef(droleRef.id, droleRef.role_id,
droleRef.tenant_id))
@ -439,7 +440,7 @@ class IdentityService(object):
if not tenant_id:
raise fault.UnauthorizedFault("Missing tenant")
tenant = api.tenant.get(tenant_id)
tenant = api.TENANT.get(tenant_id)
if not tenant.enabled:
raise fault.TenantDisabledFault("Tenant %s has been disabled!"
@ -475,14 +476,24 @@ class IdentityService(object):
def __validate_admin_token(self, token_id):
(token, user) = self.__validate_token(token_id)
for roleRef in api.role.ref_get_all_global_roles(user.id):
if roleRef.role_id == backends.KeyStoneAdminRole and \
for roleRef in api.ROLE.ref_get_all_global_roles(user.id):
if roleRef.role_id == backends.KEYSTONEADMINROLE and \
roleRef.tenant_id is None:
return (token, user)
raise fault.UnauthorizedFault(
"You are not authorized to make this call")
def __validate_service_or_keystone_admin_token(self, token_id):
(token, user) = self.__validate_token(token_id)
for roleRef in api.ROLE.ref_get_all_global_roles(user.id):
if (roleRef.role_id == backends.KEYSTONEADMINROLE or \
roleRef.role_id == backends.KEYSTONESERVICEADMINROLE) and \
roleRef.tenant_id is None:
return (token, user)
raise fault.UnauthorizedFault(
"You are not authorized to make this call")
def create_role(self, admin_token, role):
self.__validate_admin_token(admin_token)
@ -492,24 +503,24 @@ class IdentityService(object):
if role.role_id == None:
raise fault.BadRequestFault("Expecting a Role Id")
if api.role.get(role.role_id) != None:
if api.ROLE.get(role.role_id) != None:
raise fault.RoleConflictFault(
"A role with that id already exists")
drole = models.Role()
drole.id = role.role_id
drole.desc = role.desc
api.role.create(drole)
api.ROLE.create(drole)
return role
def get_roles(self, admin_token, marker, limit, url):
self.__validate_admin_token(admin_token)
ts = []
droles = api.role.get_page(marker, limit)
droles = api.ROLE.get_page(marker, limit)
for drole in droles:
ts.append(Role(drole.id,
drole.desc))
prev, next = api.role.get_page_markers(marker, limit)
prev, next = api.ROLE.get_page_markers(marker, limit)
links = []
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" \
@ -522,14 +533,21 @@ class IdentityService(object):
def get_role(self, admin_token, role_id):
self.__validate_admin_token(admin_token)
drole = api.role.get(role_id)
drole = api.ROLE.get(role_id)
if not drole:
raise fault.ItemNotFoundFault("The role could not be found")
return Role(drole.id, drole.desc)
def delete_role(self, admin_token, role_id):
self.__validate_admin_token(admin_token)
drole = api.ROLE.get(role_id)
if not drole:
raise fault.ItemNotFoundFault("The role could not be found")
api.ROLE.delete(role_id)
def create_role_ref(self, admin_token, user_id, roleRef):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
@ -540,12 +558,12 @@ class IdentityService(object):
if roleRef.role_id == None:
raise fault.BadRequestFault("Expecting a Role Id")
drole = api.role.get(roleRef.role_id)
drole = api.ROLE.get(roleRef.role_id)
if drole == None:
raise fault.ItemNotFoundFault("The role not found")
if roleRef.tenant_id != None:
dtenant = api.tenant.get(roleRef.tenant_id)
dtenant = api.TENANT.get(roleRef.tenant_id)
if dtenant == None:
raise fault.ItemNotFoundFault("The tenant not found")
@ -554,28 +572,28 @@ class IdentityService(object):
drole_ref.role_id = drole.id
if roleRef.tenant_id != None:
drole_ref.tenant_id = dtenant.id
user_role_ref = api.user.user_role_add(drole_ref)
user_role_ref = api.USER.user_role_add(drole_ref)
roleRef.role_ref_id = user_role_ref.id
return roleRef
def delete_role_ref(self, admin_token, role_ref_id):
self.__validate_admin_token(admin_token)
api.role.ref_delete(role_ref_id)
api.ROLE.ref_delete(role_ref_id)
return None
def get_user_roles(self, admin_token, marker, limit, url, user_id):
self.__validate_admin_token(admin_token)
duser = api.user.get(user_id)
duser = api.USER.get(user_id)
if not duser:
raise fault.ItemNotFoundFault("The user could not be found")
ts = []
droleRefs = api.role.ref_get_page(marker, limit, user_id)
droleRefs = api.ROLE.ref_get_page(marker, limit, user_id)
for droleRef in droleRefs:
ts.append(RoleRef(droleRef.id, droleRef.role_id,
droleRef.tenant_id))
prev, next = api.role.ref_get_page_markers(user_id, marker, limit)
prev, next = api.ROLE.ref_get_page_markers(user_id, marker, limit)
links = []
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" \
@ -589,7 +607,7 @@ class IdentityService(object):
self.__validate_admin_token(admin_token)
ts = []
dendpointTemplates = api.endpoint_template.get_page(marker, limit)
dendpointTemplates = api.ENDPOINT_TEMPLATE.get_page(marker, limit)
for dendpointTemplate in dendpointTemplates:
ts.append(EndpointTemplate(
dendpointTemplate.id,
@ -600,7 +618,7 @@ class IdentityService(object):
dendpointTemplate.internal_url,
dendpointTemplate.enabled,
dendpointTemplate.is_global))
prev, next = api.endpoint_template.get_page_markers(marker, limit)
prev, next = api.ENDPOINT_TEMPLATE.get_page_markers(marker, limit)
links = []
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" \
@ -613,7 +631,7 @@ class IdentityService(object):
def get_endpoint_template(self, admin_token, endpoint_template_id):
self.__validate_admin_token(admin_token)
dendpointTemplate = api.endpoint_template.get(endpoint_template_id)
dendpointTemplate = api.ENDPOINT_TEMPLATE.get(endpoint_template_id)
if not dendpointTemplate:
raise fault.ItemNotFoundFault(
"The endpoint template could not be found")
@ -632,13 +650,13 @@ class IdentityService(object):
if tenant_id == None:
raise fault.BadRequestFault("Expecting a Tenant Id")
if api.tenant.get(tenant_id) == None:
if api.TENANT.get(tenant_id) == None:
raise fault.ItemNotFoundFault("The tenant not found")
ts = []
dtenantEndpoints = \
api.endpoint_template.\
api.ENDPOINT_TEMPLATE.\
endpoint_get_by_tenant_get_page(
tenant_id, marker, limit)
for dtenantEndpoint in dtenantEndpoints:
@ -648,7 +666,7 @@ class IdentityService(object):
links = []
if ts.__len__():
prev, next = \
api.endpoint_template.endpoint_get_by_tenant_get_page_markers(
api.ENDPOINT_TEMPLATE.endpoint_get_by_tenant_get_page_markers(
tenant_id, marker, limit)
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" %
@ -663,22 +681,74 @@ class IdentityService(object):
self.__validate_admin_token(admin_token)
if tenant_id == None:
raise fault.BadRequestFault("Expecting a Tenant Id")
if api.tenant.get(tenant_id) == None:
if api.TENANT.get(tenant_id) == None:
raise fault.ItemNotFoundFault("The tenant not found")
dendpoint_template = api.endpoint_template.get(endpoint_template.id)
dendpoint_template = api.ENDPOINT_TEMPLATE.get(endpoint_template.id)
if not dendpoint_template:
raise fault.ItemNotFoundFault(
"The endpoint template could not be found")
dendpoint = models.Endpoints()
dendpoint.tenant_id = tenant_id
dendpoint.endpoint_template_id = endpoint_template.id
dendpoint = api.endpoint_template.endpoint_add(dendpoint)
dendpoint = api.ENDPOINT_TEMPLATE.endpoint_add(dendpoint)
dendpoint = Endpoint(dendpoint.id, url +
'/endpointTemplates/' + dendpoint.endpoint_template_id)
return dendpoint
def delete_endpoint(self, admin_token, endpoint_id):
self.__validate_admin_token(admin_token)
api.endpoint_template.endpoint_delete(endpoint_id)
api.ENDPOINT_TEMPLATE.endpoint_delete(endpoint_id)
return None
#Service Operations
def create_service(self, admin_token, service):
self.__validate_service_or_keystone_admin_token(admin_token)
if not isinstance(service, Service):
raise fault.BadRequestFault("Expecting a Service")
if service.service_id == None:
raise fault.BadRequestFault("Expecting a Service Id")
if api.SERVICE.get(service.service_id) != None:
raise fault.ServiceConflictFault(
"A service with that id already exists")
dservice = models.Service()
dservice.id = service.service_id
dservice.desc = service.desc
api.SERVICE.create(dservice)
return service
def get_services(self, admin_token, marker, limit, url):
self.__validate_service_or_keystone_admin_token(admin_token)
ts = []
dservices = api.SERVICE.get_page(marker, limit)
for dservice in dservices:
ts.append(Service(dservice.id,
dservice.desc))
prev, next = api.SERVICE.get_page_markers(marker, limit)
links = []
if prev:
links.append(atom.Link('prev', "%s?'marker=%s&limit=%s'" \
% (url, prev, limit)))
if next:
links.append(atom.Link('next', "%s?'marker=%s&limit=%s'" \
% (url, next, limit)))
return Services(ts, links)
def get_service(self, admin_token, service_id):
self.__validate_service_or_keystone_admin_token(admin_token)
dservice = api.SERVICE.get(service_id)
if not dservice:
raise fault.ItemNotFoundFault("The service could not be found")
return Service(dservice.id, dservice.desc)
def delete_service(self, admin_token, service_id):
self.__validate_service_or_keystone_admin_token(admin_token)
dservice = api.SERVICE.get(service_id)
if not dservice:
raise fault.ItemNotFoundFault("The service could not be found")
api.SERVICE.delete(service_id)

8
keystone/logic/types/fault.py Normal file → Executable file
View File

@ -151,3 +151,11 @@ class RoleConflictFault(IdentityFault):
def __init__(self, msg, details=None, code=409):
super(RoleConflictFault, self).__init__(msg, details, code)
self.key = "roleConflict"
class ServiceConflictFault(IdentityFault):
"""The Service already exists?"""
def __init__(self, msg, details=None, code=409):
super(ServiceConflictFault, self).__init__(msg, details, code)
self.key = "serviceConflict"

26
keystone/logic/types/role.py
View File

@ -21,9 +21,10 @@ from keystone.logic.types import fault
class Role(object):
def __init__(self, role_id, desc):
def __init__(self, role_id, desc, service_id=None):
self.role_id = role_id
self.desc = desc
self.service_id = service_id
@staticmethod
def from_xml(xml_str):
@ -31,14 +32,15 @@ class Role(object):
dom = etree.Element("root")
dom.append(etree.fromstring(xml_str))
root = dom.find("{http://docs.openstack.org/identity/api/v2.0}" \
"role")
"role")
if root == None:
raise fault.BadRequestFault("Expecting Role")
role_id = root.get("id")
desc = root.get("description")
if role_id == None:
raise fault.BadRequestFault("Expecting Role")
return Role(role_id, desc)
service_id = root.get("serviceId")
return Role(role_id, desc, service_id)
except etree.LxmlError as e:
raise fault.BadRequestFault("Cannot parse Role", str(e))
@ -55,8 +57,18 @@ class Role(object):
role_id = role["id"]
if role_id == None:
raise fault.BadRequestFault("Expecting Role")
desc = role["description"]
return Role(role_id, desc)
if not "description" in role:
desc = None
else:
desc = role["description"]
if not "serviceId" in role:
service_id = None
else:
service_id = role["serviceId"]
return Role(role_id, desc, service_id)
except (ValueError, TypeError) as e:
raise fault.BadRequestFault("Cannot parse Role", str(e))
@ -67,6 +79,8 @@ class Role(object):
dom.set("id", self.role_id)
if self.desc:
dom.set("description", string.lower(str(self.desc)))
if self.service_id:
dom.set("serviceId", string.lower(str(self.service_id)))
return dom
def to_xml(self):
@ -78,6 +92,8 @@ class Role(object):
role["id"] = self.role_id
if self.desc:
role["description"] = self.desc
if self.service_id:
role["serviceId"] = self.desc
return {'role': role}
def to_json(self):

109
keystone/logic/types/service.py Normal file
View File

@ -0,0 +1,109 @@
# Copyright (c) 2010-2011 OpenStack, LLC.
#
# 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 json
from lxml import etree
import string
from keystone.logic.types import fault
class Service(object):
def __init__(self, service_id, desc):
self.service_id = service_id
self.desc = desc
@staticmethod
def from_xml(xml_str):
try:
dom = etree.Element("root")
dom.append(etree.fromstring(xml_str))
root = dom.find("{http://docs.openstack.org/identity/api/v2.0}" \
"service")
if root == None:
raise fault.BadRequestFault("Expecting Service")
service_id = root.get("id")
desc = root.get("description")
if service_id == None:
raise fault.BadRequestFault("Expecting Service")
return Service(service_id, desc)
except etree.LxmlError as e:
raise fault.BadRequestFault("Cannot parse service", str(e))
@staticmethod
def from_json(json_str):
try:
obj = json.loads(json_str)
if not "service" in obj:
raise fault.BadRequestFault("Expecting service")
service = obj["service"]
if not "id" in service:
service_id = None
else:
service_id = service["id"]
if service_id == None:
raise fault.BadRequestFault("Expecting service")
desc = service["description"]
return Service(service_id, desc)
except (ValueError, TypeError) as e:
raise fault.BadRequestFault("Cannot parse service", str(e))
def to_dom(self):
dom = etree.Element("service",
xmlns="http://docs.openstack.org/identity/api/v2.0")
if self.service_id:
dom.set("id", self.service_id)
if self.desc:
dom.set("description", string.lower(str(self.desc)))
return dom
def to_xml(self):
return etree.tostring(self.to_dom())
def to_dict(self):
service = {}
if self.service_id:
service["id"] = self.service_id
if self.desc:
service["description"] = self.desc
return {'service': service}
def to_json(self):
return json.dumps(self.to_dict())
class Services(object):
"A collection of services."
def __init__(self, values, links):
self.values = values
self.links = links
def to_xml(self):
dom = etree.Element("services")
dom.set(u"xmlns", "http://docs.openstack.org/identity/api/v2.0")
for t in self.values:
dom.append(t.to_dom())
for t in self.links:
dom.append(t.to_dom())
return etree.tostring(dom)
def to_json(self):
values = [t.to_dict()["service"] for t in self.values]
links = [t.to_dict()["links"] for t in self.links]
return json.dumps({"services": {"values": values, "links": links}})

10
keystone/middleware/auth_basic.py
View File

@ -95,9 +95,8 @@ class AuthProtocol(object):
# If the user isn't authenticated, we reject the request and
# return 401 indicating we need Basic Auth credentials.
return HTTPUnauthorized("Authentication required",
[('WWW-Authenticate',
'Basic realm="Use guest/guest"')])\
(env, start_response)
[('WWW-Authenticate',
'Basic realm="Use guest/guest"')])(env, start_response)
else:
# Claims were provided - validate them
import base64
@ -109,9 +108,8 @@ class AuthProtocol(object):
if not self.delay_auth_decision:
# Reject request (or ask for valid claims)
return HTTPUnauthorized("Authentication required",
[('WWW-Authenticate',
'Basic realm="Use guest/guest"')])\
(env, start_response)
[('WWW-Authenticate',
'Basic realm="Use guest/guest"')])(env, start_response)
else:
# Claims are valid, forward request
_decorate_request_headers("X_IDENTITY_STATUS", "Invalid",

20
keystone/routers/admin.py Normal file → Executable file
View File

@ -5,6 +5,7 @@ import keystone.backends as db
from keystone.controllers.auth import AuthController
from keystone.controllers.endpointtemplates import EndpointTemplatesController
from keystone.controllers.roles import RolesController
from keystone.controllers.services import ServicesController
from keystone.controllers.staticfiles import StaticFilesController
from keystone.controllers.tenant import TenantController
from keystone.controllers.user import UserController
@ -91,10 +92,14 @@ class AdminApi(wsgi.Router):
#Roles and RoleRefs
roles_controller = RolesController(options)
mapper.connect("/roles", controller=roles_controller,
action="create_role", conditions=dict(method=["POST"]))
mapper.connect("/roles", controller=roles_controller,
action="get_roles", conditions=dict(method=["GET"]))
mapper.connect("/roles/{role_id}", controller=roles_controller,
action="get_role", conditions=dict(method=["GET"]))
mapper.connect("/roles/{role_id}", controller=roles_controller,
action="delete_role", conditions=dict(method=["DELETE"]))
mapper.connect("/users/{user_id}/roleRefs",
controller=roles_controller, action="get_role_refs",
conditions=dict(method=["GET"]))
@ -104,6 +109,7 @@ class AdminApi(wsgi.Router):
mapper.connect("/users/{user_id}/roleRefs/{role_ref_id}",
controller=roles_controller, action="delete_role_ref",
conditions=dict(method=["DELETE"]))
#EndpointTemplatesControllers and Endpoints
endpoint_templates_controller = EndpointTemplatesController(options)
mapper.connect("/endpointTemplates",
@ -153,4 +159,18 @@ class AdminApi(wsgi.Router):
action="get_xsd_atom_contract",
conditions=dict(method=["GET"]))
# Services Controller
services_controller = ServicesController(options)
mapper.connect("/services", controller=services_controller,
action="get_services", conditions=dict(method=["GET"]))
mapper.connect("/services", controller=services_controller,
action="create_service", conditions=dict(method=["POST"]))
mapper.connect("/services/{service_id}",\
controller=services_controller,
action="delete_service",
conditions=dict(method=["DELETE"]))
mapper.connect("/services/{service_id}",
controller=services_controller,
action="get_service",
conditions=dict(method=["GET"]))
super(AdminApi, self).__init__(mapper)

1
keystone/test/run_tests.py Normal file → Executable file
View File

@ -17,6 +17,7 @@ TEMP_FILES = (
'ldap.db',
'ldap.db.db')
def delete_temp_files():
"""Quietly deletes any temp files in the test directory"""
for path in TEMP_FILES:

90
keystone/test/unit/test_common.py
View File

@ -440,6 +440,10 @@ def get_auth_token():
return '999888777666'
def get_service_token():
return '111222333444'
def get_exp_auth_token():
return '000999'
@ -542,6 +546,92 @@ def create_role_xml(role_id, auth_token):
return (resp, content)
def delete_role(role_id, auth_token):
header = httplib2.Http(".cache")
url = '%sroles/%s' % (URL_V2, role_id)
resp, content = header.request(url, "DELETE", body='',
headers={"Content-Type": "application/json",
"X-Auth-Token": str(auth_token)})
return resp, content
def create_service(service_id, auth_token):
header = httplib2.Http(".cache")
url = '%sservices' % (URL_V2)
body = {"service": {"id": service_id,
"description": "A description ..."}}
resp, content = header.request(url, "POST", body=json.dumps(body),
headers={"Content-Type": "application/json",
"X-Auth-Token": auth_token})
return (resp, content)
def create_service_xml(service_id, auth_token):
header = httplib2.Http(".cache")
url = '%sservices' % (URL_V2)
body = '<?xml version="1.0" encoding="UTF-8"?>\
<service xmlns="http://docs.openstack.org/identity/api/v2.0" \
id="%s" description="A Description of the service"/>\
' % service_id
resp, content = header.request(url, "POST", body=body,
headers={"Content-Type": "application/xml",
"X-Auth-Token": auth_token,
"ACCEPT": "application/xml"})
return (resp, content)
def delete_service(service_id, auth_token):
header = httplib2.Http(".cache")
url = '%sservices/%s' % (URL_V2, service_id)
resp, content = header.request(url, "DELETE", body='',
headers={"Content-Type": "application/json",
"X-Auth-Token": str(auth_token)})
return resp, content
def get_services(auth_token):
header = httplib2.Http(".cache")
url = '%sservices' % (URL_V2)
#test for Content-Type = application/json
resp, content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": auth_token})
return (resp, content)
def get_services_xml(auth_token):
header = httplib2.Http(".cache")
url = '%sservices' % (URL_V2)
#test for Content-Type = application/xml
resp, content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/xml",
"X-Auth-Token": auth_token,
"ACCEPT": "application/xml"})
return (resp, content)
def get_service(service_id, auth_token):
header = httplib2.Http(".cache")
url = '%sservices/%s' % (URL_V2, service_id)
#test for Content-Type = application/json
resp, content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": auth_token})
return (resp, content)
def get_service_xml(service_id, auth_token):
header = httplib2.Http(".cache")
url = '%sservices/%s' % (URL_V2, service_id)
#test for Content-Type = application/xml
resp, content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/xml",
"X-Auth-Token": auth_token,
"ACCEPT": "application/xml"})
return (resp, content)
def create_endpoint(tenant_id, endpoint_templates_id, auth_token):
header = httplib2.Http(".cache")

116
keystone/test/unit/test_roles.py
View File

@ -45,13 +45,29 @@ class RolesTest(unittest.TestCase):
utils.create_tenant(self.tenant, str(self.auth_token))
utils.create_user(self.tenant, self.user, self.auth_token)
self.token = utils.get_token(self.user, 'secrete', self.tenant,
'token')
'token')
def tearDown(self):
utils.delete_user(self.user, self.auth_token)
utils.delete_tenant(self.tenant, self.auth_token)
class CreateRolesTest(RolesTest):
def test_create_role(self):
resp, content = utils.create_role('test_role', self.auth_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(201, int(resp['status']))
resp, content = utils.delete_role('test_role', self.auth_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(204, int(resp['status']))
class GetRolesTest(RolesTest):
def test_get_roles(self):
header = httplib2.Http(".cache")
@ -71,7 +87,7 @@ class GetRolesTest(RolesTest):
if not "roles" in obj:
raise self.fail("Expecting Roles")
roles = obj["roles"]["values"]
if len(roles) != 2:
if len(roles) != 3:
self.fail("Roles not of required length.")
role = roles[0]
@ -79,7 +95,7 @@ class GetRolesTest(RolesTest):
role_id = None
else:
role_id = role["id"]
if role_id not in ['Admin', 'Member']:
if role_id not in ['Admin', 'Member', 'KeystoneServiceAdmin']:
self.fail("Not the expected Role")
def test_get_roles_xml(self):
@ -95,7 +111,6 @@ class GetRolesTest(RolesTest):
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
# Validate Returned Content
dom = etree.Element("root")
dom.append(etree.fromstring(content))
@ -105,10 +120,11 @@ class GetRolesTest(RolesTest):
self.fail("Expecting Roles")
roles = roles.findall("{http://docs.openstack.org/identity/api/v2.0}" \
"role")
if len(roles) != 2:
if len(roles) != 3:
self.fail("Not the expected Role count")
for role in roles:
if role.get("id") not in ['Admin', 'Member']:
if role.get("id") not in ['Admin', 'Member', \
'KeystoneServiceAdmin']:
self.fail("Unexpected Role")
def test_get_roles_exp_token(self):
@ -126,7 +142,7 @@ class GetRolesTest(RolesTest):
def test_get_roles_exp_token_xml(self):
header = httplib2.Http(".cache")
url = '%stenants' % (utils.URL_V2)
url = '%sroles' % (utils.URL_V2)
#test for Content-Type = application/json
resp, _content = header.request(url, "GET", body='',
headers={"Content-Type": "application/xml",
@ -155,7 +171,6 @@ class GetRoleTest(RolesTest):
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
#verify content
obj = json.loads(content)
if not "role" in obj:
@ -225,9 +240,9 @@ class GetRoleTest(RolesTest):
header = httplib2.Http(".cache")
url = '%sroles/%s' % (utils.URL_V2, self.role)
#test for Content-Type = application/json
resp, _content = header.request(url, "GET", body='{}', headers={
"Content-Type": "application/json",
"X-Auth-Token": self.exp_auth_token})
resp, _content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": self.exp_auth_token})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
@ -254,9 +269,9 @@ class GetRoleTest(RolesTest):
header = httplib2.Http(".cache")
url = '%sroles/%s' % (utils.URL_V2, self.role)
#test for Content-Type = application/json
resp, _content = header.request(url, "GET", body='{}', headers={
"Content-Type": "application/json",
"X-Auth-Token": self.disabled_token})
resp, _content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": self.disabled_token})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
@ -347,8 +362,9 @@ class CreateRoleRefTest(RolesTest):
def test_role_ref_create_xml(self):
utils.add_user_json(self.auth_token)
resp, _content = utils.create_role_ref_xml(self.user, 'Admin',
self.tenant, str(self.auth_token))
resp, _content = utils.create_role_ref_xml(
self.user, 'Admin', self.tenant,
str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(201, resp_val)
@ -385,8 +401,9 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(
self.user, 'Admin', self.tenant,
str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/json
resp, content = header.request(url, "GET", body='{}',
@ -406,8 +423,9 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs_xml(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(
self.user, 'Admin', self.tenant,
str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/xml
resp, content = header.request(url, "GET", body='{}',
@ -430,13 +448,14 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs_using_expired_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(self.user,
'Admin', self.tenant,
str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/json
resp, _content = header.request(url, "GET", body='{}', headers={
"Content-Type": "application/json",
"X-Auth-Token": str(self.exp_auth_token)})
resp, _content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": str(self.exp_auth_token)})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
@ -446,14 +465,15 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs_xml_using_expired_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(
self.user, 'Admin', self.tenant,
str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/xml
resp, _content = header.request(url, "GET", body='{}', headers={
"Content-Type": "application/xml",
"X-Auth-Token": str(self.exp_auth_token),
"ACCEPT": "application/xml"})
resp, _content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/xml",
"X-Auth-Token": str(self.exp_auth_token),
"ACCEPT": "application/xml"})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
@ -463,13 +483,13 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs_using_disabled_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(self.user,
'Admin', self.tenant, str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/json
resp, _content = header.request(url, "GET", body='{}', headers={
"Content-Type": "application/json",
"X-Auth-Token": str(self.disabled_token)})
resp, _content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": str(self.disabled_token)})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
@ -479,14 +499,15 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs_xml_using_disabled_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(
self.user, 'Admin', self.tenant,
str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/xml
resp, _content = header.request(url, "GET", body='{}', headers={
"Content-Type": "application/xml",
"X-Auth-Token": str(self.disabled_token),
"ACCEPT": "application/xml"})
resp, _content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/xml",
"X-Auth-Token": str(self.disabled_token),
"ACCEPT": "application/xml"})
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
@ -545,8 +566,9 @@ class GetRoleRefsTest(RolesTest):
def test_get_rolerefs_xml_using_invalid_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
_resp, _content = utils.create_role_ref(self.user, 'Admin',
self.tenant, str(self.auth_token))
_resp, _content = utils.create_role_ref(
self.user, 'Admin', self.tenant,
str(self.auth_token))
url = '%susers/%s/roleRefs' % (URL_V2, self.user)
#test for Content-Type = application/xml
resp, _content = header.request(url, "GET", body='{}', headers={
@ -651,9 +673,9 @@ class DeleteRoleRefTest(RolesTest):
if role_ref_id is None:
raise fault.BadRequestFault("Expecting RoleRefId")
url = '%susers/%s/roleRefs/%s' % (URL_V2, self.user, role_ref_id)
resp, content = header.request(url, "DELETE", body='', headers={
"Content-Type": "application/json",
"X-Auth-Token": str(self.missing_token)})
resp, content = header.request(url, "DELETE", body='',
headers={"Content-Type": "application/json",
"X-Auth-Token": str(self.missing_token)})
resp_val = int(resp['status'])
self.assertEqual(401, resp_val)

293
keystone/test/unit/test_services.py Executable file
View File

@ -0,0 +1,293 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2010-2011 OpenStack, LLC.
#
# 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 httplib2
import json
from lxml import etree
import os
import sys
sys.path.append(os.path.abspath(os.path.join(os.path.abspath(__file__),
'..', '..', '..', '..', '..', 'keystone')))
import unittest
import test_common as utils
from test_common import URL_V2
from keystone.logic.types import fault
class ServicesTest(unittest.TestCase):
def setUp(self):
self.auth_token = utils.get_auth_token()
self.service_token = utils.get_service_token()
self.missing_token = utils.get_none_token()
self.invalid_token = utils.get_non_existing_token()
self.disabled_token = utils.get_disabled_token()
self.exp_auth_token = utils.get_exp_auth_token()
self.user = utils.get_user()
self.tenant = utils.get_tenant()
utils.create_tenant(self.tenant, str(self.auth_token))
utils.create_user(self.tenant, self.user, self.auth_token)
self.sample_service = 'sampleservice'
self.test_service = 'test_service'
self.token = utils.get_token(self.user, 'secrete', self.tenant,
'token')
utils.create_service(self.sample_service, str(self.auth_token))
def tearDown(self):
utils.delete_user(self.user, self.auth_token)
utils.delete_tenant(self.tenant, self.auth_token)
utils.delete_service(self.sample_service, str(self.auth_token))
class GetServicesTest(ServicesTest):
def test_get_services_using_keystone_admin_token_json(self):
resp, content = utils.get_services(self.auth_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
#verify content
obj = json.loads(content)
if not "services" in obj:
raise self.fail("Expecting Services")
services = obj["services"]["values"]
if len(services) < 1:
self.fail("Services not of required length.")
is_service_found = None
for service in services:
if service["id"] in [self.sample_service]:
is_service_found = True
if not is_service_found:
raise self.fail("Service not found")
def test_get_services_using_keystone_admin_token_xml(self):
resp, content = utils.get_services_xml(self.auth_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
# verify content
# Validate Returned Content
dom = etree.Element("root")
dom.append(etree.fromstring(content))
services = dom.find(
"{http://docs.openstack.org/identity/api/v2.0}" \
"services")
if services == None:
self.fail("Expecting Services")
services = services.findall(
"{http://docs.openstack.org/identity/api/v2.0}" \
"service")
if len(services) < 1:
self.fail("Not the expected Service count")
for service in services:
if service.get("id") in [self.sample_service]:
is_service_found = True
if not is_service_found:
raise self.fail("Service not found")
def test_get_services_using_service_admin_token(self):
resp, content = utils.get_services(self.service_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
#verify content
obj = json.loads(content)
if not "services" in obj:
raise self.fail("Expecting Services")
services = obj["services"]["values"]
if len(services) < 1:
self.fail("Services not of required length.")
is_service_found = None
for service in services:
if service["id"] in [self.sample_service]:
is_service_found = True
if not is_service_found:
raise self.fail("Service not found")
def test_get_services_using_service_admin_token_xml(self):
resp, content = utils.get_services_xml(self.service_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(200, int(resp['status']))
# Verify content
# Validate Returned Content
dom = etree.Element("root")
dom.append(etree.fromstring(content))
services = dom.find(
"{http://docs.openstack.org/identity/api/v2.0}" \
"services")
if services == None:
self.fail("Expecting Services")
services = services.findall(
"{http://docs.openstack.org/identity/api/v2.0}" \
"service")
if len(services) < 1:
self.fail("Not the expected Service count")
for service in services:
if service.get("id") in [self.sample_service]:
is_service_found = True
if not is_service_found:
raise self.fail("Service not found")
def test_get_services_exp_token(self):
resp, content = utils.get_services(self.exp_auth_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(403, int(resp['status']))
def test_get_services_exp_token_xml(self):
resp, content = utils.get_services_xml(self.exp_auth_token)
if int(resp['status']) == 500:
self.fail('Identity Fault')
elif int(resp['status']) == 503:
self.fail('Service Not Available')
self.assertEqual(403, int(resp['status']))
class GetServiceTest(ServicesTest):
def test_service_get_json(self):
resp, _content = utils.get_service(
self.sample_service, str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(200, resp_val)
def test_service_get_xml(self):
resp, _content = utils.get_service_xml(
self.sample_service, str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(200, resp_val)
def test_service_get_using_expired_token(self):
resp, _content = utils.get_service(
self.sample_service, str(self.exp_auth_token))
resp_val = int(resp['status'])
self.assertEqual(403, resp_val)
def test_service_get_using_disabled_token(self):
resp, _content = utils.get_service(
self.sample_service, str(self.disabled_token))
resp_val = int(resp['status'])
self.assertEqual(403, resp_val)
def test_service_get_json_using_missing_token(self):
resp, _content = utils.get_service(
self.sample_service, str(self.missing_token))
resp_val = int(resp['status'])
self.assertEqual(401, resp_val)
def test_service_get_json_using_invalid_token(self):
resp, _content = utils.get_service(
self.sample_service, str(self.invalid_token))
resp_val = int(resp['status'])
self.assertEqual(404, resp_val)
class CreateServiceTest(ServicesTest):
def test_service_create_json(self):
resp, _content = utils.create_service(
self.test_service, str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(201, resp_val)
utils.delete_service(self.test_service, self.auth_token)
def test_service_create_xml(self):
resp, _content = utils.create_service_xml(
self.test_service, str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(201, resp_val)
def test_service_create_duplicate_json(self):
resp, _content = utils.create_service(
self.test_service, str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(201, resp_val)
resp, _content = utils.create_service(
self.test_service, str(self.auth_token))
resp_val = int(resp['status'])
self.assertEqual(409, resp_val)
utils.delete_service(self.test_service, self.auth_token)
def test_service_create_using_expired_token(self):
resp, _content = utils.create_service(
self.test_service, str(self.exp_auth_token))
resp_val = int(resp['status'])
self.assertEqual(403, resp_val)
def test_service_create_using_disabled_token(self):
resp, _content = utils.create_service(
self.test_service, str(self.disabled_token))
resp_val = int(resp['status'])
self.assertEqual(403, resp_val)
def test_service_create_json_using_missing_token(self):
resp, _content = utils.create_service(
self.test_service, str(self.missing_token))
resp_val = int(resp['status'])
self.assertEqual(401, resp_val)
def test_service_create_json_using_invalid_token(self):
resp, _content = utils.create_service(
self.test_service, str(self.invalid_token))
resp_val = int(resp['status'])
self.assertEqual(404, resp_val)
class DeleteServiceTest(ServicesTest):
def test_service_delete(self):
resp, _content = utils.delete_service(
self.test_service, self.auth_token)
resp_val = int(resp['status'])
self.assertEqual(204, resp_val)
def test_service_delete_json_using_expired_token(self):
resp, _content = utils.delete_service(
self.test_service, str(self.exp_auth_token))
resp_val = int(resp['status'])
self.assertEqual(403, resp_val)
def test_service_delete_json_using_disabled_token(self):
resp, _content = utils.delete_service(
self.test_service, str(self.disabled_token))
resp_val = int(resp['status'])
self.assertEqual(403, resp_val)
def test_service_delete_json_using_missing_token(self):
resp, _content = utils.delete_service(
self.test_service, str(self.missing_token))
resp_val = int(resp['status'])
self.assertEqual(401, resp_val)
def test_service_delete_json_using_invalid_token(self):
resp, _content = utils.delete_service(
self.test_service, str(self.invalid_token))
resp_val = int(resp['status'])
self.assertEqual(404, resp_val)
if __name__ == '__main__':
unittest.main()