openstack/deb-keystone
Code Issues Proposed changes

Additional tests and minor changes to support services CRUD.

Browse Source 
And PEP8 fixes.

Change-Id: I4837c58f68ed461b65ac85c591f5e19dbbe034d3
This commit is contained in:
Yogeshwar Srikrishnan 2011-08-15 03:25:27 -05:00 committed by Dolph Mathews
parent e8c35a7f23
commit b71ce6e401
6 changed files with 200 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
keystone/backends/memcache/api/token.py
View File

@ -40,6 +40,8 @@ class TokenAPI(BaseTokenAPI):
token = self.get(id)
if token != None:
MEMCACHE_SERVER.delete(id)
if token != None and not hasattr(token, 'tenant_id'):
token.tenant_id = None
if token.tenant_id != None:
MEMCACHE_SERVER.delete(token.tenant_id + "::" + token.user_id)
else:
@ -53,7 +55,10 @@ class TokenAPI(BaseTokenAPI):
return token
def get_for_user_by_tenant(self, user_id, tenant_id):
token = MEMCACHE_SERVER.get(tenant_id + "::" + user_id)
if tenant_id != None:
token = MEMCACHE_SERVER.get(tenant_id + "::" + user_id)
else:
token = MEMCACHE_SERVER.get(user_id)
if token != None and not hasattr(token, 'tenant_id'):
token.tenant_id = None
return token

12
keystone/logic/types/endpoint.py
View File

