¡@
OpenStack Study: rest.py**** CubicPower OpenStack Study ****
# Copyright 2013 OpenStack Foundation # # 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 io from lxml import etree import six import webtest from keystone.auth import controllers as auth_controllers from keystone.common import serializer from keystone.openstack.common import jsonutils from keystone import tests from keystone.tests import default_fixtures
**** CubicPower OpenStack Study **** class RestfulTestCase(tests.TestCase): """Performs restful tests against the WSGI app over HTTP. This class launches public & admin WSGI servers for every test, which can be accessed by calling ``public_request()`` or ``admin_request()``, respectfully. ``restful_request()`` and ``request()`` methods are also exposed if you need to bypass restful conventions or access HTTP details in your test implementation. Three new asserts are provided: * ``assertResponseSuccessful``: called automatically for every request unless an ``expected_status`` is provided * ``assertResponseStatus``: called instead of ``assertResponseSuccessful``, if an ``expected_status`` is provided * ``assertValidResponseHeaders``: validates that the response headers appear as expected Requests are automatically serialized according to the
**** CubicPower OpenStack Study **** def setUp(self, app_conf='keystone'): super(RestfulTestCase, self).setUp() # Will need to reset the plug-ins self.addCleanup(setattr, auth_controllers, 'AUTH_METHODS', {}) self.load_backends() self.load_fixtures(default_fixtures) self.public_app = webtest.TestApp( self.loadapp(app_conf, name='main')) self.addCleanup(delattr, self, 'public_app') self.admin_app = webtest.TestApp( self.loadapp(app_conf, name='admin')) self.addCleanup(delattr, self, 'admin_app')
**** CubicPower OpenStack Study **** def request(self, app, path, body=None, headers=None, token=None, expected_status=None, **kwargs): if headers: headers = dict([(str(k), str(v)) for k, v in six.iteritems(headers)]) else: headers = {} if token: headers['X-Auth-Token'] = str(token) # setting body this way because of: # https://github.com/Pylons/webtest/issues/71 if body: kwargs['body_file'] = io.BytesIO(body) # sets environ['REMOTE_ADDR'] kwargs.setdefault('remote_addr', 'localhost') response = app.request(path, headers=headers, status=expected_status, **kwargs) return response
**** CubicPower OpenStack Study **** def assertResponseSuccessful(self, response): """Asserts that a status code lies inside the 2xx range. :param response: :py:class:`httplib.HTTPResponse` to be verified to have a status code between 200 and 299. example:: self.assertResponseSuccessful(response, 203) """ self.assertTrue( response.status_code >= 200 and response.status_code <= 299, 'Status code %d is outside of the expected range (2xx)\n\n%s' % (response.status, response.body))
**** CubicPower OpenStack Study **** def assertResponseStatus(self, response, expected_status): """Asserts a specific status code on the response. :param response: :py:class:`httplib.HTTPResponse` :param expected_status: The specific ``status`` result expected example:: self.assertResponseStatus(response, 203) """ self.assertEqual( response.status_code, expected_status, 'Status code %s is not %s, as expected)\n\n%s' % (response.status_code, expected_status, response.body))
**** CubicPower OpenStack Study **** def assertValidResponseHeaders(self, response): """Ensures that response headers appear as expected.""" self.assertIn('X-Auth-Token', response.headers.get('Vary'))
**** CubicPower OpenStack Study **** def assertValidErrorResponse(self, response, expected_status=400): """Verify that the error response is valid. Subclasses can override this function based on the expected response. """ self.assertEqual(response.status_code, expected_status) error = response.result['error'] self.assertEqual(error['code'], response.status_code) self.assertIsNotNone(error.get('title'))
**** CubicPower OpenStack Study **** def _to_content_type(self, body, headers, content_type=None): """Attempt to encode JSON and XML automatically.""" content_type = content_type or self.content_type if content_type == 'json': headers['Accept'] = 'application/json' if body: headers['Content-Type'] = 'application/json' return jsonutils.dumps(body) elif content_type == 'xml': headers['Accept'] = 'application/xml' if body: headers['Content-Type'] = 'application/xml' return serializer.to_xml(body)
**** CubicPower OpenStack Study **** def _from_content_type(self, response, content_type=None): """Attempt to decode JSON and XML automatically, if detected.""" content_type = content_type or self.content_type if response.body is not None and response.body.strip(): # if a body is provided, a Content-Type is also expected header = response.headers.get('Content-Type') self.assertIn(content_type, header) if content_type == 'json': response.result = jsonutils.loads(response.body) elif content_type == 'xml': response.result = etree.fromstring(response.body)
**** CubicPower OpenStack Study **** def restful_request(self, method='GET', headers=None, body=None, content_type=None, **kwargs): """Serializes/deserializes json/xml as request/response body. .. WARNING:: * Existing Accept header will be overwritten. * Existing Content-Type header will be overwritten. """ # Initialize headers dictionary headers = {} if not headers else headers body = self._to_content_type(body, headers, content_type) # Perform the HTTP request/response response = self.request(method=method, headers=headers, body=body, **kwargs) self._from_content_type(response, content_type) # we can save some code & improve coverage by always doing this if method != 'HEAD' and response.status_code >= 400: self.assertValidErrorResponse(response) # Contains the decoded response.body return response
**** CubicPower OpenStack Study **** def _request(self, convert=True, **kwargs): if convert: response = self.restful_request(**kwargs) else: response = self.request(**kwargs) self.assertValidResponseHeaders(response) return response
**** CubicPower OpenStack Study **** def public_request(self, **kwargs): return self._request(app=self.public_app, **kwargs)
**** CubicPower OpenStack Study **** def admin_request(self, **kwargs): return self._request(app=self.admin_app, **kwargs)
**** CubicPower OpenStack Study **** def _get_token(self, body): """Convenience method so that we can test authenticated requests.""" r = self.public_request(method='POST', path='/v2.0/tokens', body=body) return self._get_token_id(r)
**** CubicPower OpenStack Study **** def get_unscoped_token(self): """Convenience method so that we can test authenticated requests.""" return self._get_token({ 'auth': { 'passwordCredentials': { 'username': self.user_foo['name'], 'password': self.user_foo['password'], }, }, })
**** CubicPower OpenStack Study **** def get_scoped_token(self, tenant_id=None): """Convenience method so that we can test authenticated requests.""" if not tenant_id: tenant_id = self.tenant_bar['id'] return self._get_token({ 'auth': { 'passwordCredentials': { 'username': self.user_foo['name'], 'password': self.user_foo['password'], }, 'tenantId': tenant_id, }, })
**** CubicPower OpenStack Study **** def _get_token_id(self, r): """Helper method to return a token ID from a response. This needs to be overridden by child classes for on their content type. """ raise NotImplementedError()
|