Imported Upstream version 0.10.1

1 files changed, 265 insertions, 102 deletions

diff --git a/test_requests.py b/test_requests.py
index 61953a3..29de3ed 100755
--- a/test_requests.py
+++ b/test_requests.py
@@ -1,27 +1,29 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
 
-from __future__ import with_statement
 
+
+import io
+import json
 import time
 import os
+import sys
 import unittest
 
 import requests
-import envoy
+from requests.compat import str, bytes, StringIO
+# import envoy
 from requests import HTTPError
+from requests.auth import HTTPBasicAuth, HTTPDigestAuth
+
 
-try:
-    import omnijson as json
-except ImportError:
-    import json
 
+if (sys.platform == 'win32') and ('HTTPBIN_URL' not in os.environ):
+    os.environ['HTTPBIN_URL'] = 'http://httpbin.org/'
 
 # TODO: Detect an open port.
 PORT = os.environ.get('HTTPBIN_PORT', '7077')
-
-HTTPBIN_URL = 'http://0.0.0.0:%s/' % (PORT)
-# HTTPBIN_URL = 'http://127.0.0.1:8000/'
+HTTPBIN_URL = os.environ.get('HTTPBIN_URL', 'http://0.0.0.0:%s/' % (PORT))
 
 
 def httpbin(*suffix):
@@ -44,20 +46,10 @@ class RequestsTestSuite(unittest.TestCase):
 
         global _httpbin
 
-        if not _httpbin:
-
-            c = envoy.connect('gunicorn httpbin:app --bind=0.0.0.0:%s' % (PORT))
-
-            self.httpbin = c
+        if (not 'HTTPBIN_URL' in os.environ) and not _httpbin:
+            # c = envoy.connect('httpbin %s' % (PORT))
+            # time.sleep(1)
             _httpbin = True
-            time.sleep(1)
-
-
-
-    def tearDown(self):
-        """Teardown."""
-        # self.httpbin.kill()
-        pass
 
 
     def test_entry_points(self):
@@ -80,6 +72,11 @@ class RequestsTestSuite(unittest.TestCase):
         r = requests.get(httpbin('/get'))
         self.assertEqual(r.status_code, 200)
 
+    def test_response_sent(self):
+        r = requests.get(httpbin('/get'))
+
+        self.assertTrue(r.request.sent)
+
     def test_HTTP_302_ALLOW_REDIRECT_GET(self):
         r = requests.get(httpbin('redirect', '1'))
         self.assertEqual(r.status_code, 200)
@@ -94,7 +91,7 @@ class RequestsTestSuite(unittest.TestCase):
 
         r = requests.get(httpbin('user-agent'), headers=heads)
 
-        assert heads['User-agent'] in r.content
+        assert heads['User-agent'] in r.text
         self.assertEqual(r.status_code, 200)
 
 
@@ -114,7 +111,7 @@ class RequestsTestSuite(unittest.TestCase):
         }
 
         r = requests.get(httpbin('user-agent'), headers=heads);
-        self.assertTrue(heads['User-agent'] in r.content)
+        self.assertTrue(heads['User-agent'] in r.text)
 
         heads = {
             'user-agent':
@@ -122,7 +119,7 @@ class RequestsTestSuite(unittest.TestCase):
         }
 
         r = requests.get(httpbin('user-agent'), headers=heads);
-        self.assertTrue(heads['user-agent'] in r.content)
+        self.assertTrue(heads['user-agent'] in r.text)
 
 
     def test_HTTP_200_OK_HEAD(self):
@@ -135,21 +132,39 @@ class RequestsTestSuite(unittest.TestCase):
         self.assertEqual(r.status_code, 200)
 
 