@ -33,7 +33,7 @@ class EndpointTemplate(object):
raise fault.BadRequestFault("Expecting endpointTemplate")
id = root.get("id")
region = root.get("region")
service = root.get("serviceName")
service = root.get("serviceId")
public_url = root.get("publicURL")
admin_url = root.get("adminURL")
internal_url = root.get("internalURL")
@ -63,7 +63,7 @@ class EndpointTemplate(object):
# Check that fields are valid
invalid = [key for key in endpoint_template if key not in
['id', 'region', 'serviceName', 'publicURL',
['id', 'region', 'serviceId', 'publicURL',
'adminURL', 'internalURL', 'enabled', 'global']]
if invalid != []:
raise fault.BadRequestFault("Invalid attribute(s): %s"
@ -76,8 +76,8 @@ class EndpointTemplate(object):
if 'region' in endpoint_template:
region = endpoint_template["region"]
if 'serviceName' in endpoint_template:
service = endpoint_template["serviceName"]
if 'serviceId' in endpoint_template:
service = endpoint_template["serviceId"]
if 'publicURL' in endpoint_template:
public_url = endpoint_template["publicURL"]
if 'adminURL' in endpoint_template:
@ -114,7 +114,7 @@ class EndpointTemplate(object):
if self.region:
dom.set("region", self.region)
if self.service:
dom.set("serviceName", self.service)
dom.set("serviceId", self.service)
if self.public_url:
dom.set("publicURL", self.public_url)
if self.admin_url:
@ -137,7 +137,7 @@ class EndpointTemplate(object):
if self.region:
endpoint_template["region"] = self.region
if self.service:
endpoint_template["serviceName"] = self.service
endpoint_template["serviceId"] = self.service
if self.public_url:
endpoint_template["publicURL"] = self.public_url
if self.admin_url:

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

@ -744,7 +744,7 @@ def create_endpoint_template(region, service,
url = '%sendpointTemplates' % (URL_V2)
body = {"endpointTemplate": {"region": region,
"serviceName": service,
"serviceId": service,
"publicURL": public_url,
"adminURL": admin_url,
"internalURL": internal_url,
@ -763,13 +763,13 @@ def create_endpoint_template_xml(region, service, public_url, admin_url,
url = '%sendpointTemplates' % (URL_V2)
body = '<?xml version="1.0" encoding="UTF-8"?>\
<endpointTemplate xmlns="http://docs.openstack.org/identity/api/v2.0" \
region="%s" serviceName="%s" \
region="%s" serviceId="%s" \
publicURL="%s" adminURL="%s"\
internalURL="%s" enabled="%s"\
global="%s"/>' % (region, service, public_url,\
admin_url, internal_url, enabled, is_global)
body = {"endpointTemplate": {"region": region,
"serviceName": service,
"serviceId": service,
"publicURL": public_url,
"adminURL": admin_url,
"internalURL": internal_url,

6
keystone/test/unit/test_endpoints.py
View File

@ -80,10 +80,10 @@ class CreateEndpointTemplatesTest(EndpointTemplatesTest):
endpoint_template_id = endpoint_template["id"]
if endpoint_template_id == None:
self.fail("Not the expected Endpoint Template")
if not "serviceName" in endpoint_template:
if not "serviceId" in endpoint_template:
service_id = None
else:
service_id = endpoint_template["serviceName"]
service_id = endpoint_template["serviceId"]
if service_id != utils.get_test_service_id():
self.fail("Not the expected service")
resp, content = utils.delete_endpoint_template(
@ -123,7 +123,7 @@ class CreateEndpointTemplatesTest(EndpointTemplatesTest):
if endpoint_template_id == None:
self.fail("Not the expected Endpoint template.")
service_id = endpoint_template.get("serviceName")
service_id = endpoint_template.get("serviceId")
if service_id != utils.get_test_service_id():
self.fail("Not the expected service")

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

@ -38,6 +38,7 @@ class RolesTest(unittest.TestCase):
self.user = utils.get_user()
self.userdisabled = utils.get_userdisabled()
self.auth_token = utils.get_auth_token()
self.service_token = utils.get_service_token()
self.exp_auth_token = utils.get_exp_auth_token()
self.disabled_token = utils.get_disabled_token()
self.missing_token = utils.get_none_token()
@ -68,6 +69,46 @@ class CreateRolesTest(RolesTest):
self.fail('Service Not Available')
self.assertEqual(204, int(resp['status']))
def test_create_role_using_service_token(self):
resp, content = utils.create_role('test_role', self.service_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.service_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']))
def test_create_roles_using_invalid_tokens(self):
resp, content = utils.create_role('test_role', self.disabled_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.create_role('test_role', self.missing_token)
self.assertEqual(401, int(resp['status']))
resp, content = utils.create_role('test_role', self.exp_auth_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.create_role('test_role', self.invalid_token)
self.assertEqual(404, int(resp['status']))
def test_delete_roles_using_invalid_tokens(self):
resp, content = utils.delete_role('test_role', self.disabled_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.delete_role('test_role', self.missing_token)
self.assertEqual(401, int(resp['status']))
resp, content = utils.delete_role('test_role', self.exp_auth_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.delete_role('test_role', self.invalid_token)
self.assertEqual(404, int(resp['status']))
def test_create_role_mapped_to_a_service(self):
resp, content = utils.create_role_mapped_to_service(
'test_role', self.auth_token, self.service_id)
@ -179,6 +220,35 @@ class GetRolesTest(RolesTest):
if role_id not in ['Admin', 'Member', 'KeystoneServiceAdmin']:
self.fail("Not the expected Role")
def test_get_roles_using_service_admin_token(self):
header = httplib2.Http(".cache")
url = '%sroles' % (utils.URL_V2)
#test for Content-Type = application/json
resp, content = header.request(url, "GET", body='{}',
headers={"Content-Type": "application/json",
"X-Auth-Token": 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 "roles" in obj:
raise self.fail("Expecting Roles")
roles = obj["roles"]["values"]
if len(roles) != 3:
self.fail("Roles not of required length.")
role = roles[0]
if not "id" in role:
role_id = None
else:
role_id = role["id"]
if role_id not in ['Admin', 'Member', 'KeystoneServiceAdmin']:
self.fail("Not the expected Role")
def test_get_roles_xml(self):
header = httplib2.Http(".cache")
url = '%sroles' % (utils.URL_V2)
@ -264,6 +334,31 @@ class GetRoleTest(RolesTest):
if role_id != 'Admin':
self.fail("Not the expected Role")
def test_get_role_using_service_admin_token(self):
self.role = 'Admin'
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.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 "role" in obj:
raise fault.BadRequestFault("Expecting Role")
role = obj["role"]
if not "id" in role:
role_id = None
else:
role_id = role["id"]
if role_id != 'Admin':
self.fail("Not the expected Role")
def test_get_role_xml(self):
self.role = 'Admin'
header = httplib2.Http(".cache")
@ -449,6 +544,13 @@ class CreateRoleRefTest(RolesTest):
resp_val = int(resp['status'])
self.assertEqual(201, resp_val)
def test_role_ref_create_json_using_service_admin_token(self):
utils.add_user_json(self.auth_token)
resp, _content = utils.create_role_ref(self.user, 'Admin', self.tenant,
str(self.service_token))
resp_val = int(resp['status'])
self.assertEqual(201, resp_val)
def test_role_ref_create_json_using_expired_token(self):
utils.add_user_json(self.auth_token)
resp, _content = utils.create_role_ref(self.user, 'Admin', self.tenant,
@ -526,6 +628,31 @@ class GetRoleRefsTest(RolesTest):
if roles == None:
self.fail("Expecting Role Refs")
def test_get_rolerefs_xml_using_service_admin_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))
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.service_token),
"ACCEPT": "application/xml"})
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
dom = etree.Element("root")
dom.append(etree.fromstring(content))
roles = dom.find("{http://docs.openstack.org/identity/api/v2.0}" \
"roleRefs")
if roles == None:
self.fail("Expecting Role Refs")
def test_get_rolerefs_using_expired_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)
@ -688,6 +815,30 @@ class DeleteRoleRefTest(RolesTest):
resp_val = int(resp['status'])
self.assertEqual(204, resp_val)
def test_delete_roleref_using_service_admin_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_val = int(resp['status'])
self.assertEqual(201, resp_val)
obj = json.loads(content)
if not "roleRef" in obj:
raise fault.BadRequestFault("Expecting RoleRef")
roleRef = obj["roleRef"]
if not "id" in roleRef:
role_ref_id = None
else:
role_ref_id = roleRef["id"]
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.service_token)})
resp_val = int(resp['status'])
self.assertEqual(204, resp_val)
def test_delete_roleref_using_expired_token(self):
header = httplib2.Http(".cache")
utils.add_user_json(self.auth_token)

43
keystone/test/unit/test_services.py
View File

@ -152,22 +152,19 @@ class GetServicesTest(ServicesTest):
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')
def test_get_services_using_invalid_tokens(self):
resp, content = utils.get_services(self.disabled_token)
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')
resp, content = utils.get_services(self.missing_token)
self.assertEqual(401, int(resp['status']))
resp, content = utils.get_services(self.exp_auth_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.get_services(self.invalid_token)
self.assertEqual(404, int(resp['status']))
class GetServiceTest(ServicesTest):
def test_service_get_json(self):
@ -206,6 +203,23 @@ class GetServiceTest(ServicesTest):
resp_val = int(resp['status'])
self.assertEqual(404, resp_val)
def test_get_service_using_invalid_tokens(self):
resp, content = utils.get_service(
self.sample_service, self.disabled_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.get_service(
self.sample_service, self.missing_token)
self.assertEqual(401, int(resp['status']))
resp, content = utils.get_service(
self.sample_service, self.exp_auth_token)
self.assertEqual(403, int(resp['status']))
resp, content = utils.get_service(
self.sample_service, self.invalid_token)
self.assertEqual(404, int(resp['status']))
class CreateServiceTest(ServicesTest):
def test_service_create_json(self):
@ -288,6 +302,11 @@ class DeleteServiceTest(ServicesTest):
resp_val = int(resp['status'])
self.assertEqual(404, 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)
if __name__ == '__main__':
unittest.main()