Может выполнять только запросы GET в тесте Tornado
Я пытаюсь протестировать обработчик запросов Tornado, который использует asyncio и новый async def
синтаксис. GET запросы, кажется, работают, но другие запросы терпят неудачу.
import asyncio
import tornado.platform.asyncio as tasyncio
from tornado import testing
from tornado import web
class Handler(web.RequestHandler):
async def post(self):
await asyncio.sleep(1)
self.write('spam')
get = put = post
class HandlerTest(testing.AsyncHTTPTestCase):
def get_app(self):
return web.Application([('/', Handler)])
def get_new_ioloop(self):
return tasyncio.AsyncIOMainLoop()
def test_get(self):
response = self.fetch('/', method='GET')
self.assertEqual(b'spam', response.body)
def test_put(self):
response = self.fetch('/', method='PUT')
self.assertEqual(b'spam', response.body)
def test_post(self):
response = self.fetch('/', method='POST')
self.assertEqual(b'spam', response.body)
При запуске обработчика в интерактивном интерпретаторе он работает нормально:
>>> app = web.Application([('/', Handler)])
>>> tornado.platform.asyncio.AsyncIOMainLoop().install()
>>> app.listen(8080)
<tornado.httpserver.HTTPServer object at 0x7fbf290f3fd0>
>>> asyncio.get_event_loop().run_forever()
Итак, мой вопрос: Как я могу проверить методы POST обработчика запросов Tornado/asyncio с unittest?
1 ответ
Решение
Ваш код отсутствует body
в fetch
даже пустой. Требуется для запросов post/put - рабочий код я приложил внизу.
Кажется, что fetch
s (остановка / ожидание) скрывает значимую ошибку. Для теста я изменил его на общий асинхронный тест:
from tornado.testing import gen_test
class HandlerTest(testing.AsyncHTTPTestCase):
def get_app(self):
return web.Application([('/', Handler)])
def get_new_ioloop(self):
return tasyncio.AsyncIOMainLoop()
@gen_test
def test_put(self):
response = yield self.http_client.fetch(self.get_url('/'), method='PUT')
self.assertEqual(b'spam', response.body)
Выдает правильную ошибку
======================================================================
ERROR: test_put (__main__.HandlerTest)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/tmp/t35/lib/python3.5/site-packages/tornado/testing.py", line 132, in __call__
result = self.orig_method(*args, **kwargs)
File "/tmp/t35/lib/python3.5/site-packages/tornado/testing.py", line 525, in post_coroutine
timeout=timeout)
File "/tmp/t35/lib/python3.5/site-packages/tornado/ioloop.py", line 453, in run_sync
return future_cell[0].result()
File "/tmp/t35/lib/python3.5/site-packages/tornado/concurrent.py", line 232, in result
raise_exc_info(self._exc_info)
File "<string>", line 3, in raise_exc_info
File "/tmp/t35/lib/python3.5/site-packages/tornado/gen.py", line 1014, in run
yielded = self.gen.throw(*exc_info)
File "/tmp/t35/lib/python3.5/types.py", line 179, in throw
return self.__wrapped.throw(tp, *rest)
File "test.py", line 35, in test_put
response = yield self.http_client.fetch(self.get_url('/'), method='PUT')
File "/tmp/t35/lib/python3.5/site-packages/tornado/gen.py", line 1008, in run
value = future.result()
File "/tmp/t35/lib/python3.5/site-packages/tornado/concurrent.py", line 232, in result
raise_exc_info(self._exc_info)
File "<string>", line 3, in raise_exc_info
File "/tmp/t35/lib/python3.5/site-packages/tornado/stack_context.py", line 314, in wrapped
ret = fn(*args, **kwargs)
File "/tmp/t35/lib/python3.5/site-packages/tornado/gen.py", line 264, in <lambda>
future, lambda future: callback(future.result()))
File "/tmp/t35/lib/python3.5/site-packages/tornado/simple_httpclient.py", line 353, in _on_connect
('not ' if body_expected else '', self.request.method))
ValueError: Body must not be None for method PUT (unless allow_nonstandard_methods is true)
----------------------------------------------------------------------
Ran 1 test in 0.008s
Рабочий код:
import tornado.platform.asyncio as tasyncio
from tornado import testing
from tornado import web
class Handler(web.RequestHandler):
async def post(self):
await asyncio.sleep(0.1)
self.write('spam')
get = put = post
class HandlerTest(testing.AsyncHTTPTestCase):
def get_app(self):
return web.Application([('/', Handler)])
def get_new_ioloop(self):
return tasyncio.AsyncIOMainLoop()
def test_get(self):
response = self.fetch('/', method='GET')
self.assertEqual(b'spam', response.body)
def test_put(self):
response = self.fetch('/', method='PUT', body=b'')
self.assertEqual(b'spam', response.body)
def test_post(self):
response = self.fetch('/', method='POST', body=b'')
self.assertEqual(b'spam', response.body)
import unittest
unittest.main()