How can I write a Tornado unit test which tests concurrency

Question

I'm interested in making sure that my Tornado handlers don't block, so I'd like to write some unit tests as a sanity check.

What I have in mind is a handler which sleeps asynchronously for 2 seconds. In a test, I want to call this handler twice in a row in order to simulate "simultaneous" requests.

If I'm not mistaken, both of these requests should be operating concurrently, and thus complete in less than 4 seconds. The issue is I'm not sure how to make 2 simultaneous requests to my application via AsyncHTTPTestCase.

Here is what I have so far:

class SyncSleepHandler(tornado.web.RequestHandler):
    def get(self):
        time.sleep(2)


class AsyncSleepHandler(tornado.web.RequestHandler):
    @gen.coroutine
    def get(self):
        yield gen.sleep(2)


class SleepTest(AsyncHTTPTestCase):
    def get_app(self):
        return Application([(r'/sync', SyncSleepHandler),
                            (r'/async', AsyncSleepHandler)], debug=True)

    def test_async_sleep(self):
        start = time.time()
        resp1 = self.fetch(r'/async', method='GET')
        resp2 = self.fetch(r'/async', method='GET')
        diff = time.time() - start
        self.assertTrue(2 < diff < 4, msg="Difference is {:}".format(diff))

Answer 1

AsyncHTTPTestCase.fetch controls the IOLoop and makes a single fetch, so it cannot be used for this kind of test, but you can go directly to the underlying self.http_client, and use @tornado.testing.gen_test to control the IOLoop:

@gen_test
def test_async_sleep(self):
    start = time.time()
    resp1, resp2 = yield [
        self.http_client.fetch(self.get_url('/async')),
        self.http_client.fetch(self.get_url('/async')),
    ]
    diff = time.time() - start