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:

create an instance of ServiceLayer in setUp and save it in an instance variable for future use
add services by calling ServiceLayer.get_service for each service that your application uses.
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.
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')