Tornado 测试中只能执行 GET 请求

Question

我正在尝试测试使用 asyncio 和新 async def 语法的 Tornado 请求处理程序。 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()

所以我的问题是；如何使用 unittest 测试 Tornado/asyncio 请求处理程序的 POST 方法？

Answer 1

您的代码在fetch 中缺少body，甚至为空。 post/put 请求需要它 - 我已将工作代码附在底部。

似乎fetch 的（停止/等待）隐藏了有意义的错误。对于测试，我已将其更改为一般异步测试：

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()