openstack/python-keystoneclient
Code Issues Proposed changes

v3 Client & test utils

Browse Source 
Change-Id: I6cafaad053b7fa1ca31f4f5aed1f86aa97c4e87e
This commit is contained in:
Dolph Mathews
2012-09-11 11:20:16 -05:00
committed by Gerrit Code Review
parent 315285e76a
commit 2ca00dc617
4 changed files with 294 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
keystoneclient/v3/__init__.py Normal file
View File

@@ -0,0 +1 @@
from keystoneclient.v3.client import Client

68
keystoneclient/v3/client.py Normal file
View File

@@ -0,0 +1,68 @@
# Copyright 2011 Nebula, Inc.
# 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.
import json
import logging
from keystoneclient.v2_0 import client
_logger = logging.getLogger(__name__)
class Client(client.Client):
"""Client for the OpenStack Identity API v3.
:param string username: Username for authentication. (optional)
:param string password: Password for authentication. (optional)
:param string token: Token for authentication. (optional)
:param string tenant_name: Tenant id. (optional)
:param string tenant_id: Tenant name. (optional)
:param string auth_url: Keystone service endpoint for authorization.
:param string region_name: Name of a region to select when choosing an
endpoint from the service catalog.
:param string endpoint: A user-supplied endpoint URL for the keystone
service. Lazy-authentication is possible for API
service calls if endpoint is set at
instantiation.(optional)
:param integer timeout: Allows customization of the timeout for client
http requests. (optional)
Example::
>>> from keystoneclient.v3 import client
>>> keystone = client.Client(username=USER,
password=PASS,
tenant_name=TENANT_NAME,
auth_url=KEYSTONE_URL)
>>> keystone.tenants.list()
...
>>> user = keystone.users.get(USER_ID)
>>> user.delete()
"""
def __init__(self, endpoint=None, **kwargs):
""" Initialize a new client for the Keystone v2.0 API. """
super(Client, self).__init__(endpoint=endpoint, **kwargs)
# NOTE(gabriel): If we have a pre-defined endpoint then we can
# get away with lazy auth. Otherwise auth immediately.
if endpoint:
self.management_url = endpoint
else:
self.authenticate()
def serialize(self, entity):
return json.dumps(entity, sort_keys=True)

0
tests/v3/__init__.py Normal file
View File

225
tests/v3/utils.py Normal file
View File

