Imported Upstream version 0.8.2

1 files changed, 176 insertions, 119 deletions

diff --git a/test_requests.py b/test_requests.py
index 83b827a..61953a3 100755
--- a/test_requests.py
+++ b/test_requests.py
@@ -3,24 +3,25 @@
 
 from __future__ import with_statement
 
+import time
+import os
 import unittest
-import cookielib
+
+import requests
+import envoy
+from requests import HTTPError
 
 try:
     import omnijson as json
 except ImportError:
     import json
 
-import requests
-
-from requests.sessions import Session
 
+# TODO: Detect an open port.
+PORT = os.environ.get('HTTPBIN_PORT', '7077')
 
-HTTPBIN_URL = 'http://httpbin.ep.io/'
-HTTPSBIN_URL = 'https://httpbin.ep.io/'
-
-# HTTPBIN_URL = 'http://staging.httpbin.org/'
-# HTTPSBIN_URL = 'https://httpbin-staging.ep.io/'
+HTTPBIN_URL = 'http://0.0.0.0:%s/' % (PORT)
+# HTTPBIN_URL = 'http://127.0.0.1:8000/'
 
 
 def httpbin(*suffix):
@@ -29,15 +30,9 @@ def httpbin(*suffix):
     return HTTPBIN_URL + '/'.join(suffix)
 
 
-def httpsbin(*suffix):
-    """Returns url for HTTPSBIN resource."""
-
-    return HTTPSBIN_URL + '/'.join(suffix)
-
-
-SERVICES = (httpbin, httpsbin)
-
+SERVICES = (httpbin, )
 
+_httpbin = False
 
 class RequestsTestSuite(unittest.TestCase):
     """Requests test cases."""
@@ -46,33 +41,52 @@ class RequestsTestSuite(unittest.TestCase):
     _multiprocess_can_split_ = True
 
     def setUp(self):
-        pass
+
+        global _httpbin
+
+        if not _httpbin:
+
+            c = envoy.connect('gunicorn httpbin:app --bind=0.0.0.0:%s' % (PORT))
+
+            self.httpbin = c
+            _httpbin = True
+            time.sleep(1)
+
 
 
     def tearDown(self):
         """Teardown."""
+        # self.httpbin.kill()
         pass
 
 
+    def test_entry_points(self):
+
+        requests.session
+        requests.session().get
+        requests.session().head
+        requests.get
+        requests.head
+        requests.put
+        requests.patch
+        requests.post
+
+
+
     def test_invalid_url(self):
         self.assertRaises(ValueError, requests.get, 'hiwpefhipowhefopw')
 
-
     def test_HTTP_200_OK_GET(self):
-        r = requests.get(httpbin('/'))
+        r = requests.get(httpbin('/get'))
         self.assertEqual(r.status_code, 200)
 
     def test_HTTP_302_ALLOW_REDIRECT_GET(self):
-	r = requests.get(httpbin('redirect', '1'))
-	self.assertEqual(r.status_code, 200)
+        r = requests.get(httpbin('redirect', '1'))
+        self.assertEqual(r.status_code, 200)
 
     def test_HTTP_302_GET(self):
-	r = requests.get(httpbin('redirect', '1'), allow_redirects=False)
-	self.assertEqual(r.status_code, 302)
-
-    def test_HTTPS_200_OK_GET(self):
-        r = requests.get(httpsbin('/'))
-        self.assertEqual(r.status_code, 200)
+        r = requests.get(httpbin('redirect', '1'), allow_redirects=False)
+        self.assertEqual(r.status_code, 302)
 
 
     def test_HTTP_200_OK_GET_WITH_PARAMS(self):
@@ -112,12 +126,7 @@ class RequestsTestSuite(unittest.TestCase):
 
 
     def test_HTTP_200_OK_HEAD(self):
-        r = requests.head(httpbin('/'))
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_HTTPS_200_OK_HEAD(self):
-        r = requests.head(httpsbin('/'))
+        r = requests.head(httpbin('/get'))
         self.assertEqual(r.status_code, 200)
 
 
@@ -126,22 +135,12 @@ class RequestsTestSuite(unittest.TestCase):
         self.assertEqual(r.status_code, 200)
 
 
