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