aboutsummaryrefslogtreecommitdiff
path: root/test_requests.py
diff options
context:
space:
mode:
Diffstat (limited to 'test_requests.py')
-rwxr-xr-xtest_requests.py139
1 files changed, 129 insertions, 10 deletions
diff --git a/test_requests.py b/test_requests.py
index 0d93893..cad8c05 100755
--- a/test_requests.py
+++ b/test_requests.py
@@ -20,7 +20,7 @@ from requests.compat import (
from requests.cookies import cookiejar_from_dict, morsel_to_cookie
from requests.exceptions import (ConnectionError, ConnectTimeout,
InvalidSchema, InvalidURL, MissingSchema,
- ReadTimeout, Timeout)
+ ReadTimeout, Timeout, RetryError)
from requests.models import PreparedRequest
from requests.structures import CaseInsensitiveDict
from requests.sessions import SessionRedirectMixin
@@ -103,6 +103,14 @@ class RequestsTestCase(unittest.TestCase):
head_req = requests.Request('HEAD', httpbin('head')).prepare()
assert 'Content-Length' not in head_req.headers
+ def test_override_content_length(self):
+ headers = {
+ 'Content-Length': 'not zero'
+ }
+ r = requests.Request('POST', httpbin('post'), headers=headers).prepare()
+ assert 'Content-Length' in r.headers
+ assert r.headers['Content-Length'] == 'not zero'
+
def test_path_is_not_double_encoded(self):
request = requests.Request('GET', "http://0.0.0.0/get/test case").prepare()
@@ -250,7 +258,7 @@ class RequestsTestCase(unittest.TestCase):
"""Do not send headers in Session.headers with None values."""
ses = requests.Session()
ses.headers['Accept-Encoding'] = None
- req = requests.Request('GET', 'http://httpbin.org/get')
+ req = requests.Request('GET', httpbin('get'))
prep = ses.prepare_request(req)
assert 'Accept-Encoding' not in prep.headers
@@ -293,13 +301,20 @@ class RequestsTestCase(unittest.TestCase):
r = s.get(url)
assert r.status_code == 200
- def test_connection_error(self):
+ def test_connection_error_invalid_domain(self):
"""Connecting to an unknown domain should raise a ConnectionError"""
with pytest.raises(ConnectionError):
- requests.get("http://fooobarbangbazbing.httpbin.org")
+ requests.get("http://doesnotexist.google.com")
+ def test_connection_error_invalid_port(self):
+ """Connecting to an invalid port should raise a ConnectionError"""
with pytest.raises(ConnectionError):
- requests.get("http://httpbin.org:1")
+ requests.get("http://httpbin.org:1", timeout=1)
+
+ def test_LocationParseError(self):
+ """Inputing a URL that cannot be parsed should raise an InvalidURL error"""
+ with pytest.raises(InvalidURL):
+ requests.get("http://fe80::5054:ff:fe5a:fc0")
def test_basicauth_with_netrc(self):
auth = ('user', 'pass')
@@ -920,6 +935,27 @@ class RequestsTestCase(unittest.TestCase):
assert 'multipart/form-data' in p.headers['Content-Type']
+ def test_can_send_bytes_bytearray_objects_with_files(self):
+ # Test bytes:
+ data = {'a': 'this is a string'}
+ files = {'b': b'foo'}
+ r = requests.Request('POST', httpbin('post'), data=data, files=files)
+ p = r.prepare()
+ assert 'multipart/form-data' in p.headers['Content-Type']
+ # Test bytearrays:
+ files = {'b': bytearray(b'foo')}
+ r = requests.Request('POST', httpbin('post'), data=data, files=files)
+ p = r.prepare()
+ assert 'multipart/form-data' in p.headers['Content-Type']
+
+ def test_can_send_file_object_with_non_string_filename(self):
+ f = io.BytesIO()
+ f.name = 2
+ r = requests.Request('POST', httpbin('post'), files={'f': f})
+ p = r.prepare()
+
+ assert 'multipart/form-data' in p.headers['Content-Type']
+
def test_autoset_header_values_are_native(self):
data = 'this is a string'
length = '16'
@@ -1000,12 +1036,12 @@ class RequestsTestCase(unittest.TestCase):
assert s == "Basic dGVzdDp0ZXN0"
def test_requests_history_is_saved(self):
- r = requests.get('https://httpbin.org/redirect/5')
+ r = requests.get(httpbin('redirect/5'))
total = r.history[-1].history
i = 0
for item in r.history:
assert item.history == total[0:i]
- i=i+1
+ i = i + 1
def test_json_param_post_content_type_works(self):
r = requests.post(
@@ -1016,6 +1052,23 @@ class RequestsTestCase(unittest.TestCase):
assert 'application/json' in r.request.headers['Content-Type']
assert {'life': 42} == r.json()['json']
+ def test_response_iter_lines(self):
+ r = requests.get(httpbin('stream/4'), stream=True)
+ assert r.status_code == 200
+
+ it = r.iter_lines()
+ next(it)
+ assert len(list(it)) == 3
+
+ @pytest.mark.xfail
+ def test_response_iter_lines_reentrant(self):
+ """Response.iter_lines() is not reentrant safe"""
+ r = requests.get(httpbin('stream/4'), stream=True)
+ assert r.status_code == 200
+
+ next(r.iter_lines())
+ assert len(list(r.iter_lines())) == 3
+
class TestContentEncodingDetection(unittest.TestCase):
@@ -1244,6 +1297,32 @@ class UtilsTestCase(unittest.TestCase):
'http://localhost.localdomain:5000/v1.0/') == {}
assert get_environ_proxies('http://www.requests.com/') != {}
+ def test_guess_filename_when_int(self):
+ from requests.utils import guess_filename
+ assert None is guess_filename(1)
+
+ def test_guess_filename_when_filename_is_an_int(self):
+ from requests.utils import guess_filename
+ fake = type('Fake', (object,), {'name': 1})()
+ assert None is guess_filename(fake)
+
+ def test_guess_filename_with_file_like_obj(self):
+ from requests.utils import guess_filename
+ from requests import compat
+ fake = type('Fake', (object,), {'name': b'value'})()
+ guessed_name = guess_filename(fake)
+ assert b'value' == guessed_name
+ assert isinstance(guessed_name, compat.bytes)
+
+ def test_guess_filename_with_unicode_name(self):
+ from requests.utils import guess_filename
+ from requests import compat
+ filename = b'value'.decode('utf-8')
+ fake = type('Fake', (object,), {'name': filename})()
+ guessed_name = guess_filename(fake)
+ assert filename == guessed_name
+ assert isinstance(guessed_name, compat.str)
+
def test_is_ipv4_address(self):
from requests.utils import is_ipv4_address
assert is_ipv4_address('8.8.8.8')
@@ -1280,6 +1359,22 @@ class UtilsTestCase(unittest.TestCase):
assert username == percent_encoding_test_chars
assert password == percent_encoding_test_chars
+ def test_requote_uri_with_unquoted_percents(self):
+ """Ensure we handle unquoted percent signs in redirects.
+
+ See: https://github.com/kennethreitz/requests/issues/2356
+ """
+ from requests.utils import requote_uri
+ bad_uri = 'http://example.com/fiz?buz=%ppicture'
+ quoted = 'http://example.com/fiz?buz=%25ppicture'
+ assert quoted == requote_uri(bad_uri)
+
+ def test_requote_uri_properly_requotes(self):
+ """Ensure requoting doesn't break expectations."""
+ from requests.utils import requote_uri
+ quoted = 'http://example.com/fiz?buz=%25ppicture'
+ assert quoted == requote_uri(quoted)
+
class TestMorselToCookieExpires(unittest.TestCase):
@@ -1342,7 +1437,7 @@ class TestMorselToCookieMaxAge(unittest.TestCase):
class TestTimeout:
def test_stream_timeout(self):
try:
- requests.get('https://httpbin.org/delay/10', timeout=2.0)
+ requests.get(httpbin('delay/10'), timeout=2.0)
except requests.exceptions.Timeout as e:
assert 'Read timed out' in e.args[0].args[0]
@@ -1389,6 +1484,11 @@ class TestTimeout:
except ConnectTimeout:
pass
+ def test_encoded_methods(self):
+ """See: https://github.com/kennethreitz/requests/issues/2316"""
+ r = requests.request(b'GET', httpbin('get'))
+ assert r.ok
+
SendCall = collections.namedtuple('SendCall', ('args', 'kwargs'))
@@ -1437,7 +1537,7 @@ class TestRedirects:
def test_requests_are_updated_each_time(self):
session = RedirectSession([303, 307])
- prep = requests.Request('POST', 'http://httpbin.org/post').prepare()
+ prep = requests.Request('POST', httpbin('post')).prepare()
r0 = session.send(prep)
assert r0.request.method == 'POST'
assert session.calls[-1] == SendCall((r0.request,), {})
@@ -1507,13 +1607,32 @@ def test_prepared_request_complete_copy():
)
assert_copy(p, p.copy())
+
def test_prepare_unicode_url():
p = PreparedRequest()
p.prepare(
method='GET',
- url=u('http://www.example.com/üniçø∂é')
+ url=u('http://www.example.com/üniçø∂é'),
)
assert_copy(p, p.copy())
+
+def test_urllib3_retries():
+ from requests.packages.urllib3.util import Retry
+ s = requests.Session()
+ s.mount('http://', HTTPAdapter(max_retries=Retry(
+ total=2, status_forcelist=[500]
+ )))
+
+ with pytest.raises(RetryError):
+ s.get(httpbin('status/500'))
+
+def test_vendor_aliases():
+ from requests.packages import urllib3
+ from requests.packages import chardet
+
+ with pytest.raises(ImportError):
+ from requests.packages import webbrowser
+
if __name__ == '__main__':
unittest.main()