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
insetUp
and save it in an instance variable for future useadd 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 theService.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 standardAsyncTestCase
supplies and manages the Tornado machinery necessary to glueService
instances to the ioloop. Each external service that you need to test with is represented by a namedService
instance. TheService
instance maintains the list of programmed responses. TheServiceLayer
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 theServiceLayer
instance. They are created on-demand by callingget_service()
with the name that you assign to them. TheServiceLayer
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.
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 aServiceLayer
instance and callServiceLayer.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.
-
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_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
-
Request¶
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')