-    def test_HTTP_200_OK_PATCH(self):
-        r = requests.patch(httpbin('patch'))
-        self.assertEqual(r.status_code, 200)
+    def test_BASICAUTH_TUPLE_HTTP_200_OK_GET(self):
+
+        for service in SERVICES:
+
+            auth = ('user', 'pass')
+            url = service('basic-auth', 'user', 'pass')
+
+            r = requests.get(url, auth=auth)
+            self.assertEqual(r.status_code, 200)
+
+            r = requests.get(url)
+            self.assertEqual(r.status_code, 401)
+
+
+            s = requests.session(auth=auth)
+            r = s.get(url)
+            self.assertEqual(r.status_code, 200)
 
 
     def test_BASICAUTH_HTTP_200_OK_GET(self):
 
         for service in SERVICES:
 
-            auth = ('user', 'pass')
+            auth = HTTPBasicAuth('user', 'pass')
             url = service('basic-auth', 'user', 'pass')
 
             r = requests.get(url, auth=auth)
             self.assertEqual(r.status_code, 200)
 
+            auth = ('user', 'pass')
+            r = requests.get(url, auth=auth)
+            self.assertEqual(r.status_code, 200)
+
             r = requests.get(url)
             self.assertEqual(r.status_code, 401)
 
@@ -163,7 +178,7 @@ class RequestsTestSuite(unittest.TestCase):
 
         for service in SERVICES:
 
-            auth = ('digest', 'user', 'pass')
+            auth = HTTPDigestAuth('user', 'pass')
             url = service('digest-auth', 'auth', 'user', 'pass')
 
             r = requests.get(url, auth=auth)
@@ -249,11 +264,31 @@ class RequestsTestSuite(unittest.TestCase):
         r.raise_for_status()
 
 
+    def test_default_status_raising(self):
+        config = {'danger_mode': True}
+        args = [httpbin('status', '404')]
+        kwargs = dict(config=config)
+        self.assertRaises(HTTPError, requests.get, *args, **kwargs)
+
+        r = requests.get(httpbin('status', '200'))
+        self.assertEqual(r.status_code, 200)
+
+
     def test_decompress_gzip(self):
 
         r = requests.get(httpbin('gzip'))
         r.content.decode('ascii')
 
+    def test_response_has_unicode_url(self):
+
+        for service in SERVICES:
+
+            url = service('get')
+
+            response = requests.get(url)
+
+            assert isinstance(response.url, str)
+
 
     def test_unicode_get(self):
 
@@ -261,20 +296,20 @@ class RequestsTestSuite(unittest.TestCase):
 
             url = service('/get')
 
-            requests.get(url, params={'foo': u'føø'})
-            requests.get(url, params={u'føø': u'føø'})
+            requests.get(url, params={'foo': 'føø'})
+            requests.get(url, params={'føø': 'føø'})
             requests.get(url, params={'føø': 'føø'})
-            requests.get(url, params={'foo': u'foo'})
-            requests.get(service('ø'), params={'foo': u'foo'})
+            requests.get(url, params={'foo': 'foo'})
+            requests.get(service('ø'), params={'foo': 'foo'})
 
 
     def test_httpauth_recursion(self):
 
-        http_auth = ('user', 'BADpass')
+        http_auth = HTTPBasicAuth('user', 'BADpass')
 
         for service in SERVICES:
             r = requests.get(service('basic-auth', 'user', 'pass'), auth=http_auth)
-            self.assertEquals(r.status_code, 401)
+            self.assertEqual(r.status_code, 401)
 
 
     def test_urlencoded_post_data(self):
@@ -283,14 +318,14 @@ class RequestsTestSuite(unittest.TestCase):
 
             r = requests.post(service('post'), data=dict(test='fooaowpeuf'))
 
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post'))
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.url, service('post'))
 
-            rbody = json.loads(r.content)
+            rbody = json.loads(r.text)
 
-            self.assertEquals(rbody.get('form'), dict(test='fooaowpeuf'))
-            self.assertEquals(rbody.get('data'), '')
+            self.assertEqual(rbody.get('form'), dict(test='fooaowpeuf'))
+            self.assertEqual(rbody.get('data'), '')
 
 
     def test_nonurlencoded_post_data(self):
@@ -299,15 +334,16 @@ class RequestsTestSuite(unittest.TestCase):
 
             r = requests.post(service('post'), data='fooaowpeuf')
 
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post'))
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.url, service('post'))
 
