glinda

ReadTheDocs Travis CodeClimate

Glinda is a companion library for tornado. It is an attempt to make your time with the framework less painful. In fact, I want to make it downright enjoyable. I started down the path of developing HTTP endpoints in Tornado and needing to test them. The tornado.testing package is handy for testing endpoints in isolation. But what do you do when you have a HTTP service that is calling other HTTP services asynchronously. It turns out that testing that is not as easy as it should be. That is the first thing that I tackled and it is the first thing that this library is going to offer – a way to test non-trivial services.

Once you can test your application, the next step is to write a well-behaved application that fits into the WWW nicely. Tornado does a pretty nice job of handling the nitty gritty HTTP details (e.g., CTE, transfer encodings). It doesn’t provide a clean way to handle representations transparently so I decided to add that into this library as well.

Content Handling

Tornado has some internal content decoding accessible by calling the get_body_arguments method of tornado.web.RequestHandler. It will decode basic form data, application/x-www-form-urlencoded and multipart/form-data specifically. Anything else is left up to you. glinda exposes a content handling mix-in that imbues a standard RequestHandler with a property that is the decoded request body and a new method to encode a response. Here’s what it looks like:

class MyHandler(glinda.content.HandlerMixin, web.RequestHandler):
    def post(self, *args, **kwargs):
        body_argument = self.request_body['arg']
        # do stuff
        self.send_response(response_dict)
        self.finish()

if __name__ == '__main__':
    glinda.content.register_text_type('application/json',
                                      default_charset='utf-8',
                                      dumper=json.dumps, loader=json.loads)
    glinda.content.register_binary_type('application/msgpack',
                                        msgpack.dumpb, msgpack.loadb)

When the client sends a post with a content type of application/json, it will decode the binary body to a string according to the HTTP headers and call json.loads to decode the body when you reference the request_body property. Failures are handled by raising a HTTPError(400) so you don’t have to worry about handling malformed messages. The send_response method will take care of figuring out the appropriate content type based on any included Accept headers. All that you have to do is install encoding and decoding handlers for expected content types.

The glinda.content package implements content handling as described in RFC7231. Specifically, it decodes request bodies as described in section 3.1 and proactive content negotiation as described in sections 3.4.1 and 5.3.

Testing

Here’s an example of testing a Tornado endpoint that asynchronously calls another service. In this case, the application interacts with with the /add endpoint of some other service. Testing in isolation can be tricky without having to have a copy of the service running. You could mock out the AsyncHTTPClient and return fake futures and what not but that has the nasty side-effect of hiding defects around how content type or headers are handled – no HTTP requests means that you have untested assumptions.

The following snippet tests the application under test using the ServiceLayer abstraction that glinda.testing provides.

from tornado import testing
from glinda.testing import services


class MyServiceTests(testing.AsyncHTTPTestCase):

   def setUp(self):
      service_layer = services.ServiceLayer()
      self.service = service_layer['adder']
      # TODO configured your application here using
      # self.service.url_for('/add') or self.service.host
      super(MyServiceTests, self).setUp()

   def get_app(self):
      return MyApplication()

   def test_that_my_service_calls_other_service(self):
      self.service.add_response(
         services.Request('POST', '/add'),
         services.Response(200, body='{"result": 10}'))
      self.fetch(self.get_url('/do-stuff'), method='GET')

      recorded = self.service.get_request('/add')
      self.assertEqual(recorded.method, 'POST')
      self.assertEqual(recorded.body, '[1,2,3,4]')
      self.assertEqual(recorded.headers['Content-Type'], 'application/json')

The application under test is linked in by implementing the standard tornado.testing.AsyncHTTPTestCase.get_app method. Then you add in a glinda.testing.services.ServiceLayer object and configure it to look like the services that you depend on by adding endpoints and then configuring your application to point at the service layer. When you invoke the application under test using self.fetch(...), it will send HTTP requests through the Tornado stack (using the testing ioloop) to the service layer which will respond appropriately. The beauty is that the entire HTTP stack is exercised locally so that you can easily test edge cases such as correct handling of status codes, custom headers, or malformed bodies without resorting to deep mocking.

Documentation

Content Handling

