diff options
Diffstat (limited to 'test_requests.py')
| -rwxr-xr-x | test_requests.py | 139 |
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() |