Unit test HTTPServer creation using the Tornado framework

Question

I'm trying to unit test a Tornado application.

The goal of my test (test_POST_empty_json_in_do_nothing) is now only to send a POST request of an empty zipped json. When it receive the request the HTTPServer must only return an HTTP code 200.

Following this example I override get_http_server to return my HTTPServer using the make_server function. As far ass I understand, in this way the test module will automatically use this server during the tests.

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)

But doing this both the tests fail with:

AssertionError: Async operation timed out after 5 seconds

So my first problem is: why this happen?

Of course if I completely remove the get_http_server function both tests pass, but Tornado also return:

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

And that make sense, because I'm using an HTTPServer that does not have the decompress_request parameter.

I don't understand how can I use in the test module the HTTPServer returned by the make_server function, that is the server created with the parameter I want.

Putting it in another way: how can I test the fact that my server need the decompress_request parameter?

Answer 1

I managed to solve the issue thanks to the help of the guys of Tornado. I report it here what they told me:

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

So I created a new certificate:

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

And used the files created by this command to create a ssl certificate. Than I pass it to the 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)