The glinda.content module includes functions and classes that make content negotiation in Tornado easier to extend. Tornado’s web framework includes get_body_argument() which handles a handful of body encodings. It is difficult to add new content type handlers in the current framework. Instead of adding all of the logic into the RequestHandler, glinda.content maintains a mapping from content type to encoder and decoder callables and exposes a mix-in that implements content negotiation over a RequestHandler.

The get_request_body() method added by the HandlerMixin class will decode and cache the request body based on a set of registered content types. You register encoder and decoder functions associated with specific content types when your application starts up and the HandlerMixin will call them when it decodes the request body. Request bodies are exposed from Tornado as raw byte strings. Calling register_binary_type() associates binary transcoding functions with a specific MIME content type.

from glinda import content
import msgpack

content.register_binary_type('application/msgpack', msgpack.dumpb,
                             msgpack.loadb)

The transcoding functions are called to translate between dict and byte representations when the Content-Type header matches the specified value.

Many HTTP payloads are text-based and the protocol includes character set negotiation separately from the content type. The character set of the request body is usually indicated by the charset content parameter ala Content-Type: application/json; charset=utf-8. You can register string- based transcoding functions with register_text_type(). Request body processing will decode the byte string into a str instance according to the detected character set before calling text-based decoding functions. If a character set is not included in the request headers, then an application specified default value is used.

from glinda import content
import json

content.register_text_type('application/json', 'utf-8',
                           json.dumps, json.loads)

Binary registrations are preferred over text since they do not require the character transcoding process.

Once you have registered some content handlers, use the HandlerMixin class to de-serialize requests and serialize responses. The following class mimics the GET and POST functionality of the excellent http://httpbin.org utility site.

class HttpbinHandler(content.HandlerMixin, web.RequestHandler):
    """Mimics http://httpbin.org/{get,post}"""

    def get(self):
        self.send_response(self.standard_response_dict)
        self.finish()

    def post(self):
        response = self.standard_response_dict
        response['data'] = repr(self.request.body)
        response['files'] = {}
        response['form'] = {}
        response['body'] = self.get_request_body()
        self.send_response(response)
        self.finish()

    @property
    def standard_response_dict(self):
        return {
            'args': httpcompat.parse_qs(self.request.query),
            'headers': dict(self.request.headers),
            'origin': self.request.remote_ip,
            'url': self.request.uri,
        }

When you run examples/contentneg.py, it will run a Tornado application listening on port 8000 with at least the JSON content handler enabled. If the msgpack module is available, then the application/x-msgpack content type will be enabled. Likewise for the yaml module and application/yaml. Assuming that you have the PyYAML package installed, then the following examples should work.

A request that explicitly requests a JSON response will get one.

GET / HTTP/1.1
Accept: application/json
Accept-Encoding: gzip, deflate
Connection: keep-alive
Host: localhost:8000
User-Agent: HTTPie/0.9.2

HTTP/1.1 200 OK
Content-Length: 204
Content-Type: application/json; charset=utf-8
Date: Sun, 09 Aug 2015 17:00:30 GMT
Etag: "7bccfbf9d3f99b4b9bc88ec4f27b1913e5c0b27e"
Server: TornadoServer/4.2

{
    "args": {},
    "headers": {
        "Accept": "application/json",
        "Accept-Encoding": "gzip, deflate",
        "Connection": "keep-alive",
        "Host": "localhost:8000",
        "User-Agent": "HTTPie/0.9.2"
    },
    "origin": "::1",
    "url": "/"
}

If you explicitly request application/yaml, then the same data will be encoded as a YAML document.

GET / HTTP/1.1
Accept: application/yaml
Accept-Encoding: gzip, deflate
Connection: keep-alive
Host: localhost:8000
User-Agent: HTTPie/0.9.2

HTTP/1.1 200 OK
Content-Length: 174
Content-Type: application/yaml; charset=utf-8
Date: Sun, 09 Aug 2015 17:04:23 GMT
Etag: "3d88b7fc99bb1b31807e88e4ea3d312d391c037b"
Server: TornadoServer/4.2

args: {}
headers: {Accept: application/yaml, Accept-Encoding: 'gzip, deflate',
  Connection: keep-alive, Host: 'localhost:8000', User-Agent: HTTPie/0.9.2}
origin: ::1
url: /

The request handler simple needs to use HandlerMixin.get_request_body() method to retrieve the request body and HandlerMixin.send_response() to transmit a response body.

