Unit testing a Tornado web application

[Jeremy Whitlock]

Hi All,
All examples and usage patterns I see show the ioloop being
started and running indefinitely and control stops there. What I need
is a way to have the Tornado server startup before my tests, run my
tests and then stop the Tornado server. I've found a lot in the group
and on Google about daemonizing and such but before I go through all
of that trouble, I wanted to make sure that was the only option. Are
there any best practices for this? I'll continue to search the group
and Google in the mean time.

Take care,

Jeremy

[Bret Taylor]

For our tests, we use a pattern like:

ioloop = IOLoop.instance()
def callback():
   ioloop.stop()
server = HTTPServer(..)
server.listen(...)
runTests(callback)
ioloop.start()

runTests does a bunch of httpclient requests, calling callback when the requests are complete.

Bret

[Jeremy Whitlock]

> For our tests, we use a pattern like:
>
> ioloop = IOLoop.instance()
> def callback():
>    ioloop.stop()
> server = HTTPServer(..)
> server.listen(...)
> runTests(callback)
> ioloop.start()
>
> runTests does a bunch of httpclient requests, calling callback when the
> requests are complete.

I assume to do this you have to do some threading, so your call to
runTests doesn't complete before ioloop.start() calls right?

[Bret Taylor]

No, we typically run test with httpclient, which is non-blocking, so the tests will not start until IOLoop starts.

[Jeremy Whitlock]

> No, we typically run test with httpclient, which is non-blocking, so the
> tests will not start until IOLoop starts.

Alright. I'll see if I can conjure up a little something. Are there
any plans to put some unit tests in the tornadoweb source code?

[Jeremy Whitlock]

> Alright.  I'll see if I can conjure up a little something.  Are there
> any plans to put some unit tests in the tornadoweb source code?

For those in the same situation, I found some information in
httpclient.py that backs up what Bret was talking about. Here is an
example:

import ioloop

def handle_request(response):
if response.error:
print "Error:", response.error
else:
print response.body

ioloop.IOLoop.instance().stop()

http_client = httpclient.AsyncHTTPClient()
http_client.fetch("http://www.google.com/", handle_request)
ioloop.IOLoop.instance().start()

Once I get a more specific example, like running TestSuite or
something, I'll update. If anyone else has any sample code that might
help, feel free to post.

[Jeremy Whitlock]

Here is a more complete example. Basically, it shows how you can take
the "Hello World!" tutorial and make an httpclient.fetch() call to the
application and get its output:

import tornado.httpserver
import tornado.httpclient
import tornado.ioloop
import tornado.web

class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write("Hello, world")

application = tornado.web.Application([
(r"/", MainHandler),
])

def handle_request(response):
if response.error:
print "Error:", response.error
else:
print response.body

tornado.ioloop.IOLoop.instance().stop()

if __name__ == "__main__":
http_server = tornado.httpserver.HTTPServer(application)
http_server.listen(8888)

http_client = tornado.httpclient.AsyncHTTPClient()
http_client.fetch("http://localhost:8888/", handle_request)

tornado.ioloop.IOLoop.instance().start()

I still need to come up with a working example that would make this
testable, like getting the output from the httpclient.fetch() and
comparing it to an expected string.

[Jeremy Whitlock]

Finally...a fully-functional Python unittest example where we have a
Tornado web application that is being tested:

import tornado.httpserver
import tornado.httpclient
import tornado.ioloop
import tornado.web

import unittest

class MainHandler(tornado.web.RequestHandler):
def get(self):
self.write('Hello, world')

class TestTornadoWeb(unittest.TestCase):
response = None

def setUp(self):

application = tornado.web.Application([
(r'/', MainHandler),
])

http_server = tornado.httpserver.HTTPServer(application)
http_server.listen(8888)

def handle_request(self, response):
self.response = response

tornado.ioloop.IOLoop.instance().stop()

def testHelloWorldHandler(self):

http_client = tornado.httpclient.AsyncHTTPClient()
http_client.fetch('http://localhost:8888/',

self.handle_request)

tornado.ioloop.IOLoop.instance().start()

self.failIf(self.response.error)
self.assertEqual(self.response.body, 'Hello, world')

if __name__ == '__main__':
unittest.main()

[Elias Torres]

How much further did you get with this test?

If I add any more tests in the same class it fails. This is because the socket is still bound to that HTTPServer. I did some hackery to remove the handler from the ioloop but I was wondering if there's an official way to stop the HTTPServer.

-Elias

[Elias Torres]

I added this and things are working now for me:

class HTTPServer(object):

...

def stop(self):
self.io_loop.remove_handler(self._socket.fileno())
self._socket.close()

-Elias Torres

[David P. Novakovic]

Hey Elias,

Can you post your full example? :)

David

[Elias Torres]

On Nov 25, 2009, at 10:01 PM, David P. Novakovic wrote:

Hey Elias,

Can you post your full example? :)

OK. Here we go:

By just adding a second method, Jeremy's unit test breaks.

    def testHelloWorldHandler2(self): 

        http_client = tornado.httpclient.AsyncHTTPClient() 

        http_client.fetch('http://localhost:8888/', self.handle_request) 

        tornado.ioloop.IOLoop.instance().start() 

        self.failIf(self.response.error) 

        self.assertEqual(self.response.body, 'Hello, world') 

if __name__ == '__main__': 

    unittest.main() 

Output:

[elias@manta] ~$ python e.py

.E

======================================================================

ERROR: testHelloWorldHandler2 (__main__.TestTornadoWeb)

----------------------------------------------------------------------

Traceback (most recent call last):

  File "e.py", line 16, in setUp

    http_server.listen(8888)

  File "/Users/elias/Development/tornado/tornado/httpserver.py", line 101, in listen

    self._socket.bind((address, port))

  File "<string>", line 1, in bind

error: [Errno 48] Address already in use

----------------------------------------------------------------------

Ran 2 tests in 0.004s

FAILED (errors=1)

Now if I do this:

import tornado.httpserver 

import tornado.httpclient 

import tornado.ioloop 

import tornado.web 

import unittest 

class MainHandler(tornado.web.RequestHandler): 

    def get(self): 

        self.write('Hello, world') 

class TestTornadoWeb(unittest.TestCase): 

    http_server = None                                                                                                                                                                                                                                                                                                                                                    

    response = None 

    def setUp(self): 

        application = tornado.web.Application([ 

                (r'/', MainHandler), 

                ])  

        self.http_server = tornado.httpserver.HTTPServer(application) 

        self.http_server.listen(8888) 

    def tearDown(self):

        self.http_server.stop()

    def handle_request(self, response): 

        self.response = response 

        tornado.ioloop.IOLoop.instance().stop() 

    def testHelloWorldHandler(self): 

        http_client = tornado.httpclient.AsyncHTTPClient() 

        http_client.fetch('http://localhost:8888/', self.handle_request) 

        tornado.ioloop.IOLoop.instance().start() 

        self.failIf(self.response.error) 

        self.assertEqual(self.response.body, 'Hello, world') 

    def testHelloWorldHandler2(self): 

        http_client = tornado.httpclient.AsyncHTTPClient() 

        http_client.fetch('http://localhost:8888/', self.handle_request) 

        tornado.ioloop.IOLoop.instance().start() 

        self.failIf(self.response.error) 

        self.assertEqual(self.response.body, 'Hello, world') 

if __name__ == '__main__': 

    unittest.main() 

Pay attention to keeping track of the http_server to stop it after the test has been completed.

[elias@manta] ~$ python e.py 

..

----------------------------------------------------------------------

Ran 2 tests in 0.005s

OK

Here's basically what I did.

diff --git a/tornado/httpserver.py b/tornado/httpserver.py

index 386b80b..3b1efd0 100644

--- a/tornado/httpserver.py

+++ b/tornado/httpserver.py

@@ -102,6 +102,10 @@ class HTTPServer(object):

         self._socket.listen(128)

         self.io_loop.add_handler(self._socket.fileno(), self._handle_events,

                                  self.io_loop.READ)

+                                 

+    def stop(self):

+      self.io_loop.remove_handler(self._socket.fileno())

+      self._socket.close()

 

     def _handle_events(self, fd, events):

         while True:

-Elias

[David P. Novakovic]

Awesome, thanks a lot for that.

David

[Elias Torres]

Bret,

Did you see this thread about adding a stop method to the HTTPServer for unit testing?

-Elias

[Van]

Is it possible to authenticate using the test? If say, I'm using the
Google auth mixin, how would I test a page that requires the
authentication?

[Alek Storm]

I'm not sure what Tornado looked like in 2009, but nowadays we have the `testing` module to provide TestCase superclasses for unit testing. Look in tornado/test/web_test.py for examples.

Alek

On Tue, Jul 3, 2012 at 2:50 AM, Richard Wong <cha...@gmail.com> wrote:

Anybody using Webtest to deal with this problem?

Or I misunderstood the webtest's purpose.

[A. Jesse Jiryu Davis]

I wrote two posts on unittesting in Tornado, presenting some more intricate techniques; curious what people think of them:

http://emptysquare.net/blog/tornado-unittesting-eventually-correct/

http://emptysquare.net/blog/tornado-unittesting-with-generators/

[Ben Darnell]

You should check out the tornado.testing module. AsyncTestCase's
stop() and wait() methods make testing async code a lot easier. The
last example from your "eventually correct" post would look like:

class MyTest(tornado.testing.AsyncTestCase):
def get_new_ioloop(self):
# use the singleton IOLoop because async_calculate doesn't take
# an IOLoop as an argument. If it did, you could remove this function
# and pass self.io_loop to async_calculate, which would provide
# a little extra isolation between tests.
return IOLoop.instance()

def test_calculate(self):
async_calculate(callback=self.stop)
result = self.wait()
self.assertEqual(result, 42)

The stop/wait idiom used here predates gen.engine, although now that
we have it it would be nice to support generator-style tests as well.

-Ben

[Richard Wong]

Wow, It's really great. also, in some extent, intricate.

[Richard Wong]

Awesome! use the origin test tools may obtain a better support. ;)