openstack
/
nova
Code Issues Proposed changes

Add API Response class for more complex testing 

the api client .api_(get|delete|post|put) do useful encoding/decoding
of the request/response from dictionaries, and to dictionaries. For
negative and functional testing we're going to often need the status
code as well.

This creates an APIResponse class which wraps and decodes the Requests
library response that is returned. That provides easy access to the
decoded json, as well as access to things like status code and
headers.

It's used in follow on tests to make them more concise.

The convenience methods in the test client (such as get_server) are
adapted to return the same information as before, but through this new
method. These will be further documented in future commits.

Change-Id: I692cbc1a39c3802799de45b4c40306fd717feabf
This commit is contained in:
Sean Dague 2015-02-25 16:53:18 -05:00
parent 5208925c29
commit c6adf3298b
1 changed files with 85 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

115
nova/tests/functional/api/client.py
View File

@ -24,6 +24,50 @@ from nova.tests.unit.image import fake
LOG = logging.getLogger(__name__)
class APIResponse(object):
"""Decoded API Response
This provides a decoded version of the Requests response which
include a json decoded body, far more convenient for testing that
returned structures are correct, or using parts of returned
structures in tests.
This class is a simple wrapper around dictionaries for API
responses in tests. It includes extra attributes so that they can
be inspected in addition to the attributes.
All json responses from Nova APIs are dictionary compatible, or
blank, so other possible base classes are not needed.
"""
status = 200
"""The HTTP status code as an int"""
content = ""
"""The Raw HTTP response body as a string"""
body = {}
"""The decoded json body as a dictionary"""
headers = {}
"""Response headers as a dictionary"""
def __init__(self, response):
"""Construct an API response from a Requests response
:param response: a ``requests`` library response
"""
super(APIResponse, self).__init__()
self.status = response.status_code
self.content = response.content
if self.content:
self.body = jsonutils.loads(self.content)
self.headers = response.headers
def __str__(self):
# because __str__ falls back to __repr__ we can still use repr
# on self but add in the other attributes.
return "<Response body:%r, status_code:%s>" % (self.body, self.status)
class OpenStackApiException(Exception):
def __init__(self, message=None, response=None):
self.response = response
@ -145,17 +189,14 @@ class TestOpenStackClient(object):
return response
def _decode_json(self, response):
body = response.content
LOG.debug("Decoding JSON: %s", body)
if body:
return jsonutils.loads(body)
else:
return ""
resp = APIResponse(status=response.status_code)
if response.content:
resp.body = jsonutils.loads(response.content)
return resp
def api_get(self, relative_uri, **kwargs):
kwargs.setdefault('check_response_status', [200])
response = self.api_request(relative_uri, **kwargs)
return self._decode_json(response)
return APIResponse(self.api_request(relative_uri, **kwargs))
def api_post(self, relative_uri, body, **kwargs):
kwargs['method'] = 'POST'
@ -165,8 +206,7 @@ class TestOpenStackClient(object):
kwargs['body'] = jsonutils.dumps(body)
kwargs.setdefault('check_response_status', [200, 202])
response = self.api_request(relative_uri, **kwargs)
return self._decode_json(response)
return APIResponse(self.api_request(relative_uri, **kwargs))
def api_put(self, relative_uri, body, **kwargs):
kwargs['method'] = 'PUT'
@ -176,16 +216,30 @@ class TestOpenStackClient(object):
kwargs['body'] = jsonutils.dumps(body)
kwargs.setdefault('check_response_status', [200, 202, 204])
response = self.api_request(relative_uri, **kwargs)
return self._decode_json(response)
return APIResponse(self.api_request(relative_uri, **kwargs))
def api_delete(self, relative_uri, **kwargs):
kwargs['method'] = 'DELETE'
kwargs.setdefault('check_response_status', [200, 202, 204])
return self.api_request(relative_uri, **kwargs)
return APIResponse(self.api_request(relative_uri, **kwargs))
#####################################
#
# Convenience methods
#
# The following are a set of convenience methods to get well known
# resources, they can be helpful in setting up resources in
# tests. All of these convenience methods throw exceptions if they
# get a non 20x status code, so will appropriately abort tests if
# they fail.
#
# They all return the most relevant part of their response body as
# decoded data structure.
#
#####################################
def get_server(self, server_id):
return self.api_get('/servers/%s' % server_id)['server']
return self.api_get('/servers/%s' % server_id).body['server']
def get_servers(self, detail=True, search_opts=None):
rel_url = '/servers/detail' if detail else '/servers'
@ -197,75 +251,76 @@ class TestOpenStackClient(object):
if qparams:
query_string = "?%s" % urllib.urlencode(qparams)
rel_url += query_string
return self.api_get(rel_url)['servers']
return self.api_get(rel_url).body['servers']
def post_server(self, server):
response = self.api_post('/servers', server)
response = self.api_post('/servers', server).body
if 'reservation_id' in response:
return response
else:
return response['server']
def put_server(self, server_id, server):
return self.api_put('/servers/%s' % server_id, server)
return self.api_put('/servers/%s' % server_id, server).body
def post_server_action(self, server_id, data):
return self.api_post('/servers/%s/action' % server_id, data)
return self.api_post('/servers/%s/action' % server_id, data).body
def delete_server(self, server_id):
return self.api_delete('/servers/%s' % server_id)
def get_image(self, image_id):
return self.api_get('/images/%s' % image_id)['image']
return self.api_get('/images/%s' % image_id).body['image']
def get_images(self, detail=True):
rel_url = '/images/detail' if detail else '/images'
return self.api_get(rel_url)['images']
return self.api_get(rel_url).body['images']
def post_image(self, image):
return self.api_post('/images', image)['image']
return self.api_post('/images', image).body['image']
def delete_image(self, image_id):
return self.api_delete('/images/%s' % image_id)
def get_flavor(self, flavor_id):
return self.api_get('/flavors/%s' % flavor_id)['flavor']
return self.api_get('/flavors/%s' % flavor_id).body['flavor']
def get_flavors(self, detail=True):
rel_url = '/flavors/detail' if detail else '/flavors'
return self.api_get(rel_url)['flavors']
return self.api_get(rel_url).body['flavors']
def post_flavor(self, flavor):
return self.api_post('/flavors', flavor)['flavor']
return self.api_post('/flavors', flavor).body['flavor']
def delete_flavor(self, flavor_id):
return self.api_delete('/flavors/%s' % flavor_id)
def get_volume(self, volume_id):
return self.api_get('/volumes/%s' % volume_id)['volume']
return self.api_get('/volumes/%s' % volume_id).body['volume']
def get_volumes(self, detail=True):
rel_url = '/volumes/detail' if detail else '/volumes'
return self.api_get(rel_url)['volumes']
return self.api_get(rel_url).body['volumes']
def post_volume(self, volume):
return self.api_post('/volumes', volume)['volume']
return self.api_post('/volumes', volume).body['volume']
def delete_volume(self, volume_id):
return self.api_delete('/volumes/%s' % volume_id)
def get_server_volume(self, server_id, attachment_id):
return self.api_get('/servers/%s/os-volume_attachments/%s' %
(server_id, attachment_id))['volumeAttachment']
(server_id, attachment_id)
).body['volumeAttachment']
def get_server_volumes(self, server_id):
return self.api_get('/servers/%s/os-volume_attachments' %
(server_id))['volumeAttachments']
(server_id)).body['volumeAttachments']
def post_server_volume(self, server_id, volume_attachment):
return self.api_post('/servers/%s/os-volume_attachments' %
(server_id), volume_attachment
)['volumeAttachment']
).body['volumeAttachment']
def delete_server_volume(self, server_id, attachment_id):
return self.api_delete('/servers/%s/os-volume_attachments/%s' %