1#!/usr/bin/env python 2# -*- coding: utf-8 -*- 3 4"""Tests for Requests.""" 5 6from requests import Response 7from os import environ 8from txrequests import Session 9from twisted.trial.unittest import TestCase 10from twisted.internet import defer 11 12HTTPBIN = environ.get('HTTPBIN_URL', 'http://httpbin.org/') 13 14def httpbin(*suffix): 15 """Returns url for HTTPBIN resource.""" 16 return HTTPBIN + '/'.join(suffix) 17 18 19class RequestsTestCase(TestCase): 20 21 @defer.inlineCallbacks 22 def test_session(self): 23 # basic futures get 24 with Session() as sess: 25 d = sess.get(httpbin('get')) 26 self.assertIsInstance(d, defer.Deferred) 27 resp = yield d 28 self.assertIsInstance(resp, Response) 29 self.assertEqual(200, resp.status_code) 30 31 # non-200, 404 32 d = sess.get(httpbin('status/404')) 33 resp = yield d 34 self.assertEqual(404, resp.status_code) 35 36 def cb(s, r): 37 self.assertIsInstance(s, Session) 38 self.assertIsInstance(r, Response) 39 # add the parsed json data to the response 40 r.data = r.json() 41 return r 42 43 d = sess.get(httpbin('get'), background_callback=cb) 44 # this should block until complete 45 resp = yield d 46 self.assertEqual(200, resp.status_code) 47 # make sure the callback was invoked 48 self.assertTrue(hasattr(resp, 'data')) 49 50 def rasing_cb(s, r): 51 raise Exception('boom') 52 53 d = sess.get(httpbin('get'), background_callback=rasing_cb) 54 raised = False 55 try: 56 resp = yield d 57 except Exception as e: 58 self.assertEqual('boom', e.args[0]) 59 raised = True 60 self.assertTrue(raised) 61 62 def test_max_workers(self): 63 """ Tests the `max_workers` shortcut. """ 64 from twisted.python.threadpool import ThreadPool 65 66 with Session() as session: 67 self.assertEqual(session.pool.max, 4) 68 with Session(maxthreads=5) as session: 69 self.assertEqual(session.pool.max, 5) 70 with Session(pool=ThreadPool(maxthreads=10)) as session: 71 self.assertEqual(session.pool.max, 10) 72 with Session(pool=ThreadPool(maxthreads=10), 73 maxthreads=5) as session: 74 self.assertEqual(session.pool.max, 10) 75 76 @defer.inlineCallbacks 77 def test_redirect(self): 78 """ Tests for the ability to cleanly handle redirects. """ 79 with Session() as sess: 80 d = sess.get(httpbin('redirect-to?url=get')) 81 resp = yield d 82 self.assertIsInstance(resp, Response) 83 self.assertEqual(200, resp.status_code) 84 85 d = sess.get(httpbin('redirect-to?url=status/404')) 86 resp = yield d 87 self.assertEqual(404, resp.status_code) 88 89