"Fossies" - the Fresh Open Source Software Archive

Member "keystone-17.0.0/keystone/tests/unit/rest.py" (13 May 2020, 8113 Bytes) of package /linux/misc/openstack/keystone-17.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "rest.py": 16.0.1_vs_17.0.0.

    1 # Copyright 2013 OpenStack Foundation
    2 #
    3 # Licensed under the Apache License, Version 2.0 (the "License"); you may
    4 # not use this file except in compliance with the License. You may obtain
    5 # a copy of the License at
    6 #
    7 #      http://www.apache.org/licenses/LICENSE-2.0
    8 #
    9 # Unless required by applicable law or agreed to in writing, software
   10 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   11 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
   12 # License for the specific language governing permissions and limitations
   13 # under the License.
   14 
   15 import http.client
   16 from oslo_serialization import jsonutils
   17 import webtest
   18 
   19 from keystone.tests import unit
   20 from keystone.tests.unit import default_fixtures
   21 from keystone.tests.unit import ksfixtures
   22 from keystone.tests.unit.ksfixtures import database
   23 
   24 
   25 class RestfulTestCase(unit.TestCase):
   26     """Performs restful tests against the WSGI app over HTTP.
   27 
   28     This class launches public & admin WSGI servers for every test, which can
   29     be accessed by calling ``public_request()`` or ``admin_request()``,
   30     respectfully.
   31 
   32     ``restful_request()`` and ``request()`` methods are also exposed if you
   33     need to bypass restful conventions or access HTTP details in your test
   34     implementation.
   35 
   36     Three new asserts are provided:
   37 
   38     * ``assertResponseSuccessful``: called automatically for every request
   39         unless an ``expected_status`` is provided
   40     * ``assertResponseStatus``: called instead of ``assertResponseSuccessful``,
   41         if an ``expected_status`` is provided
   42     * ``assertValidResponseHeaders``: validates that the response headers
   43         appear as expected
   44 
   45     Requests are automatically serialized according to the defined
   46     ``content_type``. Responses are automatically deserialized as well, and
   47     available in the ``response.body`` attribute. The original body content is
   48     available in the ``response.raw`` attribute.
   49 
   50     """
   51 
   52     # default content type to test
   53     content_type = 'json'
   54 
   55     def setUp(self):
   56         super(RestfulTestCase, self).setUp()
   57 
   58         self.auth_plugin_config_override()
   59 
   60         self.useFixture(database.Database())
   61         self.load_backends()
   62         self.load_fixtures(default_fixtures)
   63 
   64         self.public_app = webtest.TestApp(
   65             self.loadapp(name='public'))
   66         self.addCleanup(delattr, self, 'public_app')
   67 
   68     def auth_plugin_config_override(self, methods=None, **method_classes):
   69         self.useFixture(
   70             ksfixtures.ConfigAuthPlugins(self.config_fixture,
   71                                          methods,
   72                                          **method_classes))
   73 
   74     def request(self, app, path, body=None, headers=None, token=None,
   75                 expected_status=None, **kwargs):
   76         if headers:
   77             headers = {str(k): str(v) for k, v in headers.items()}
   78         else:
   79             headers = {}
   80 
   81         if token:
   82             headers['X-Auth-Token'] = str(token)
   83 
   84         # sets environ['REMOTE_ADDR']
   85         kwargs.setdefault('remote_addr', 'localhost')
   86 
   87         response = app.request(path, headers=headers,
   88                                status=expected_status, body=body,
   89                                **kwargs)
   90 
   91         return response
   92 
   93     def assertResponseSuccessful(self, response):
   94         """Assert that a status code lies inside the 2xx range.
   95 
   96         :param response: :py:class:`httplib.HTTPResponse` to be
   97           verified to have a status code between 200 and 299.
   98 
   99         example::
  100 
  101              self.assertResponseSuccessful(response)
  102         """
  103         self.assertTrue(
  104             200 <= response.status_code <= 299,
  105             'Status code %d is outside of the expected range (2xx)\n\n%s' %
  106             (response.status, response.body))
  107 
  108     def assertResponseStatus(self, response, expected_status):
  109         """Assert a specific status code on the response.
  110 
  111         :param response: :py:class:`httplib.HTTPResponse`
  112         :param expected_status: The specific ``status`` result expected
  113 
  114         example::
  115 
  116             self.assertResponseStatus(response, http.client.NO_CONTENT)
  117         """
  118         self.assertEqual(
  119             expected_status, response.status_code,
  120             'Status code %s is not %s, as expected\n\n%s' %
  121             (response.status_code, expected_status, response.body))
  122 
  123     def assertValidResponseHeaders(self, response):
  124         """Ensure that response headers appear as expected."""
  125         self.assertIn('X-Auth-Token', response.headers.get('Vary'))
  126 
  127     def assertValidErrorResponse(self, response,
  128                                  expected_status=http.client.BAD_REQUEST):
  129         """Verify that the error response is valid.
  130 
  131         Subclasses can override this function based on the expected response.
  132 
  133         """
  134         self.assertEqual(expected_status, response.status_code)
  135         error = response.result['error']
  136         self.assertEqual(response.status_code, error['code'])
  137         self.assertIsNotNone(error.get('title'))
  138 
  139     def _to_content_type(self, body, headers, content_type=None):
  140         """Attempt to encode JSON and XML automatically."""
  141         content_type = content_type or self.content_type
  142 
  143         if content_type == 'json':
  144             headers['Accept'] = 'application/json'
  145             if body:
  146                 headers['Content-Type'] = 'application/json'
  147                 # NOTE(davechen):dump the body to bytes since WSGI requires
  148                 # the body of the response to be `Bytestrings`.
  149                 # see pep-3333:
  150                 # https://www.python.org/dev/peps/pep-3333/#a-note-on-string-types
  151                 return jsonutils.dump_as_bytes(body)
  152 
  153     def _from_content_type(self, response, content_type=None):
  154         """Attempt to decode JSON and XML automatically, if detected."""
  155         content_type = content_type or self.content_type
  156 
  157         if response.body is not None and response.body.strip():
  158             # if a body is provided, a Content-Type is also expected
  159             header = response.headers.get('Content-Type')
  160             self.assertIn(content_type, header)
  161 
  162             if content_type == 'json':
  163                 response.result = jsonutils.loads(response.body)
  164             else:
  165                 response.result = response.body
  166 
  167     def restful_request(self, method='GET', headers=None, body=None,
  168                         content_type=None, response_content_type=None,
  169                         **kwargs):
  170         """Serialize/deserialize json as request/response body.
  171 
  172         .. WARNING::
  173 
  174             * Existing Accept header will be overwritten.
  175             * Existing Content-Type header will be overwritten.
  176 
  177         """
  178         # Initialize headers dictionary
  179         headers = {} if not headers else headers
  180 
  181         body = self._to_content_type(body, headers, content_type)
  182 
  183         # Perform the HTTP request/response
  184         response = self.request(method=method, headers=headers, body=body,
  185                                 **kwargs)
  186 
  187         response_content_type = response_content_type or content_type
  188         self._from_content_type(response, content_type=response_content_type)
  189 
  190         # we can save some code & improve coverage by always doing this
  191         if (method != 'HEAD' and
  192                 response.status_code >= http.client.BAD_REQUEST):
  193             self.assertValidErrorResponse(response)
  194 
  195         # Contains the decoded response.body
  196         return response
  197 
  198     def _request(self, convert=True, **kwargs):
  199         if convert:
  200             response = self.restful_request(**kwargs)
  201         else:
  202             response = self.request(**kwargs)
  203 
  204         self.assertValidResponseHeaders(response)
  205         return response
  206 
  207     def public_request(self, **kwargs):
  208         return self._request(app=self.public_app, **kwargs)
  209 
  210     def admin_request(self, **kwargs):
  211         return self._request(app=self.public_app, **kwargs)
  212 
  213     def _get_token_id(self, r):
  214         """Helper method to return a token ID from a response.
  215 
  216         This needs to be overridden by child classes for on their content type.
  217 
  218         """
  219         raise NotImplementedError()