Functions

glinda.content.register_binary_type(content_type, dumper, loader)[source]

Register handling for a binary content type.

Parameters

  • content_type (str) – content type to register the hooks for

  • dumper – called to decode bytes into a dictionary. Calling convention: dumper(obj_dict) -> bytes.

  • loader – called to encode a dictionary into a byte string. Calling convention: loader(obj_bytes) -> dict

glinda.content.register_text_type(content_type, default_encoding, dumper, loader)[source]

Register handling for a text-based content type.

Parameters

  • content_type (str) – content type to register the hooks for

  • default_encoding (str) – encoding to use if none is present in the request

  • dumper – called to decode a string into a dictionary. Calling convention: dumper(obj_dict).encode(encoding) -> bytes

  • loader – called to encode a dictionary to a string. Calling convention: loader(obj_bytes.decode(encoding)) -> dict

The decoding of a text content body takes into account decoding the binary request body into a string before calling the underlying dump/load routines.

glinda.content.clear_handlers()[source]

Clears registered type handlers.

Classes

class glinda.content.HandlerMixin(*args, **kwargs)[source]

Mix this in over RequestHandler to enable content handling.

get_request_body()[source]

Decodes the request body and returns it.

Returns

the decoded request body as a dict instance.

Raises

tornado.web.HTTPError if the body cannot be decoded (415) or if decoding fails (400)

property registered_content_types

Yields the currently registered content types in some order.

send_response(response_dict)[source]

Encode a response according to the request.

Parameters

response_dict (dict) – the response to send

Raises

tornado.web.HTTPError if no acceptable content type exists

This method will encode response_dict using the most appropriate encoder based on the Accept request header and the available encoders. The result is written to the client by calling self.write after setting the response content type using self.set_header.

Testing Utilities

glinda.testing.services contains four classes that make testing Tornado applications that make HTTP requests against other services much easier. The problem that I set out to solve was how to exercise as much of the HTTP stack as possible without running an external service in it’s entirety. What I came up with is the ServiceLayer class. It is responsible for creating and maintaining a Application instance that you configure to look and act like the external services that the application under test interacts with during the test run.

The expected usage pattern is as follows:

  1. create an instance of ServiceLayer in setUp and save it in an instance variable for future use

  2. add services by calling ServiceLayer.get_service for each service that your application uses.

  3. configure your application to connect to the fake services. The Service.url_for() method can be used to build URLs or you can retrieve the IP address and port from the Service.host attribute.

  4. add requests and responses using Service.add_response() to configure each specific test before calling your application endpoints

There is a fully functional example below in Example Test

Classes

ServiceLayer

class glinda.testing.services.ServiceLayer[source]

Represents any number of HTTP services.

Create an instance of this class to represent any number of HTTP services that your application depends on. It attaches to the IOLoop instance that the standard AsyncTestCase supplies and manages the Tornado machinery necessary to glue Service instances to the ioloop. Each external service that you need to test with is represented by a named Service instance. The Service instance maintains the list of programmed responses. The ServiceLayer exists to make sure that the requests get to the appropriate handler.

Each managed service is exposed as named Service instance that is owned by the ServiceLayer instance. They are created on-demand by calling get_service() with the name that you assign to them. The ServiceLayer instance implements the same behavior as item lookup as well (e.g., __getitem__):

>>> from tornado.ioloop import IOLoop
>>> services = ServiceLayer()
>>> services['service-name'] is services.get_service('service-name')
True

Once you have a named service instance, you can configure it to respond by calling Service.add_response() method.

>>> from tornado import httpclient, ioloop
>>> services = ServiceLayer()
>>> service = services.get_service('endpoint')
>>> service.add_response(Request('GET', '/endpoint'),
...                      Response(222))
>>> def get():
...    client = httpclient.AsyncHTTPClient()
...    return client.fetch(service.url_for('endpoint'))
>>> rsp = ioloop.IOLoop.instance().run_sync(get)
>>> rsp.code
222
>>>

Note that the get function in the example mimics the application under test asynchronously interacting with a service within the service layer.

get_service(service)[source]

Retrieve a named service, creating it if necessary.

Parameters

service (str) – name to assign to the service

Returns

a Service instance