-            rbody = json.loads(r.content)
+            rbody = json.loads(r.text)
             # Body wasn't valid url encoded data, so the server returns None as
             # "form" and the raw body as "data".
-            self.assertEquals(rbody.get('form'), {})
-            self.assertEquals(rbody.get('data'), 'fooaowpeuf')
+
+            assert rbody.get('form') in (None, {})
+            self.assertEqual(rbody.get('data'), 'fooaowpeuf')
 
 
     def test_urlencoded_post_querystring(self):
@@ -316,13 +352,13 @@ class RequestsTestSuite(unittest.TestCase):
 
             r = requests.post(service('post'), params=dict(test='fooaowpeuf'))
 
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post?test=fooaowpeuf'))
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.url, service('post?test=fooaowpeuf'))
 
-            rbody = json.loads(r.content)
-            self.assertEquals(rbody.get('form'), {}) # No form supplied
-            self.assertEquals(rbody.get('data'), '')
+            rbody = json.loads(r.text)
+            self.assertEqual(rbody.get('form'), {}) # No form supplied
+            self.assertEqual(rbody.get('data'), '')
 
 
     def test_urlencoded_post_query_and_data(self):
@@ -334,13 +370,13 @@ class RequestsTestSuite(unittest.TestCase):
                 params=dict(test='fooaowpeuf'),
                 data=dict(test2="foobar"))
 
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post?test=fooaowpeuf'))
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.url, service('post?test=fooaowpeuf'))
 
-            rbody = json.loads(r.content)
-            self.assertEquals(rbody.get('form'), dict(test2='foobar'))
-            self.assertEquals(rbody.get('data'), '')
+            rbody = json.loads(r.text)
+            self.assertEqual(rbody.get('form'), dict(test2='foobar'))
+            self.assertEqual(rbody.get('data'), '')
 
 
     def test_nonurlencoded_postdata(self):
@@ -349,13 +385,13 @@ class RequestsTestSuite(unittest.TestCase):
 
             r = requests.post(service('post'), data="foobar")
 
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
 
-            rbody = json.loads(r.content)
+            rbody = json.loads(r.text)
 
-            self.assertEquals(rbody.get('form'), {})
-            self.assertEquals(rbody.get('data'), 'foobar')
+            assert rbody.get('form') in (None, {})
+            self.assertEqual(rbody.get('data'), 'foobar')
 
 
     # def test_idna(self):
@@ -368,8 +404,8 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.get(service('get'), params=dict(test=['foo','baz']))
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.url, service('get?test=foo&test=baz'))
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.url, service('get?test=foo&test=baz'))
 
 
     def test_urlencoded_post_querystring_multivalued(self):
@@ -377,13 +413,13 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.post(service('post'), params=dict(test=['foo','baz']))
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post?test=foo&test=baz'))
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.url, service('post?test=foo&test=baz'))
 
-            rbody = json.loads(r.content)
-            self.assertEquals(rbody.get('form'), {}) # No form supplied
-            self.assertEquals(rbody.get('data'), '')
+            rbody = json.loads(r.text)
+            self.assertEqual(rbody.get('form'), {}) # No form supplied
+            self.assertEqual(rbody.get('data'), '')
 
 
     def test_urlencoded_post_query_multivalued_and_data(self):
@@ -395,12 +431,16 @@ class RequestsTestSuite(unittest.TestCase):
                 params=dict(test=['foo','baz']),
                 data=dict(test2="foobar",test3=['foo','baz']))
 
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post?test=foo&test=baz'))
-            rbody = json.loads(r.content)
-            self.assertEquals(rbody.get('form'), dict(test2='foobar',test3='foo'))
-            self.assertEquals(rbody.get('data'), '')
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(r.headers['content-type'], 'application/json')
+            self.assertEqual(r.url, service('post?test=foo&test=baz'))
+
+            # print(r.text)
+            # print('-----------------------')
+
+            rbody = json.loads(r.text)
+            self.assertEqual(rbody.get('form'), dict(test2='foobar',test3=['foo','baz']))
+            self.assertEqual(rbody.get('data'), '')
 
 
     def test_GET_no_redirect(self):
