使用 Tornado 框架对 HTTPServer 创建进行单元测试

Question

我正在尝试对 Tornado 应用程序进行单元测试。

我的测试 (test_POST_empty_json_in_do_nothing) 的目标现在只是发送 POST 空压缩请求 json。当它收到请求时，HTTPServer 必须只 return 一个 HTTP 代码 200.

在 this example 之后，我使用 make_server 函数将 get_http_server 覆盖到 return 我的 HTTPServer。据我了解，通过这种方式，测试模块将在测试期间自动使用该服务器。

test_main_server.py

class TestMainServer(AsyncHTTPSTestCase):

    def get_app(self):
        return ecomtranslatorSrv.make_app()

    def get_http_server(self):
        return ecomtranslatorSrv.make_server(self._app, self.io_loop)

    def test_GET_main_handler(self):
        response = self.fetch('/')
        self.assertEqual(response.code, 200)

    def test_POST_empty_json_in_do_nothing(self):
        headers = tornado.httputil.HTTPHeaders({"Content-Type": "application/json", 'Content-Encoding': 'gzip'})
        response = self.fetch(method='POST', path='/basket/json_in', headers=headers, body='{}')
        self.assertEqual(response.code, 200)


def main():
    tornado.testing.main()


if __name__ == '__main__':
    main()

main_server.py

class MainHandler(tornado.web.RequestHandler):
    def get(self):
        pass

class NewBasketHandler(tornado.web.RequestHandler):
    def post(self):
        pass


def make_app():
    return tornado.web.Application([
        (MAIN_HANDLER, MainHandler),
        (NEW_BASKET, NewBasketHandler)
    ])


def make_server(app, io_loop=None):
    return tornado.httpserver.HTTPServer(app, io_loop=io_loop, decompress_request=True)

但是这样做两个测试都失败了：

AssertionError: Async operation timed out after 5 seconds

所以我的第一个问题是：为什么会这样？

当然，如果我完全删除 get_http_server 函数，两个测试都会通过，但 Tornado 也会 return:

WARNING:tornado.general:Unsupported Content-Encoding: gzip

这是有道理的，因为我使用的 HTTPServer 没有 decompress_request 参数。

我不明白如何在测试模块中使用由 make_server 函数 returned 的 HTTPServer，即使用我想要的参数创建的服务器。

换句话说：我如何测试我的服务器需要 decompress_request 参数这一事实？

Answer 1

感谢 Tornado 的帮助，我设法解决了 issue。我在这里报告他们告诉我的：

You should not override get_http_server() - if you want the decompress_request option, you should return it in get_httpserver_options()

Overriding get_httpserver_options() is the documented way to customize this. As you've seen, you have to rely on the private _app attribute to override get_http_server.

But why is it timing out? You're subclassing AsyncHTTPSTestCase - note the S. This means that you need to create a TLS-enabled server, so you must pass ssl_options to the HTTPServer constructor. Without that you're creating an unencrypted server, which doesn't know how to interpret the TLS handshake (and it doesn't contain the magic \r\n\r\n sequence of bytes, so the server just sits there waiting for the HTTP request to be completed).

所以我创建了一个新证书：

openssl req -x509 -newkey rsa:4096 -sha256 -nodes -keyout test.key -out test.crt -subj "/CN=example.com" -days 3650

并使用此命令创建的文件来创建 ssl 证书。然后我将它传递给 HTTPServer。

def make_server(app):
    ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    ssl_ctx.load_cert_chain(os.path.join(CERTIFICATE_PATH, test.crt), os.path.join(KEY_PATH, test.key))
    return tornado.httpserver.HTTPServer(app, decompress_request=True, ssl_options=ssl_ctx)

使用 Tornado 框架对 HTTPServer 创建进行单元测试

Unit test HTTPServer creation using the Tornado framework

python

unit-testing

tornado