Ситуация
Если ваш сервис внутри асинхронных обработчиков сам создает асинхронные задачи - например, для того чтобы забрать что-то с внешнего сервиса - в стиле
class MyAsyncHandler(tornado.web.RequestHandler):
@tornado.web.asynchronous
def get(self, *args, **kwargs):
try:
# тут чего то берем с внешнего сервера
self.auth_request({})
except Exception, ex:
self.error('Async - %s'%ex)
self.reply(None)
def auth_request(self, params, callback=None):
url = self.application.settings['external_service'] + '?' + urlencode(params)
if not callback:
callback = self._on_load
http = httpclient.AsyncHTTPClient()
http.fetch(url, callback, validate_cert=False)
def _on_load(self, response):
# обрабатываем полученные результат
pass
То при написании юнит-тестов через AsyncHTTPTestCase в подобном ключе по официальной документации(!!!)
class MyTestServer(AsyncHTTPTestCase):
def get_host(self):
return 'http://%s:%d'%(SERVER['host'], SERVER['port'])
def get_app(self):
myserver = MyServer()
myserver.run(SERVER['host'], SERVER['port'])
return myserver.app # приложение Tornado
def get_response_obj(self, url, **kwargs):
if len(kwargs) > 0:
url += '?' + urllib.urlencode(kwargs)
logging.info('Request [%s]'%url)
self.http_client.fetch(url, self.stop)
response = self.wait()
print response
if response.body:
return response.body
return None
При запуске тестов начинаем ловить потрясающие спецэффекты,
Traceback (most recent call last):
File "/home/xxxx/test_wait.py", line 9, in test_1
self.wait(timeout = 5)
File "/xxxxx/site-packages/tornado/
testing.py", line 169, in timeout_func
timeout)
AssertionError: Async operation timed out after 5 seconds
Лечится это добавлением в класс теста переопределенного метода get_new_ioloopclass MyTestServer(AsyncHTTPTestCase):
....
def get_new_ioloop(self):
return ioloop.IOLoop.instance()
Такие дела...
Тема в гуглогруппе, Gist с примером