@@ -408,8 +448,8 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.get(service('redirect', '3'), allow_redirects=False)
-            self.assertEquals(r.status_code, 302)
-            self.assertEquals(len(r.history), 0)
+            self.assertEqual(r.status_code, 302)
+            self.assertEqual(len(r.history), 0)
 
 
     def test_HEAD_no_redirect(self):
@@ -417,8 +457,8 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.head(service('redirect', '3'), allow_redirects=False)
-            self.assertEquals(r.status_code, 302)
-            self.assertEquals(len(r.history), 0)
+            self.assertEqual(r.status_code, 302)
+            self.assertEqual(len(r.history), 0)
 
 
     def test_redirect_history(self):
@@ -426,8 +466,8 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.get(service('redirect', '3'))
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(len(r.history), 3)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(len(r.history), 3)
 
 
     def test_relative_redirect_history(self):
@@ -435,8 +475,8 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.get(service('relative-redirect', '3'))
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(len(r.history), 3)
+            self.assertEqual(r.status_code, 200)
+            self.assertEqual(len(r.history), 3)
 
 
     def test_session_HTTP_200_OK_GET(self):
@@ -455,17 +495,76 @@ class RequestsTestSuite(unittest.TestCase):
 
         # Make 2 requests from Session object, should send header both times
         r1 = s.get(httpbin('user-agent'))
-        assert heads['User-agent'] in r1.content
+        assert heads['User-agent'] in r1.text
 
         r2 = s.get(httpbin('user-agent'))
-        assert heads['User-agent'] in r2.content
+        assert heads['User-agent'] in r2.text
 
         new_heads = {'User-agent': 'blah'}
         r3 = s.get(httpbin('user-agent'), headers=new_heads)
-        assert new_heads['User-agent'] in r3.content
+        assert new_heads['User-agent'] in r3.text
 
         self.assertEqual(r2.status_code, 200)
 
+    def test_single_hook(self):
+
+        def add_foo_header(args):
+            if not args.get('headers'):
+                args['headers'] = {}
+
+            args['headers'].update({
+                'X-Foo': 'foo'
+            })
+
+            return args
+
+        for service in SERVICES:
+            url = service('headers')
+
+            response = requests.get(
+                url = url,
+                hooks = {
+                    'args': add_foo_header
+                }
+            )
+
+            assert 'foo' in response.text
+
+    def test_multiple_hooks(self):
+
+        def add_foo_header(args):
+            if not args.get('headers'):
+                args['headers'] = {}
+
+            args['headers'].update({
+                'X-Foo': 'foo'
+            })
+
+            return args
+
+        def add_bar_header(args):
+            if not args.get('headers'):
+                args['headers'] = {}
+
+            args['headers'].update({
+                'X-Bar': 'bar'
+            })
+
+            return args
+
+        for service in SERVICES:
+            url = service('headers')
+
+            response = requests.get(
+                url = url,
+                hooks = {
+                    'args': [add_foo_header, add_bar_header]
+                }
+            )
+
+            assert 'foo' in response.text
+            assert 'bar' in response.text
+
     def test_session_persistent_cookies(self):
 
         s = requests.session()
@@ -476,35 +575,35 @@ class RequestsTestSuite(unittest.TestCase):
         r = s.get(httpbin('cookies'))
 
         # Those cookies persist transparently.
-        c = json.loads(r.content).get('cookies')
+        c = json.loads(r.text).get('cookies')
         assert c == _c
 
         # Double check.
         r = s.get(httpbin('cookies'), cookies={})
-        c = json.loads(r.content).get('cookies')
+        c = json.loads(r.text).get('cookies')
         assert c == _c
 
         # Remove a cookie by setting it's value to None.
         r = s.get(httpbin('cookies'), cookies={'bessie': None})
-        c = json.loads(r.content).get('cookies')
+        c = json.loads(r.text).get('cookies')
         del _c['bessie']
         assert c == _c
 
         # Test session-level cookies.
         s = requests.session(cookies=_c)
         r = s.get(httpbin('cookies'))