@@ -0,0 +1,225 @@
import json
import uuid
import time
import urlparse
import httplib2
import mox
import unittest2 as unittest
from keystoneclient.v3 import client
def parameterize(ref):
"""Rewrites attributes to match the kwarg naming convention in client.
>>> paramterize({'project_id': 0})
{'project': 0}
"""
params = ref.copy()
for key in ref:
if key[-3:] == '_id':
params.setdefault(key[:-3], params.pop(key))
return params
class TestCase(unittest.TestCase):
TEST_TENANT_NAME = 'aTenant'
TEST_TOKEN = 'aToken'
TEST_USER = 'test'
TEST_ROOT_URL = 'http://127.0.0.1:5000/'
TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3')
TEST_ROOT_ADMIN_URL = 'http://127.0.0.1:35357/'
TEST_ADMIN_URL = '%s%s' % (TEST_ROOT_ADMIN_URL, 'v3')
def setUp(self):
super(TestCase, self).setUp()
self.mox = mox.Mox()
self._original_time = time.time
time.time = lambda: 1234
httplib2.Http.request = self.mox.CreateMockAnything()
self.client = client.Client(username=self.TEST_USER,
token=self.TEST_TOKEN,
tenant_name=self.TEST_TENANT_NAME,
auth_url=self.TEST_URL,
endpoint=self.TEST_URL)
def tearDown(self):
time.time = self._original_time
super(TestCase, self).tearDown()
self.mox.UnsetStubs()
self.mox.VerifyAll()
class UnauthenticatedTestCase(unittest.TestCase):
""" Class used as base for unauthenticated calls """
TEST_ROOT_URL = 'http://127.0.0.1:5000/'
TEST_URL = '%s%s' % (TEST_ROOT_URL, 'v3')
TEST_ROOT_ADMIN_URL = 'http://127.0.0.1:35357/'
TEST_ADMIN_URL = '%s%s' % (TEST_ROOT_ADMIN_URL, 'v3')
def setUp(self):
super(UnauthenticatedTestCase, self).setUp()
self.mox = mox.Mox()
self._original_time = time.time
time.time = lambda: 1234
httplib2.Http.request = self.mox.CreateMockAnything()
def tearDown(self):
time.time = self._original_time
super(UnauthenticatedTestCase, self).tearDown()
self.mox.UnsetStubs()
self.mox.VerifyAll()
class CrudTests(object):
key = None
collection_key = None
model = None
manager = None
def new_ref(self, **kwargs):
kwargs.setdefault('id', uuid.uuid4().hex)
return kwargs
def additionalSetUp(self):
self.headers = {
'GET': {
'X-Auth-Token': 'aToken',
'User-Agent': 'python-keystoneclient',
}
}
self.headers['DELETE'] = self.headers['GET'].copy()
self.headers['POST'] = self.headers['GET'].copy()
self.headers['POST']['Content-Type'] = 'application/json'
self.headers['PATCH'] = self.headers['POST'].copy()
def serialize(self, entity):
if isinstance(entity, dict):
return json.dumps({self.key: entity}, sort_keys=True)
if isinstance(entity, list):
return json.dumps({self.collection_key: entity}, sort_keys=True)
raise NotImplementedError('Are you sure you want to serialize that?')
def test_create(self):
ref = self.new_ref()
resp = httplib2.Response({
'status': 201,
'body': self.serialize(ref),
})
method = 'POST'
req_ref = ref.copy()
req_ref.pop('id')
httplib2.Http.request(
urlparse.urljoin(
self.TEST_URL,
'v3/%s' % self.collection_key),
method,
body=self.serialize(req_ref),
headers=self.headers[method]) \
.AndReturn((resp, resp['body']))
self.mox.ReplayAll()
returned = self.manager.create(**parameterize(req_ref))
self.assertTrue(isinstance(returned, self.model))
for attr in ref:
self.assertEqual(
getattr(returned, attr),
ref[attr],
'Expected different %s' % attr)
def test_get(self):
ref = self.new_ref()
resp = httplib2.Response({
'status': 200,
'body': self.serialize(ref),
})
method = 'GET'
httplib2.Http.request(
urlparse.urljoin(
self.TEST_URL,
'v3/%s/%s' % (self.collection_key, ref['id'])),
method,
headers=self.headers[method]) \
.AndReturn((resp, resp['body']))
self.mox.ReplayAll()
returned = self.manager.get(ref['id'])
self.assertTrue(isinstance(returned, self.model))
for attr in ref:
self.assertEqual(
getattr(returned, attr),
ref[attr],
'Expected different %s' % attr)
def test_list(self):
ref_list = [self.new_ref(), self.new_ref()]
resp = httplib2.Response({
'status': 200,
'body': self.serialize(ref_list),
})
method = 'GET'
httplib2.Http.request(
urlparse.urljoin(
self.TEST_URL,
'v3/%s' % self.collection_key),
method,
headers=self.headers[method]) \
.AndReturn((resp, resp['body']))
self.mox.ReplayAll()
returned_list = self.manager.list()
self.assertTrue(len(returned_list))
[self.assertTrue(isinstance(r, self.model)) for r in returned_list]
def test_update(self):
ref = self.new_ref()
req_ref = ref.copy()
del req_ref['id']
resp = httplib2.Response({
'status': 200,
'body': self.serialize(ref),
})
method = 'PATCH'
httplib2.Http.request(
urlparse.urljoin(
self.TEST_URL,
'v3/%s/%s' % (self.collection_key, ref['id'])),
method,
body=self.serialize(req_ref),
headers=self.headers[method]) \
.AndReturn((resp, resp['body']))
self.mox.ReplayAll()
returned = self.manager.update(ref['id'], **parameterize(req_ref))
self.assertTrue(isinstance(returned, self.model))
for attr in ref:
self.assertEqual(
getattr(returned, attr),
ref[attr],
'Expected different %s' % attr)
def test_delete(self):
ref = self.new_ref()
method = 'DELETE'
resp = httplib2.Response({
'status': 204,
'body': '',
})
httplib2.Http.request(
urlparse.urljoin(
self.TEST_URL,
'v3/%s/%s' % (self.collection_key, ref['id'])),
method,
headers=self.headers[method]) \
.AndReturn((resp, resp['body']))
self.mox.ReplayAll()
self.manager.delete(ref['id'])