tornadogenexample.py

from tornado.web import asynchronous, RequestHandler, Application
from tornado import gen
from tornado.testing import AsyncHTTPTestCase, gen_test
from tornado.httpclient import AsyncHTTPClient, HTTPRequest, HTTPResponse
from mock import patch, Mock
from tornado.concurrent import Future
import StringIO


class FooHandler(RequestHandler):
    @asynchronous # means that we take care of finishing the request
    @gen.coroutine # will contain a yield statement
    def get(self):
        result = yield self.some_gen()
        self.finish(result + "Foo!")

    def get_client(self):
        """To be able to mock away the real http client and leave the http
        client used for testing alone"""
        return AsyncHTTPClient()

    @gen.coroutine
    def some_gen(self):
        client = self.get_client()
        # client.fetch can be mocked
        response = yield client.fetch('http://something_that_will_never_resolve/')
        # now we can post-process the response
        raise gen.Return(response.body[:100])


class TestFooHandler(AsyncHTTPTestCase):

    def get_app(self):
        application = Application([
            (r"/", FooHandler),
            ])
        return application

    @gen_test
    def test_get(self):
        with patch.object(FooHandler, 'get_client') as get_client:
            setup_fetch(get_client().fetch, 200, 'Bar!')
            response = yield self.http_client.fetch(self.get_url('/'))
            print response.body
            assert response.body.startswith('Bar!')
            assert response.body.endswith('Foo!')


def setup_fetch(fetch_mock, status_code, body=None):
    def side_effect(request, **kwargs):
        if request is not HTTPRequest:
            request = HTTPRequest(request)
        buffer = StringIO.StringIO(body)
        response = HTTPResponse(request, status_code, None, buffer)
        future = Future()
        future.set_result(response)
        return future

    fetch_mock.side_effect = side_effect