-    def test_HTTPS_200_OK_PUT(self):
-        r = requests.put(httpsbin('put'))
-        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_HTTPS_200_OK_PATCH(self):
-        r = requests.patch(httpsbin('patch'))
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_AUTH_HTTP_200_OK_GET(self):
+    def test_BASICAUTH_HTTP_200_OK_GET(self):
 
         for service in SERVICES:
 
@@ -149,14 +148,35 @@ class RequestsTestSuite(unittest.TestCase):
             url = service('basic-auth', 'user', 'pass')
 
             r = requests.get(url, auth=auth)
-            # print r.__dict__
             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_DIGESTAUTH_HTTP_200_OK_GET(self):
+
+        for service in SERVICES:
+
+            auth = ('digest', 'user', 'pass')
+            url = service('digest-auth', '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_POSTBIN_GET_POST_FILES(self):
 
         for service in SERVICES:
@@ -206,7 +226,7 @@ class RequestsTestSuite(unittest.TestCase):
             r = requests.get(service('status', '500'))
             self.assertEqual(bool(r), False)
 
-            r = requests.get(service('/'))
+            r = requests.get(service('/get'))
             self.assertEqual(bool(r), True)
 
 
@@ -215,38 +235,20 @@ class RequestsTestSuite(unittest.TestCase):
         for service in SERVICES:
 
             r = requests.get(service('status', '404'))
+            # print r.status_code
+            # r.raise_for_status()
             self.assertEqual(r.ok, False)
 
 
     def test_status_raising(self):
         r = requests.get(httpbin('status', '404'))
-        self.assertRaises(requests.HTTPError, r.raise_for_status)
+        self.assertRaises(HTTPError, r.raise_for_status)
 
         r = requests.get(httpbin('status', '200'))
         self.assertFalse(r.error)
         r.raise_for_status()
 
 
-    def test_cookie_jar(self):
-
-        jar = cookielib.CookieJar()
-        self.assertFalse(jar)
-
-        url = httpbin('cookies', 'set', 'requests_cookie', 'awesome')
-        r = requests.get(url, cookies=jar)
-        self.assertTrue(jar)
-
-        cookie_found = False
-        for cookie in jar:
-            if cookie.name == 'requests_cookie':
-                self.assertEquals(cookie.value, 'awesome')
-                cookie_found = True
-        self.assertTrue(cookie_found)
-
-        r = requests.get(httpbin('cookies'), cookies=jar)
-        self.assertTrue('awesome' in r.content)
-
-
     def test_decompress_gzip(self):
 
         r = requests.get(httpbin('gzip'))
@@ -257,7 +259,7 @@ class RequestsTestSuite(unittest.TestCase):
 
         for service in SERVICES:
 
-            url = service('/')
+            url = service('/get')
 
             requests.get(url, params={'foo': u'føø'})
             requests.get(url, params={u'føø': u'føø'})
@@ -275,19 +277,6 @@ class RequestsTestSuite(unittest.TestCase):
             self.assertEquals(r.status_code, 401)
 
 
-    def test_settings(self):
-
-        def test():
-            r = requests.get(httpbin(''))
-            r.raise_for_status()
-
-        with requests.settings(timeout=0.0000001):
-            self.assertRaises(requests.Timeout, test)
-
-        with requests.settings(timeout=100):
-            requests.get(httpbin(''))
-
-
     def test_urlencoded_post_data(self):
 
         for service in SERVICES:
@@ -317,7 +306,7 @@ class RequestsTestSuite(unittest.TestCase):
             rbody = json.loads(r.content)
             # 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'), None)
+            self.assertEquals(rbody.get('form'), {})
             self.assertEquals(rbody.get('data'), 'fooaowpeuf')
 
 
@@ -336,21 +325,6 @@ class RequestsTestSuite(unittest.TestCase):
             self.assertEquals(rbody.get('data'), '')
 
 
-    def test_nonurlencoded_post_querystring(self):
-
-        for service in SERVICES:
-
-            r = requests.post(service('post'), params='fooaowpeuf')
-
-            self.assertEquals(r.status_code, 200)
-            self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post?fooaowpeuf'))
-
-            rbody = json.loads(r.content)
-            self.assertEquals(rbody.get('form'), {}) # No form supplied
-            self.assertEquals(rbody.get('data'), '')
-
-
     def test_urlencoded_post_query_and_data(self):
 
         for service in SERVICES:
@@ -369,26 +343,24 @@ class RequestsTestSuite(unittest.TestCase):
             self.assertEquals(rbody.get('data'), '')
 
 
-    def test_nonurlencoded_post_query_and_data(self):
+    def test_nonurlencoded_postdata(self):
 
         for service in SERVICES:
 
-            r = requests.post(service('post'),
-                params='fooaowpeuf', data="foobar")
+            r = requests.post(service('post'), data="foobar")
 
             self.assertEquals(r.status_code, 200)
             self.assertEquals(r.headers['content-type'], 'application/json')
-            self.assertEquals(r.url, service('post?fooaowpeuf'))
 
             rbody = json.loads(r.content)
 
-            self.assertEquals(rbody.get('form'), None)
+            self.assertEquals(rbody.get('form'), {})
             self.assertEquals(rbody.get('data'), 'foobar')
 
 
-    def test_idna(self):
-        r = requests.get(u'http://➡.ws/httpbin')
-        assert 'httpbin' in r.url
+    # def test_idna(self):
+    #     r = requests.get(u'http://➡.ws/httpbin')
+    #     assert 'httpbin' in r.url
 
 
     def test_urlencoded_get_query_multivalued_param(self):
@@ -469,15 +441,8 @@ class RequestsTestSuite(unittest.TestCase):
 
     def test_session_HTTP_200_OK_GET(self):
 
-        s = Session()
-        r = s.get(httpbin('/'))
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_session_HTTPS_200_OK_GET(self):
-
-        s = Session()
-        r = s.get(httpsbin('/'))
+        s = requests.session()
+        r = s.get(httpbin('/get'))
         self.assertEqual(r.status_code, 200)
 
 
@@ -485,17 +450,109 @@ class RequestsTestSuite(unittest.TestCase):
 
         heads = {'User-agent': 'Mozilla/5.0'}
 
-        s = Session()
+        s = requests.session()
         s.headers = heads
+
         # Make 2 requests from Session object, should send header both times
         r1 = s.get(httpbin('user-agent'))
-
         assert heads['User-agent'] in r1.content
-        r2 = s.get(httpbin('user-agent'))
 
+        r2 = s.get(httpbin('user-agent'))
         assert heads['User-agent'] in r2.content
+
+        new_heads = {'User-agent': 'blah'}
+        r3 = s.get(httpbin('user-agent'), headers=new_heads)
+        assert new_heads['User-agent'] in r3.content
+
         self.assertEqual(r2.status_code, 200)
 
+    def test_session_persistent_cookies(self):
+
+        s = requests.session()
+
+        # Internally dispatched cookies are sent.
+        _c = {'kenneth': 'reitz', 'bessie': 'monke'}
+        r = s.get(httpbin('cookies'), cookies=_c)
+        r = s.get(httpbin('cookies'))
+
+        # Those cookies persist transparently.
+        c = json.loads(r.content).get('cookies')
+        assert c == _c
+
+        # Double check.
+        r = s.get(httpbin('cookies'), cookies={})
+        c = json.loads(r.content).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')
+        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')
+        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')
+
+        assert 'k' in c
+
+        # And server-set cookie persistience.
+        r = s.get(httpbin('cookies'))
+        c = json.loads(r.content).get('cookies')
+
+        assert 'k' in c
+
+
+
+    def test_session_persistent_params(self):
+
+        params = {'a': 'a_test'}
+
+        s = requests.session()
+        s.params = params
+
+        # Make 2 requests from Session object, should send header both times
+        r1 = s.get(httpbin('get'))
+        assert params['a'] in r1.content
+
+
+        params2 = {'b': 'b_test'}
+
+        r2 = s.get(httpbin('get'), params=params2)
+        assert params['a'] in r2.content
+        assert params2['b'] in r2.content
+
+
+        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
+
+    def test_invalid_content(self):
+        # WARNING: if you're using a terrible DNS provider (comcast),
+        # this will fail.
+        try:
+            hah = 'http://somedomainthatclearlydoesntexistg.com'
+            r = requests.get(hah, allow_redirects=False)
+        except requests.ConnectionError:
+            pass   # \o/
+        else:
+            assert False
+
+
+        config = {'safe_mode': True}
+        r = requests.get(hah, allow_redirects=False, config=config)
+        assert r.content == None
+
 
 if __name__ == '__main__':
     unittest.main()