This method creates the new Service instance and wires it into the tornado stack listening on its own port number. If the service already exists, then it is returned without modification.

Service

class glinda.testing.services.Service(name, add_resource_callback)[source]

Represents a logical HTTP service.

A service is a collection of HTTP resources that are available from a ephemeral port that the service sets up when it is created. When the instance is create, it is isolated behind a unique socket (and port number). The url_for() method will construct and return a URL containing the ephemeral port number and host name so that you can easily interact with the service.

A service is a collection of related endpoints that is attached to a specific port on localhost. It is responsible for keeping track of the programmed responses and dispatching them when requests come in. Responses are added to the service with add_response(). They will be returned from the request handler associated with a resource path in the order that they are added.

Note that you should not create Service instances yourself. If you do, they will not be wired into the Tornado framework appropriately. Instead, you should create a ServiceLayer instance and call ServiceLayer.get_service() to create services.

add_endpoint(*path)[source]

Add an endpoint without configuring a response.

Parameters

path – resource path

You only need to call this method if you want to create a resource without configuring a response. Otherwise, you should call add_response() which will create the resource if necessary.

add_response(request, response)[source]

Configure the service to respond to a specific request.

Parameters

  • request (Request) – request to match against

  • response (Response) – response to return when the handler receives request

assert_request(method, *path, **query)[source]

Assert that a specific request was made to the service.

Parameters

  • method (str) – the HTTP method to match

  • path – the resource to match

  • query – optional query parameters to match

Raises

AssertionError if no matching request to the service is found

get_next_response(tornado_request)[source]

Retrieve the next response for a request.

Parameters

tornado_request (tornado.httputil.HTTPRequest) –

Responses are matched to the request using the request method and URI as a key. If a response was registered for the method and URI, then it is popped and returned. If there is no response configured for tornado_request, then an exception is raised.

Raises

tornado.web.HTTPError – with a status of 456 if the service doesn’t have a response configured for tornado_request

get_request(*path)[source]

Convenience method to fetch a single request.

get_requests_for(*path)[source]

Retrieve the requests made for path.

Parameters

path – resource to retrieve requests for

This method returns the requests made for path in the order that they were made. It uses a generator to return value values so either call it in a loop or use next() to get the first value.

Returns

Request instances made for the resource by way of a generator

Raises

AssertionError if no requests were made for the resource

record_request(request)[source]

Record a client request to a service.

Parameters

request (tornado.httputil.HTTPRequest) – client request made to one of the services endpoints

url_for(*path, **query)[source]

Retrieve a URL that targets the service.

Parameters

  • path – list of path elements

  • query – optional query parameters to include in the URL

Returns

the full URL that targets the address and port for this service and includes the specified path and query

Request

class glinda.testing.services.Request(method, *path)[source]

Matches a request from a client.

Parameters

  • method (str) – HTTP method to match

  • path – optional resource path to match

Instances of this class are used by Service instances to identify patterns that a client will request.

Response

class glinda.testing.services.Response(status, reason=None, body=None, headers=None)[source]

Records a response from the server.

Parameters

  • status (int) – HTTP status code to return

  • reason (str) – optional phrase to return on the status line

  • body (bytes) – optional payload to return

  • headers (dict) – optional response headers

Example Test

"""
Example of using glinda to test a service.

This file is an example of using the ``glinda.testing`` package to
test a very simple Tornado application.

"""
import os

from tornado import gen, httpclient, web
import tornado.testing

from glinda.testing import services


class MyHandler(web.RequestHandler):
    """
    Simple handler that makes a few asynchronous requests.

    The "backing" service is configured by setting the ``SERVICE_NETLOC``
    environment variable to the host and port of the service endpoint.
    This is pretty typical for 12 Factor Applications and it makes it
    very easy to test the service using ``glinda``.

    """

    @gen.coroutine
    def get(self):
        netloc = os.environ['SERVICE_NETLOC']
        client = httpclient.AsyncHTTPClient()
        try:
            yield client.fetch('http://{0}/status'.format(netloc),
                               method='HEAD')
        except web.HTTPError as error:
            if error.code >= 300:
                raise web.HTTPError(504)

        try:
            response = yield client.fetch('http://{0}/do-stuff'.format(netloc),
                                          method='POST',
                                          body='important stuff')
        except web.HTTPError as error:
            if error.code >= 300:
                raise web.HTTPError(500)

        self.set_status(200)
        self.set_header('Custom', response.headers.get('Custom', ''))
        self.write(response.body)
        self.finish()


