Как я могу написать единичный тест Tornado, который проверяет параллелизм

Question

Я заинтересован в том, чтобы мои обработчики Tornado не блокировались, поэтому я хотел бы написать некоторые модульные тесты как проверку работоспособности.

Что я имею в виду, это обработчик, который асинхронно спит в течение 2 секунд. В тесте я хочу дважды вызвать этот обработчик, чтобы имитировать «одновременные» запросы.

Если я не ошибаюсь, оба этих запроса должны работать одновременно и, таким образом, быть завершены менее чем за 4 секунды. Проблема в том, что я не уверен, как сделать 2 одновременных запроса к моему приложению через AsyncHTTPTestCase.

Вот то, что я до сих пор:

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 управляет IOLoop и делает один выборку, поэтому он не может быть использован для такого рода испытаний, но вы можете перейти непосредственно к нижележащим self.http_client, и использовать @tornado.testing.gen_test для управления 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