Imported Upstream version 0.10.8

1 files changed, 0 insertions, 721 deletions

diff --git a/test_requests.py b/test_requests.py
deleted file mode 100755
index 29de3ed..0000000
--- a/test_requests.py
+++ /dev/null
@@ -1,721 +0,0 @@
-#!/usr/bin/env python
-# -*- coding: utf-8 -*-
-
-
-
-import io
-import json
-import time
-import os
-import sys
-import unittest
-
-import requests
-from requests.compat import str, bytes, StringIO
-# import envoy
-from requests import HTTPError
-from requests.auth import HTTPBasicAuth, HTTPDigestAuth
-
-
-
-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 = os.environ.get('HTTPBIN_URL', 'http://0.0.0.0:%s/' % (PORT))
-
-
-def httpbin(*suffix):
-    """Returns url for HTTPBIN resource."""
-
-    return HTTPBIN_URL + '/'.join(suffix)
-
-
-SERVICES = (httpbin, )
-
-_httpbin = False
-
-class RequestsTestSuite(unittest.TestCase):
-    """Requests test cases."""
-
-    # It goes to eleven.
-    _multiprocess_can_split_ = True
-
-    def setUp(self):
-
-        global _httpbin
-
-        if (not 'HTTPBIN_URL' in os.environ) and not _httpbin:
-            # c = envoy.connect('httpbin %s' % (PORT))
-            # time.sleep(1)
-            _httpbin = True
-
-
-    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('/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)
-
-    def test_HTTP_302_GET(self):
-        r = requests.get(httpbin('redirect', '1'), allow_redirects=False)
-        self.assertEqual(r.status_code, 302)
-
-
-    def test_HTTP_200_OK_GET_WITH_PARAMS(self):
-        heads = {'User-agent': 'Mozilla/5.0'}
-
-        r = requests.get(httpbin('user-agent'), headers=heads)
-
-        assert heads['User-agent'] in r.text
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_HTTP_200_OK_GET_WITH_MIXED_PARAMS(self):
-        heads = {'User-agent': 'Mozilla/5.0'}
-
-        r = requests.get(httpbin('get') + '?test=true', params={'q': 'test'}, headers=heads)
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_user_agent_transfers(self):
-        """Issue XX"""
-
-        heads = {
-            'User-agent':
-                'Mozilla/5.0 (github.com/kennethreitz/requests)'
-        }
-
-        r = requests.get(httpbin('user-agent'), headers=heads);
-        self.assertTrue(heads['User-agent'] in r.text)
-
-        heads = {
-            'user-agent':
-                'Mozilla/5.0 (github.com/kennethreitz/requests)'
-        }
-
-        r = requests.get(httpbin('user-agent'), headers=heads);
-        self.assertTrue(heads['user-agent'] in r.text)
-
-
-    def test_HTTP_200_OK_HEAD(self):
-        r = requests.head(httpbin('/get'))
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_HTTP_200_OK_PUT(self):
-        r = requests.put(httpbin('put'))
-        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 = 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)
-
-
-            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 = HTTPDigestAuth('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:
-
-            url = service('post')
-            post = requests.post(url).raise_for_status()
-
-            post = requests.post(url, data={'some': 'data'})
-            self.assertEqual(post.status_code, 200)
-
-            post2 = requests.post(url, files={'some': open('test_requests.py')})
-            self.assertEqual(post2.status_code, 200)
-
-            post3 = requests.post(url, data='[{"some": "json"}]')
-            self.assertEqual(post3.status_code, 200)
-
-
-    def test_POSTBIN_GET_POST_FILES_WITH_PARAMS(self):
-
-        for service in SERVICES:
-
-            url = service('post')
-            post = requests.post(url,
-                files={'some': open('test_requests.py')},
-                data={'some': 'data'})
-
-            self.assertEqual(post.status_code, 200)
-
-
-    def test_POSTBIN_GET_POST_FILES_WITH_HEADERS(self):
-
-        for service in SERVICES:
-
-            url = service('post')
-
-            post2 = requests.post(url,
-                files={'some': open('test_requests.py')},
-                headers = {'User-Agent': 'requests-tests'})
-
-            self.assertEqual(post2.status_code, 200)
-
-
-    def test_nonzero_evaluation(self):
-
-        for service in SERVICES:
-
-            r = requests.get(service('status', '500'))
-            self.assertEqual(bool(r), False)
-
-            r = requests.get(service('/get'))
-            self.assertEqual(bool(r), True)
-
-
-    def test_request_ok_set(self):
-
-        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(HTTPError, r.raise_for_status)
-
-        r = requests.get(httpbin('status', '200'))
-        self.assertFalse(r.error)
-        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):
-
-        for service in SERVICES:
-
-            url = service('/get')
-
-            requests.get(url, params={'foo': 'føø'})
-            requests.get(url, params={'føø': 'føø'})
-            requests.get(url, params={'føø': 'føø'})
-            requests.get(url, params={'foo': 'foo'})
-            requests.get(service('ø'), params={'foo': 'foo'})
-
-
-    def test_httpauth_recursion(self):
-
-        http_auth = HTTPBasicAuth('user', 'BADpass')
-
-        for service in SERVICES:
-            r = requests.get(service('basic-auth', 'user', 'pass'), auth=http_auth)
-            self.assertEqual(r.status_code, 401)
-
-
-    def test_urlencoded_post_data(self):
-
-        for service in SERVICES:
-
-            r = requests.post(service('post'), data=dict(test='fooaowpeuf'))
-
-            self.assertEqual(r.status_code, 200)
-            self.assertEqual(r.headers['content-type'], 'application/json')
-            self.assertEqual(r.url, service('post'))
-
-            rbody = json.loads(r.text)
-
-            self.assertEqual(rbody.get('form'), dict(test='fooaowpeuf'))
-            self.assertEqual(rbody.get('data'), '')
-
-
-    def test_nonurlencoded_post_data(self):
-
-        for service in SERVICES:
-
-            r = requests.post(service('post'), data='fooaowpeuf')
-
-            self.assertEqual(r.status_code, 200)
-            self.assertEqual(r.headers['content-type'], 'application/json')
-            self.assertEqual(r.url, service('post'))
-
-            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".
-
-            assert rbody.get('form') in (None, {})
-            self.assertEqual(rbody.get('data'), 'fooaowpeuf')
-
-
-    def test_urlencoded_post_querystring(self):
-
-        for service in SERVICES:
-
-            r = requests.post(service('post'), params=dict(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.text)
-            self.assertEqual(rbody.get('form'), {}) # No form supplied
-            self.assertEqual(rbody.get('data'), '')
-
-
-    def test_urlencoded_post_query_and_data(self):
-
-        for service in SERVICES:
-
-            r = requests.post(
-                service('post'),
-                params=dict(test='fooaowpeuf'),
-                data=dict(test2="foobar"))
-
-            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.text)
-            self.assertEqual(rbody.get('form'), dict(test2='foobar'))
-            self.assertEqual(rbody.get('data'), '')
-
-
-    def test_nonurlencoded_postdata(self):
-
-        for service in SERVICES:
-
-            r = requests.post(service('post'), data="foobar")
-
-            self.assertEqual(r.status_code, 200)
-            self.assertEqual(r.headers['content-type'], 'application/json')
-
-            rbody = json.loads(r.text)
-
-            assert rbody.get('form') in (None, {})
-            self.assertEqual(rbody.get('data'), 'foobar')
-
-
-    # def test_idna(self):
-    #     r = requests.get(u'http://➡.ws/httpbin')
-    #     assert 'httpbin' in r.url
-
-
-    def test_urlencoded_get_query_multivalued_param(self):
-
-        for service in SERVICES:
-
-            r = requests.get(service('get'), params=dict(test=['foo','baz']))
-            self.assertEqual(r.status_code, 200)
-            self.assertEqual(r.url, service('get?test=foo&test=baz'))
-
-
-    def test_urlencoded_post_querystring_multivalued(self):
-
-        for service in SERVICES:
-
-            r = requests.post(service('post'), params=dict(test=['foo','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.text)
-            self.assertEqual(rbody.get('form'), {}) # No form supplied
-            self.assertEqual(rbody.get('data'), '')
-
-
-    def test_urlencoded_post_query_multivalued_and_data(self):
-
-        for service in SERVICES:
-
-            r = requests.post(
-                service('post'),
-                params=dict(test=['foo','baz']),
-                data=dict(test2="foobar",test3=['foo','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'))
-
-            # 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):
-
-        for service in SERVICES:
-
-            r = requests.get(service('redirect', '3'), allow_redirects=False)
-            self.assertEqual(r.status_code, 302)
-            self.assertEqual(len(r.history), 0)
-
-
-    def test_HEAD_no_redirect(self):
-
-        for service in SERVICES:
-
-            r = requests.head(service('redirect', '3'), allow_redirects=False)
-            self.assertEqual(r.status_code, 302)
-            self.assertEqual(len(r.history), 0)
-
-
-    def test_redirect_history(self):
-
-        for service in SERVICES:
-
-            r = requests.get(service('redirect', '3'))
-            self.assertEqual(r.status_code, 200)
-            self.assertEqual(len(r.history), 3)
-
-
-    def test_relative_redirect_history(self):
-
-        for service in SERVICES:
-
-            r = requests.get(service('relative-redirect', '3'))
-            self.assertEqual(r.status_code, 200)
-            self.assertEqual(len(r.history), 3)
-
-
-    def test_session_HTTP_200_OK_GET(self):
-
-        s = requests.session()
-        r = s.get(httpbin('/get'))
-        self.assertEqual(r.status_code, 200)
-
-
-    def test_session_persistent_headers(self):
-
-        heads = {'User-agent': 'Mozilla/5.0'}
-
-        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.text
-
-        r2 = s.get(httpbin('user-agent'))
-        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.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()
-
-        # 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.text).get('cookies')
-        assert c == _c
-
-        # Double check.
-        r = s.get(httpbin('cookies'), 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.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.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.text).get('cookies')
-
-        assert 'k' in c
-
-        # And server-set cookie persistience.
-        r = s.get(httpbin('cookies'))
-        c = json.loads(r.text).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.text
-
-
-        params2 = {'b': 'b_test'}
-
-        r2 = s.get(httpbin('get'), params=params2)
-        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.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),
-        # 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
-
-    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()