-        c = json.loads(r.content).get('cookies')
+        c = json.loads(r.text).get('cookies')
         assert c == _c
 
         # Have the server set a cookie.
         r = s.get(httpbin('cookies', 'set', 'k', 'v'), allow_redirects=True)
-        c = json.loads(r.content).get('cookies')
+        c = json.loads(r.text).get('cookies')
 
         assert 'k' in c
 
         # And server-set cookie persistience.
         r = s.get(httpbin('cookies'))
-        c = json.loads(r.content).get('cookies')
+        c = json.loads(r.text).get('cookies')
 
         assert 'k' in c
 
@@ -519,23 +618,23 @@ class RequestsTestSuite(unittest.TestCase):
 
         # Make 2 requests from Session object, should send header both times
         r1 = s.get(httpbin('get'))
-        assert params['a'] in r1.content
+        assert params['a'] in r1.text
 
 
         params2 = {'b': 'b_test'}
 
         r2 = s.get(httpbin('get'), params=params2)
-        assert params['a'] in r2.content
-        assert params2['b'] in r2.content
+        assert params['a'] in r2.text
+        assert params2['b'] in r2.text
 
 
         params3 = {'b': 'b_test', 'a': None, 'c': 'c_test'}
 
         r3 = s.get(httpbin('get'), params=params3)
 
-        assert not params['a'] in r3.content
-        assert params3['b'] in r3.content
-        assert params3['c'] in r3.content
+        assert not params['a'] in r3.text
+        assert params3['b'] in r3.text
+        assert params3['c'] in r3.text
 
     def test_invalid_content(self):
         # WARNING: if you're using a terrible DNS provider (comcast),
@@ -553,6 +652,70 @@ class RequestsTestSuite(unittest.TestCase):
         r = requests.get(hah, allow_redirects=False, config=config)
         assert r.content == None
 
+    def test_cached_response(self):
+
+        r1 = requests.get(httpbin('get'), prefetch=False)
+        assert not r1._content
+        assert r1.content
+        assert r1.text
+
+        r2 = requests.get(httpbin('get'), prefetch=True)
+        assert r2._content
+        assert r2.content
+        assert r2.text
+
+    def test_iter_lines(self):
+
+        lines = (0, 2, 10, 100)
+
+        for i in lines:
+            r = requests.get(httpbin('stream', str(i)), prefetch=False)
+            lines = list(r.iter_lines())
+            len_lines = len(lines)
+
+            self.assertEqual(i, len_lines)
+
+        # Test 'dangling' fragment in responses that do not terminate in
+        # a newline.
+        quote = (
+            '''Why will he not upon our fair request\n'''
+            '''Untent his person and share the air with us?'''
+        )
+
+        # Make a request and monkey-patch its contents
+        r = requests.get(httpbin('get'))
+        r.raw = StringIO(quote)
+
+        # Make sure iter_lines doesn't chop the trailing bit
+        lines = '\n'.join(r.iter_lines())
+        self.assertEqual(lines, quote)
+
+    def test_safe_mode(self):
+
+        safe = requests.session(config=dict(safe_mode=True))
+
+        # Safe mode creates empty responses for failed requests.
+        # Iterating on these responses should produce empty sequences
+        r = safe.get('http://_/')
+        self.assertEqual(list(r.iter_lines()), [])
+        assert isinstance(r.error, requests.exceptions.ConnectionError)
+
+        r = safe.get('http://_/')
+        self.assertEqual(list(r.iter_content()), [])
+        assert isinstance(r.error, requests.exceptions.ConnectionError)
+
+        # When not in safe mode, should raise Timeout exception
+        self.assertRaises(
+            requests.exceptions.Timeout,
+            requests.get,
+            httpbin('stream', '1000'), timeout=0.0001)
+
+        # In safe mode, should return a blank response
+        r = requests.get(httpbin('stream', '1000'), timeout=0.0001,
+                config=dict(safe_mode=True))
+        assert r.content is None
+        assert isinstance(r.error, requests.exceptions.Timeout)
+
 
 if __name__ == '__main__':
     unittest.main()