class HandlerTests(tornado.testing.AsyncHTTPTestCase):
    """Traditional :class:`tornado.testing.AsyncHTTPTestCase`"""

    def setUp(self):
        super(HandlerTests, self).setUp()
        service_layer = services.ServiceLayer()
        self.external_service = service_layer['status']
        os.environ['SERVICE_NETLOC'] = self.external_service.host

    def get_app(self):
        return web.Application([web.url('/do-the-things', MyHandler)])

    def test_that_status_failure_results_in_504(self):
        self.external_service.add_response(
            services.Request('HEAD', '/status'), services.Response(500))
        try:
            self.fetch('/do-the-things')
        except web.HTTPError as error:
            self.assertEqual(error.status_code, 504)

    def test_that_service_failure_results_in_500(self):
        self.external_service.add_response(
            services.Request('HEAD', '/status'), services.Response(200))
        self.external_service.add_response(
            services.Request('POST', '/do-stuff'), services.Response(400))
        try:
            self.fetch('/do-the-things')
        except web.HTTPError as error:
            self.assertEqual(error.status_code, 500)

    def test_that_status_is_fetched(self):
        self.external_service.add_response(
            services.Request('HEAD', '/status'), services.Response(200))
        self.external_service.add_response(
            services.Request('POST', '/do-stuff'), services.Response(200))
        self.fetch('/do-the-things')
        self.external_service.assert_request('HEAD', '/status')

    def test_that_stuff_is_posted(self):
        self.external_service.add_response(
            services.Request('HEAD', '/status'), services.Response(200))
        self.external_service.add_response(
            services.Request('POST', '/do-stuff'), services.Response(200))
        self.fetch('/do-the-things')

        request = self.external_service.get_request('do-stuff')
        self.assertEqual(request.body, b'important stuff')

    def test_that_post_response_is_preserved(self):
        self.external_service.add_response(
            services.Request('HEAD', '/status'), services.Response(200))
        self.external_service.add_response(
            services.Request('POST', '/do-stuff'),
            services.Response(200, body='foo', headers={'Custom': 'header'}))

        response = self.fetch('/do-the-things')
        self.assertEqual(response.body.decode(), 'foo')
        self.assertEqual(response.headers['Custom'], 'header')

Hacking on this Code

I’m always open to contributions from others provided that the are not completely worthless. I do request that you follow a few basic rules though.

  • (Almost) All code is tested: if you think that your contributions do not require tests for whatever reason, please justify it.

  • All tests pass. This is non-negotiable. If you cannot get a particular test case to pass. Submit the PR on github and we can probably work through it. What I won’t do is accept code that breaks a pile of tests without any effort to fix them. Run env/bin/tox to run the tests.

  • Features are documented. I use sphinx for all documentation. Edit the documentation sources as required and make sure that running python setup.py build_sphinx produces something reasonable.

I apologize for being pretty opinionated about this but code quality doesn’t disappear suddenly, it continually degrades. This set of minimum requirements is my attempt to help prevent quality issues.

Getting Started

Congratulations for not being immediately put off by my attitude ;) Let’s get started. First thing is to create a new sandbox for you to play in. If you are using Python 3.4 or newer, use pyvenv env to create a new virtual environment in the env directory. Otherwise, install virtualenv and run virtualenv env to create the environment.

PLEASE DO NOT USE SUDO! DO NOT INSTALL INTO YOUR SYSTEM ENVIRONMENT!!

If you did not chuckle at the previous line, then you really do need to pay careful attention to it. Use virtual environments for developing code.

Once you have an environment, install the requirements listed in dev-requirements.txt. This is a pip-formatted requirements file that will pull in packages that the code depends on as well as packages required to test, document, and generally hack on the code base.

TL;DR

$ python3 -mvenv env
$ env/bin/pip install -r dev-requirements.txt

Running Tests

Tests are written using the standard unittest package and run using nose and tox.

$ env/bin/tox

Documentation

All documentation is written in ReStructuredText with HTML generated by the sphinx utility.

$ env/bin/python setup.py build_sphinx

The HTML output is written to build/sphinx/html.

Changelog