Filtered by Tornado

Page 3

Reset

My tricks for using AsyncHTTPClient in Tornado

October 13, 2010
1 comment Python, Tornado

I've been doing more and more web development with Tornado recently. It's got an awesome class for running client HTTP calls in your integration tests. To run a normal GET it looks something like this:


from tornado.testing import AsyncHTTPTestCase
class ApplicationTestCase(AsyncHTTPTestCase):
   def get_app(self):
       return app.Application(database_name='test', xsrf_cookies=False)

   def test_homepage(self):
       url = '/'
       self.http_client.fetch(self.get_url(url), self.stop)
       response = self.wait()
       self.assertTrue('Click here to login' in response.body)

Now, to run a POST request you can use the same client. It looks something like this:


   def test_post_entry(self):
       url = '/entries'
       data = dict(comment='Test comment')
       from urllib import urlencode
       self.http_client.fetch(self.get_url(url), self.stop, 
                              method="POST",
                              data=urlencode(data))
       response = self.wait()
       self.assertEqual(response.code, 302)

That's fine but it gets a bit verbose after a while. So instead I've added this little cute mixin class:


from urllib import urlencode

class HTTPClientMixin(object):

   def get(self, url, data=None, headers=None):
       if data is not None:
           if isinstance(data, dict):
               data = urlencode(data)
           if '?' in url:
               url += '&%s' % data
           else:
               url += '?%s' % data
       return self._fetch(url, 'GET', headers=headers)

   def post(self, url, data, headers=None):
       if data is not None:
           if isinstance(data, dict):
               data = urlencode(data)
       return self._fetch(url, 'POST', data, headers)

   def _fetch(self, url, method, data=None, headers=None):
       self.http_client.fetch(self.get_url(url), self.stop, method=method,
                              body=data, headers=headers)
       return self.wait()

Now you can easily write some brief and neat tests:


class ApplicationTestCase(AsyncHTTPTestCase, HTTPClientMixin):
   def get_app(self):
       return app.Application(database_name='test', xsrf_cookies=False)

   def test_homepage(self):
       response = self.get('/')
       self.assertTrue('Click here to login' in response.body)

   def test_post_entry(self):
       # rendering the homepage creates a user and sets a cookie
       response = self.get('/')

       user_id_cookie = re.findall('user_id=([\w\|]+);', 
                                 response.headers['Set-Cookie'])[0]
       cookie = 'user_id=%s;' % user_id_cookie
       import base64
       guid = base64.b64decode(user_id_cookie.split('|')[0])
       self.assertEqual(db.users.User.find(
           {'_id':ObjectId(user_id_cookie)}).count(), 1)

       data = dict(comment='Test comment')
       response = self.post('/entries', data, headers={'Cookie': cookie})
       self.assertEqual(response.code, 302)
       self.assertTrue('/thanks' in response.headers['Location'])

So far it's just a neat wrapper to save me some typing and it makes the actual tests look a lot neater. I haven't tested this in anger yet and there might be several interesting corner cases surrounding headers and POST data and what not. Hopefully people can chip in and share ideas on this snippet and perhaps I can fork this into Tornado's core

Previous page
Next page