# -*- coding: utf-8 -*- """Tests for Requests.""" from __future__ import division import json import os import pickle import collections import contextlib import warnings import re import io import requests import pytest from requests.adapters import HTTPAdapter from requests.auth import HTTPDigestAuth, _basic_auth_str from requests.compat import ( Morsel, cookielib, getproxies, str, urlparse, builtin_str) from requests.cookies import ( cookiejar_from_dict, morsel_to_cookie) from requests.exceptions import ( ConnectionError, ConnectTimeout, InvalidSchema, InvalidURL, MissingSchema, ReadTimeout, Timeout, RetryError, TooManyRedirects, ProxyError, InvalidHeader, UnrewindableBodyError, SSLError, InvalidProxyURL) from requests.models import PreparedRequest from requests.structures import CaseInsensitiveDict from requests.sessions import SessionRedirectMixin from requests.models import urlencode from requests.hooks import default_hooks from requests.compat import MutableMapping from .compat import StringIO, u from .utils import override_environ from urllib3.util import Timeout as Urllib3Timeout # Requests to this URL should always fail with a connection timeout (nothing # listening on that port) TARPIT = 'http://10.255.255.1' try: from ssl import SSLContext del SSLContext HAS_MODERN_SSL = True except ImportError: HAS_MODERN_SSL = False try: requests.pyopenssl HAS_PYOPENSSL = True except AttributeError: HAS_PYOPENSSL = False class TestRequests: digest_auth_algo = ('MD5', 'SHA-256', 'SHA-512') def test_entry_points(self): requests.session requests.session().get requests.session().head requests.get requests.head requests.put requests.patch requests.post # Not really an entry point, but people rely on it. from requests.packages.urllib3.poolmanager import PoolManager @pytest.mark.parametrize( 'exception, url', ( (MissingSchema, 'hiwpefhipowhefopw'), (InvalidSchema, 'localhost:3128'), (InvalidSchema, 'localhost.localdomain:3128/'), (InvalidSchema, '10.122.1.1:3128/'), (InvalidURL, 'http://'), )) def test_invalid_url(self, exception, url): with pytest.raises(exception): requests.get(url) def test_basic_building(self): req = requests.Request() req.url = 'http://kennethreitz.org/' req.data = {'life': '42'} pr = req.prepare() assert pr.url == req.url assert pr.body == 'life=42' @pytest.mark.parametrize('method', ('GET', 'HEAD')) def test_no_content_length(self, httpbin, method): req = requests.Request(method, httpbin(method.lower())).prepare() assert 'Content-Length' not in req.headers @pytest.mark.parametrize('method', ('POST', 'PUT', 'PATCH', 'OPTIONS')) def test_no_body_content_length(self, httpbin, method): req = requests.Request(method, httpbin(method.lower())).prepare() assert req.headers['Content-Length'] == '0' @pytest.mark.parametrize('method', ('POST', 'PUT', 'PATCH', 'OPTIONS')) def test_empty_content_length(self, httpbin, method): req = requests.Request(method, httpbin(method.lower()), data='').prepare() assert req.headers['Content-Length'] == '0' def test_override_content_length(self, httpbin): 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() assert request.path_url == '/get/test%20case' @pytest.mark.parametrize( 'url, expected', ( ('http://example.com/path#fragment', 'http://example.com/path?a=b#fragment'), ('http://example.com/path?key=value#fragment', 'http://example.com/path?key=value&a=b#fragment') )) def test_params_are_added_before_fragment(self, url, expected): request = requests.Request('GET', url, params={"a": "b"}).prepare() assert request.url == expected def test_params_original_order_is_preserved_by_default(self): param_ordered_dict = collections.OrderedDict((('z', 1), ('a', 1), ('k', 1), ('d', 1))) session = requests.Session() request = requests.Request('GET', 'http://example.com/', params=param_ordered_dict) prep = session.prepare_request(request) assert prep.url == 'http://example.com/?z=1&a=1&k=1&d=1' def test_params_bytes_are_encoded(self): request = requests.Request('GET', 'http://example.com', params=b'test=foo').prepare() assert request.url == 'http://example.com/?test=foo' def test_binary_put(self): request = requests.Request('PUT', 'http://example.com', data=u"ööö".encode("utf-8")).prepare() assert isinstance(request.body, bytes) def test_whitespaces_are_removed_from_url(self): # Test for issue #3696 request = requests.Request('GET', ' http://example.com').prepare() assert request.url == 'http://example.com/' @pytest.mark.parametrize('scheme', ('http://', 'HTTP://', 'hTTp://', 'HttP://')) def test_mixed_case_scheme_acceptable(self, httpbin, scheme): s = requests.Session() s.proxies = getproxies() parts = urlparse(httpbin('get')) url = scheme + parts.netloc + parts.path r = requests.Request('GET', url) r = s.send(r.prepare()) assert r.status_code == 200, 'failed for scheme {}'.format(scheme) def test_HTTP_200_OK_GET_ALTERNATIVE(self, httpbin): r = requests.Request('GET', httpbin('get')) s = requests.Session() s.proxies = getproxies() r = s.send(r.prepare()) assert r.status_code == 200 def test_HTTP_302_ALLOW_REDIRECT_GET(self, httpbin): r = requests.get(httpbin('redirect', '1')) assert r.status_code == 200 assert r.history[0].status_code == 302 assert r.history[0].is_redirect def test_HTTP_307_ALLOW_REDIRECT_POST(self, httpbin): r = requests.post(httpbin('redirect-to'), data='test', params={'url': 'post', 'status_code': 307}) assert r.status_code == 200 assert r.history[0].status_code == 307 assert r.history[0].is_redirect assert r.json()['data'] == 'test' def test_HTTP_307_ALLOW_REDIRECT_POST_WITH_SEEKABLE(self, httpbin): byte_str = b'test' r = requests.post(httpbin('redirect-to'), data=io.BytesIO(byte_str), params={'url': 'post', 'status_code': 307}) assert r.status_code == 200 assert r.history[0].status_code == 307 assert r.history[0].is_redirect assert r.json()['data'] == byte_str.decode('utf-8') def test_HTTP_302_TOO_MANY_REDIRECTS(self, httpbin): try: requests.get(httpbin('relative-redirect', '50')) except TooManyRedirects as e: url = httpbin('relative-redirect', '20') assert e.request.url == url assert e.response.url == url assert len(e.response.history) == 30 else: pytest.fail('Expected redirect to raise TooManyRedirects but it did not') def test_HTTP_302_TOO_MANY_REDIRECTS_WITH_PARAMS(self, httpbin): s = requests.session() s.max_redirects = 5 try: s.get(httpbin('relative-redirect', '50')) except TooManyRedirects as e: url = httpbin('relative-redirect', '45') assert e.request.url == url assert e.response.url == url assert len(e.response.history) == 5 else: pytest.fail('Expected custom max number of redirects to be respected but was not') def test_http_301_changes_post_to_get(self, httpbin): r = requests.post(httpbin('status', '301')) assert r.status_code == 200 assert r.request.method == 'GET' assert r.history[0].status_code == 301 assert r.history[0].is_redirect def test_http_301_doesnt_change_head_to_get(self, httpbin): r = requests.head(httpbin('status', '301'), allow_redirects=True) print(r.content) assert r.status_code == 200 assert r.request.method == 'HEAD' assert r.history[0].status_code == 301 assert r.history[0].is_redirect def test_http_302_changes_post_to_get(self, httpbin): r = requests.post(httpbin('status', '302')) assert r.status_code == 200 assert r.request.method == 'GET' assert r.history[0].status_code == 302 assert r.history[0].is_redirect def test_http_302_doesnt_change_head_to_get(self, httpbin): r = requests.head(httpbin('status', '302'), allow_redirects=True) assert r.status_code == 200 assert r.request.method == 'HEAD' assert r.history[0].status_code == 302 assert r.history[0].is_redirect def test_http_303_changes_post_to_get(self, httpbin): r = requests.post(httpbin('status', '303')) assert r.status_code == 200 assert r.request.method == 'GET' assert r.history[0].status_code == 303 assert r.history[0].is_redirect def test_http_303_doesnt_change_head_to_get(self, httpbin): r = requests.head(httpbin('status', '303'), allow_redirects=True) assert r.status_code == 200 assert r.request.method == 'HEAD' assert r.history[0].status_code == 303 assert r.history[0].is_redirect def test_header_and_body_removal_on_redirect(self, httpbin): purged_headers = ('Content-Length', 'Content-Type') ses = requests.Session() req = requests.Request('POST', httpbin('post'), data={'test': 'data'}) prep = ses.prepare_request(req) resp = ses.send(prep) # Mimic a redirect response resp.status_code = 302 resp.headers['location'] = 'get' # Run request through resolve_redirects next_resp = next(ses.resolve_redirects(resp, prep)) assert next_resp.request.body is None for header in purged_headers: assert header not in next_resp.request.headers def test_transfer_enc_removal_on_redirect(self, httpbin): purged_headers = ('Transfer-Encoding', 'Content-Type') ses = requests.Session() req = requests.Request('POST', httpbin('post'), data=(b'x' for x in range(1))) prep = ses.prepare_request(req) assert 'Transfer-Encoding' in prep.headers # Create Response to avoid https://github.com/kevin1024/pytest-httpbin/issues/33 resp = requests.Response() resp.raw = io.BytesIO(b'the content') resp.request = prep setattr(resp.raw, 'release_conn', lambda *args: args) # Mimic a redirect response resp.status_code = 302 resp.headers['location'] = httpbin('get') # Run request through resolve_redirect next_resp = next(ses.resolve_redirects(resp, prep)) assert next_resp.request.body is None for header in purged_headers: assert header not in next_resp.request.headers def test_fragment_maintained_on_redirect(self, httpbin): fragment = "#view=edit&token=hunter2" r = requests.get(httpbin('redirect-to?url=get')+fragment) assert len(r.history) > 0 assert r.history[0].request.url == httpbin('redirect-to?url=get')+fragment assert r.url == httpbin('get')+fragment def test_HTTP_200_OK_GET_WITH_PARAMS(self, httpbin): heads = {'User-agent': 'Mozilla/5.0'} r = requests.get(httpbin('user-agent'), headers=heads) assert heads['User-agent'] in r.text assert r.status_code == 200 def test_HTTP_200_OK_GET_WITH_MIXED_PARAMS(self, httpbin): heads = {'User-agent': 'Mozilla/5.0'} r = requests.get(httpbin('get') + '?test=true', params={'q': 'test'}, headers=heads) assert r.status_code == 200 def test_set_cookie_on_301(self, httpbin): s = requests.session() url = httpbin('cookies/set?foo=bar') s.get(url) assert s.cookies['foo'] == 'bar' def test_cookie_sent_on_redirect(self, httpbin): s = requests.session() s.get(httpbin('cookies/set?foo=bar')) r = s.get(httpbin('redirect/1')) # redirects to httpbin('get') assert 'Cookie' in r.json()['headers'] def test_cookie_removed_on_expire(self, httpbin): s = requests.session() s.get(httpbin('cookies/set?foo=bar')) assert s.cookies['foo'] == 'bar' s.get( httpbin('response-headers'), params={ 'Set-Cookie': 'foo=deleted; expires=Thu, 01-Jan-1970 00:00:01 GMT' } ) assert 'foo' not in s.cookies def test_cookie_quote_wrapped(self, httpbin): s = requests.session() s.get(httpbin('cookies/set?foo="bar:baz"')) assert s.cookies['foo'] == '"bar:baz"' def test_cookie_persists_via_api(self, httpbin): s = requests.session() r = s.get(httpbin('redirect/1'), cookies={'foo': 'bar'}) assert 'foo' in r.request.headers['Cookie'] assert 'foo' in r.history[0].request.headers['Cookie'] def test_request_cookie_overrides_session_cookie(self, httpbin): s = requests.session() s.cookies['foo'] = 'bar' r = s.get(httpbin('cookies'), cookies={'foo': 'baz'}) assert r.json()['cookies']['foo'] == 'baz' # Session cookie should not be modified assert s.cookies['foo'] == 'bar' def test_request_cookies_not_persisted(self, httpbin): s = requests.session() s.get(httpbin('cookies'), cookies={'foo': 'baz'}) # Sending a request with cookies should not add cookies to the session assert not s.cookies def test_generic_cookiejar_works(self, httpbin): cj = cookielib.CookieJar() cookiejar_from_dict({'foo': 'bar'}, cj) s = requests.session() s.cookies = cj r = s.get(httpbin('cookies')) # Make sure the cookie was sent assert r.json()['cookies']['foo'] == 'bar' # Make sure the session cj is still the custom one assert s.cookies is cj def test_param_cookiejar_works(self, httpbin): cj = cookielib.CookieJar() cookiejar_from_dict({'foo': 'bar'}, cj) s = requests.session() r = s.get(httpbin('cookies'), cookies=cj) # Make sure the cookie was sent assert r.json()['cookies']['foo'] == 'bar' def test_cookielib_cookiejar_on_redirect(self, httpbin): """Tests resolve_redirect doesn't fail when merging cookies with non-RequestsCookieJar cookiejar. See GH #3579 """ cj = cookiejar_from_dict({'foo': 'bar'}, cookielib.CookieJar()) s = requests.Session() s.cookies = cookiejar_from_dict({'cookie': 'tasty'}) # Prepare request without using Session req = requests.Request('GET', httpbin('headers'), cookies=cj) prep_req = req.prepare() # Send request and simulate redirect resp = s.send(prep_req) resp.status_code = 302 resp.headers['location'] = httpbin('get') redirects = s.resolve_redirects(resp, prep_req) resp = next(redirects) # Verify CookieJar isn't being converted to RequestsCookieJar assert isinstance(prep_req._cookies, cookielib.CookieJar) assert isinstance(resp.request._cookies, cookielib.CookieJar) assert not isinstance(resp.request._cookies, requests.cookies.RequestsCookieJar) cookies = {} for c in resp.request._cookies: cookies[c.name] = c.value assert cookies['foo'] == 'bar' assert cookies['cookie'] == 'tasty' def test_requests_in_history_are_not_overridden(self, httpbin): resp = requests.get(httpbin('redirect/3')) urls = [r.url for r in resp.history] req_urls = [r.request.url for r in resp.history] assert urls == req_urls def test_history_is_always_a_list(self, httpbin): """Show that even with redirects, Response.history is always a list.""" resp = requests.get(httpbin('get')) assert isinstance(resp.history, list) resp = requests.get(httpbin('redirect/1')) assert isinstance(resp.history, list) assert not isinstance(resp.history, tuple) def test_headers_on_session_with_None_are_not_sent(self, httpbin): """Do not send headers in Session.headers with None values.""" ses = requests.Session() ses.headers['Accept-Encoding'] = None req = requests.Request('GET', httpbin('get')) prep = ses.prepare_request(req) assert 'Accept-Encoding' not in prep.headers def test_headers_preserve_order(self, httpbin): """Preserve order when headers provided as OrderedDict.""" ses = requests.Session() ses.headers = collections.OrderedDict() ses.headers['Accept-Encoding'] = 'identity' ses.headers['First'] = '1' ses.headers['Second'] = '2' headers = collections.OrderedDict([('Third', '3'), ('Fourth', '4')]) headers['Fifth'] = '5' headers['Second'] = '222' req = requests.Request('GET', httpbin('get'), headers=headers) prep = ses.prepare_request(req) items = list(prep.headers.items()) assert items[0] == ('Accept-Encoding', 'identity') assert items[1] == ('First', '1') assert items[2] == ('Second', '222') assert items[3] == ('Third', '3') assert items[4] == ('Fourth', '4') assert items[5] == ('Fifth', '5') @pytest.mark.parametrize('key', ('User-agent', 'user-agent')) def test_user_agent_transfers(self, httpbin, key): heads = {key: 'Mozilla/5.0 (github.com/psf/requests)'} r = requests.get(httpbin('user-agent'), headers=heads) assert heads[key] in r.text def test_HTTP_200_OK_HEAD(self, httpbin): r = requests.head(httpbin('get')) assert r.status_code == 200 def test_HTTP_200_OK_PUT(self, httpbin): r = requests.put(httpbin('put')) assert r.status_code == 200 def test_BASICAUTH_TUPLE_HTTP_200_OK_GET(self, httpbin): auth = ('user', 'pass') url = httpbin('basic-auth', 'user', 'pass') r = requests.get(url, auth=auth) assert r.status_code == 200 r = requests.get(url) assert r.status_code == 401 s = requests.session() s.auth = auth r = s.get(url) assert r.status_code == 200 @pytest.mark.parametrize( 'username, password', ( ('user', 'pass'), (u'имя'.encode('utf-8'), u'пароль'.encode('utf-8')), (42, 42), (None, None), )) def test_set_basicauth(self, httpbin, username, password): auth = (username, password) url = httpbin('get') r = requests.Request('GET', url, auth=auth) p = r.prepare() assert p.headers['Authorization'] == _basic_auth_str(username, password) def test_basicauth_encodes_byte_strings(self): """Ensure b'test' formats as the byte string "test" rather than the unicode string "b'test'" in Python 3. """ auth = (b'\xc5\xafsername', b'test\xc6\xb6') r = requests.Request('GET', 'http://localhost', auth=auth) p = r.prepare() assert p.headers['Authorization'] == 'Basic xa9zZXJuYW1lOnRlc3TGtg==' @pytest.mark.parametrize( 'url, exception', ( # Connecting to an unknown domain should raise a ConnectionError ('http://doesnotexist.google.com', ConnectionError), # Connecting to an invalid port should raise a ConnectionError ('http://localhost:1', ConnectionError), # Inputing a URL that cannot be parsed should raise an InvalidURL error ('http://fe80::5054:ff:fe5a:fc0', InvalidURL) )) def test_errors(self, url, exception): with pytest.raises(exception): requests.get(url, timeout=1) def test_proxy_error(self): # any proxy related error (address resolution, no route to host, etc) should result in a ProxyError with pytest.raises(ProxyError): requests.get('http://localhost:1', proxies={'http': 'non-resolvable-address'}) def test_proxy_error_on_bad_url(self, httpbin, httpbin_secure): with pytest.raises(InvalidProxyURL): requests.get(httpbin_secure(), proxies={'https': 'http:/badproxyurl:3128'}) with pytest.raises(InvalidProxyURL): requests.get(httpbin(), proxies={'http': 'http://:8080'}) with pytest.raises(InvalidProxyURL): requests.get(httpbin_secure(), proxies={'https': 'https://'}) with pytest.raises(InvalidProxyURL): requests.get(httpbin(), proxies={'http': 'http:///example.com:8080'}) def test_basicauth_with_netrc(self, httpbin): auth = ('user', 'pass') wrong_auth = ('wronguser', 'wrongpass') url = httpbin('basic-auth', 'user', 'pass') old_auth = requests.sessions.get_netrc_auth try: def get_netrc_auth_mock(url): return auth requests.sessions.get_netrc_auth = get_netrc_auth_mock # Should use netrc and work. r = requests.get(url) assert r.status_code == 200 # Given auth should override and fail. r = requests.get(url, auth=wrong_auth) assert r.status_code == 401 s = requests.session() # Should use netrc and work. r = s.get(url) assert r.status_code == 200 # Given auth should override and fail. s.auth = wrong_auth r = s.get(url) assert r.status_code == 401 finally: requests.sessions.get_netrc_auth = old_auth def test_DIGEST_HTTP_200_OK_GET(self, httpbin): for authtype in self.digest_auth_algo: auth = HTTPDigestAuth('user', 'pass') url = httpbin('digest-auth', 'auth', 'user', 'pass', authtype, 'never') r = requests.get(url, auth=auth) assert r.status_code == 200 r = requests.get(url) assert r.status_code == 401 print(r.headers['WWW-Authenticate']) s = requests.session() s.auth = HTTPDigestAuth('user', 'pass') r = s.get(url) assert r.status_code == 200 def test_DIGEST_AUTH_RETURNS_COOKIE(self, httpbin): for authtype in self.digest_auth_algo: url = httpbin('digest-auth', 'auth', 'user', 'pass', authtype) auth = HTTPDigestAuth('user', 'pass') r = requests.get(url) assert r.cookies['fake'] == 'fake_value' r = requests.get(url, auth=auth) assert r.status_code == 200 def test_DIGEST_AUTH_SETS_SESSION_COOKIES(self, httpbin): for authtype in self.digest_auth_algo: url = httpbin('digest-auth', 'auth', 'user', 'pass', authtype) auth = HTTPDigestAuth('user', 'pass') s = requests.Session() s.get(url, auth=auth) assert s.cookies['fake'] == 'fake_value' def test_DIGEST_STREAM(self, httpbin): for authtype in self.digest_auth_algo: auth = HTTPDigestAuth('user', 'pass') url = httpbin('digest-auth', 'auth', 'user', 'pass', authtype) r = requests.get(url, auth=auth, stream=True) assert r.raw.read() != b'' r = requests.get(url, auth=auth, stream=False) assert r.raw.read() == b'' def test_DIGESTAUTH_WRONG_HTTP_401_GET(self, httpbin): for authtype in self.digest_auth_algo: auth = HTTPDigestAuth('user', 'wrongpass') url = httpbin('digest-auth', 'auth', 'user', 'pass', authtype) r = requests.get(url, auth=auth) assert r.status_code == 401 r = requests.get(url) assert r.status_code == 401 s = requests.session() s.auth = auth r = s.get(url) assert r.status_code == 401 def test_DIGESTAUTH_QUOTES_QOP_VALUE(self, httpbin): for authtype in self.digest_auth_algo: auth = HTTPDigestAuth('user', 'pass') url = httpbin('digest-auth', 'auth', 'user', 'pass', authtype) r = requests.get(url, auth=auth) assert '"auth"' in r.request.headers['Authorization'] def test_POSTBIN_GET_POST_FILES(self, httpbin): url = httpbin('post') requests.post(url).raise_for_status() post1 = requests.post(url, data={'some': 'data'}) assert post1.status_code == 200 with open('requirements-dev.txt') as f: post2 = requests.post(url, files={'some': f}) assert post2.status_code == 200 post4 = requests.post(url, data='[{"some": "json"}]') assert post4.status_code == 200 with pytest.raises(ValueError): requests.post(url, files=['bad file data']) def test_invalid_files_input(self, httpbin): url = httpbin('post') post = requests.post(url, files={"random-file-1": None, "random-file-2": 1}) assert b'name="random-file-1"' not in post.request.body assert b'name="random-file-2"' in post.request.body def test_POSTBIN_SEEKED_OBJECT_WITH_NO_ITER(self, httpbin): class TestStream(object): def __init__(self, data): self.data = data.encode() self.length = len(self.data) self.index = 0 def __len__(self): return self.length def read(self, size=None): if size: ret = self.data[self.index:self.index + size] self.index += size else: ret = self.data[self.index:] self.index = self.length return ret def tell(self): return self.index def seek(self, offset, where=0): if where == 0: self.index = offset elif where == 1: self.index += offset elif where == 2: self.index = self.length + offset test = TestStream('test') post1 = requests.post(httpbin('post'), data=test) assert post1.status_code == 200 assert post1.json()['data'] == 'test' test = TestStream('test') test.seek(2) post2 = requests.post(httpbin('post'), data=test) assert post2.status_code == 200 assert post2.json()['data'] == 'st' def test_POSTBIN_GET_POST_FILES_WITH_DATA(self, httpbin): url = httpbin('post') requests.post(url).raise_for_status() post1 = requests.post(url, data={'some': 'data'}) assert post1.status_code == 200 with open('requirements-dev.txt') as f: post2 = requests.post(url, data={'some': 'data'}, files={'some': f}) assert post2.status_code == 200 post4 = requests.post(url, data='[{"some": "json"}]') assert post4.status_code == 200 with pytest.raises(ValueError): requests.post(url, files=['bad file data']) def test_post_with_custom_mapping(self, httpbin): class CustomMapping(MutableMapping): def __init__(self, *args, **kwargs): self.data = dict(*args, **kwargs) def __delitem__(self, key): del self.data[key] def __getitem__(self, key): return self.data[key] def __setitem__(self, key, value): self.data[key] = value def __iter__(self): return iter(self.data) def __len__(self): return len(self.data) data = CustomMapping({'some': 'data'}) url = httpbin('post') found_json = requests.post(url, data=data).json().get('form') assert found_json == {'some': 'data'} def test_conflicting_post_params(self, httpbin): url = httpbin('post') with open('requirements-dev.txt') as f: with pytest.raises(ValueError): requests.post(url, data='[{"some": "data"}]', files={'some': f}) with pytest.raises(ValueError): requests.post(url, data=u('[{"some": "data"}]'), files={'some': f}) def test_request_ok_set(self, httpbin): r = requests.get(httpbin('status', '404')) assert not r.ok def test_status_raising(self, httpbin): r = requests.get(httpbin('status', '404')) with pytest.raises(requests.exceptions.HTTPError): r.raise_for_status() r = requests.get(httpbin('status', '500')) assert not r.ok def test_decompress_gzip(self, httpbin): r = requests.get(httpbin('gzip')) r.content.decode('ascii') @pytest.mark.parametrize( 'url, params', ( ('/get', {'foo': 'føø'}), ('/get', {'føø': 'føø'}), ('/get', {'føø': 'føø'}), ('/get', {'foo': 'foo'}), ('ø', {'foo': 'foo'}), )) def test_unicode_get(self, httpbin, url, params): requests.get(httpbin(url), params=params) def test_unicode_header_name(self, httpbin): requests.put( httpbin('put'), headers={str('Content-Type'): 'application/octet-stream'}, data='\xff') # compat.str is unicode. def test_pyopenssl_redirect(self, httpbin_secure, httpbin_ca_bundle): requests.get(httpbin_secure('status', '301'), verify=httpbin_ca_bundle) def test_invalid_ca_certificate_path(self, httpbin_secure): INVALID_PATH = '/garbage' with pytest.raises(IOError) as e: requests.get(httpbin_secure(), verify=INVALID_PATH) assert str(e.value) == 'Could not find a suitable TLS CA certificate bundle, invalid path: {}'.format(INVALID_PATH) def test_invalid_ssl_certificate_files(self, httpbin_secure): INVALID_PATH = '/garbage' with pytest.raises(IOError) as e: requests.get(httpbin_secure(), cert=INVALID_PATH) assert str(e.value) == 'Could not find the TLS certificate file, invalid path: {}'.format(INVALID_PATH) with pytest.raises(IOError) as e: requests.get(httpbin_secure(), cert=('.', INVALID_PATH)) assert str(e.value) == 'Could not find the TLS key file, invalid path: {}'.format(INVALID_PATH) def test_http_with_certificate(self, httpbin): r = requests.get(httpbin(), cert='.') assert r.status_code == 200 def test_https_warnings(self, httpbin_secure, httpbin_ca_bundle): """warnings are emitted with requests.get""" if HAS_MODERN_SSL or HAS_PYOPENSSL: warnings_expected = ('SubjectAltNameWarning', ) else: warnings_expected = ('SNIMissingWarning', 'InsecurePlatformWarning', 'SubjectAltNameWarning', ) with pytest.warns(None) as warning_records: warnings.simplefilter('always') requests.get(httpbin_secure('status', '200'), verify=httpbin_ca_bundle) warning_records = [item for item in warning_records if item.category.__name__ != 'ResourceWarning'] warnings_category = tuple( item.category.__name__ for item in warning_records) assert warnings_category == warnings_expected def test_certificate_failure(self, httpbin_secure): """ When underlying SSL problems occur, an SSLError is raised. """ with pytest.raises(SSLError): # Our local httpbin does not have a trusted CA, so this call will # fail if we use our default trust bundle. requests.get(httpbin_secure('status', '200')) def test_urlencoded_get_query_multivalued_param(self, httpbin): r = requests.get(httpbin('get'), params={'test': ['foo', 'baz']}) assert r.status_code == 200 assert r.url == httpbin('get?test=foo&test=baz') def test_form_encoded_post_query_multivalued_element(self, httpbin): r = requests.Request(method='POST', url=httpbin('post'), data=dict(test=['foo', 'baz'])) prep = r.prepare() assert prep.body == 'test=foo&test=baz' def test_different_encodings_dont_break_post(self, httpbin): r = requests.post(httpbin('post'), data={'stuff': json.dumps({'a': 123})}, params={'blah': 'asdf1234'}, files={'file': ('test_requests.py', open(__file__, 'rb'))}) assert r.status_code == 200 @pytest.mark.parametrize( 'data', ( {'stuff': u('ëlïxr')}, {'stuff': u('ëlïxr').encode('utf-8')}, {'stuff': 'elixr'}, {'stuff': 'elixr'.encode('utf-8')}, )) def test_unicode_multipart_post(self, httpbin, data): r = requests.post(httpbin('post'), data=data, files={'file': ('test_requests.py', open(__file__, 'rb'))}) assert r.status_code == 200 def test_unicode_multipart_post_fieldnames(self, httpbin): filename = os.path.splitext(__file__)[0] + '.py' r = requests.Request( method='POST', url=httpbin('post'), data={'stuff'.encode('utf-8'): 'elixr'}, files={'file': ('test_requests.py', open(filename, 'rb'))}) prep = r.prepare() assert b'name="stuff"' in prep.body assert b'name="b\'stuff\'"' not in prep.body def test_unicode_method_name(self, httpbin): files = {'file': open(__file__, 'rb')} r = requests.request( method=u('POST'), url=httpbin('post'), files=files) assert r.status_code == 200 def test_unicode_method_name_with_request_object(self, httpbin): files = {'file': open(__file__, 'rb')} s = requests.Session() req = requests.Request(u('POST'), httpbin('post'), files=files) prep = s.prepare_request(req) assert isinstance(prep.method, builtin_str) assert prep.method == 'POST' resp = s.send(prep) assert resp.status_code == 200 def test_non_prepared_request_error(self): s = requests.Session() req = requests.Request(u('POST'), '/') with pytest.raises(ValueError) as e: s.send(req) assert str(e.value) == 'You can only send PreparedRequests.' def test_custom_content_type(self, httpbin): r = requests.post( httpbin('post'), data={'stuff': json.dumps({'a': 123})}, files={ 'file1': ('test_requests.py', open(__file__, 'rb')), 'file2': ('test_requests', open(__file__, 'rb'), 'text/py-content-type')}) assert r.status_code == 200 assert b"text/py-content-type" in r.request.body def test_hook_receives_request_arguments(self, httpbin): def hook(resp, **kwargs): assert resp is not None assert kwargs != {} s = requests.Session() r = requests.Request('GET', httpbin(), hooks={'response': hook}) prep = s.prepare_request(r) s.send(prep) def test_session_hooks_are_used_with_no_request_hooks(self, httpbin): hook = lambda x, *args, **kwargs: x s = requests.Session() s.hooks['response'].append(hook) r = requests.Request('GET', httpbin()) prep = s.prepare_request(r) assert prep.hooks['response'] != [] assert prep.hooks['response'] == [hook] def test_session_hooks_are_overridden_by_request_hooks(self, httpbin): hook1 = lambda x, *args, **kwargs: x hook2 = lambda x, *args, **kwargs: x assert hook1 is not hook2 s = requests.Session() s.hooks['response'].append(hook2) r = requests.Request('GET', httpbin(), hooks={'response': [hook1]}) prep = s.prepare_request(r) assert prep.hooks['response'] == [hook1] def test_prepared_request_hook(self, httpbin): def hook(resp, **kwargs): resp.hook_working = True return resp req = requests.Request('GET', httpbin(), hooks={'response': hook}) prep = req.prepare() s = requests.Session() s.proxies = getproxies() resp = s.send(prep) assert hasattr(resp, 'hook_working') def test_prepared_from_session(self, httpbin): class DummyAuth(requests.auth.AuthBase): def __call__(self, r): r.headers['Dummy-Auth-Test'] = 'dummy-auth-test-ok' return r req = requests.Request('GET', httpbin('headers')) assert not req.auth s = requests.Session() s.auth = DummyAuth() prep = s.prepare_request(req) resp = s.send(prep) assert resp.json()['headers'][ 'Dummy-Auth-Test'] == 'dummy-auth-test-ok' def test_prepare_request_with_bytestring_url(self): req = requests.Request('GET', b'https://httpbin.org/') s = requests.Session() prep = s.prepare_request(req) assert prep.url == "https://httpbin.org/" def test_request_with_bytestring_host(self, httpbin): s = requests.Session() resp = s.request( 'GET', httpbin('cookies/set?cookie=value'), allow_redirects=False, headers={'Host': b'httpbin.org'} ) assert resp.cookies.get('cookie') == 'value' def test_links(self): r = requests.Response() r.headers = { 'cache-control': 'public, max-age=60, s-maxage=60', 'connection': 'keep-alive', 'content-encoding': 'gzip', 'content-type': 'application/json; charset=utf-8', 'date': 'Sat, 26 Jan 2013 16:47:56 GMT', 'etag': '"6ff6a73c0e446c1f61614769e3ceb778"', 'last-modified': 'Sat, 26 Jan 2013 16:22:39 GMT', 'link': ('; rel="next", ; ' ' rel="last"'), 'server': 'GitHub.com', 'status': '200 OK', 'vary': 'Accept', 'x-content-type-options': 'nosniff', 'x-github-media-type': 'github.beta', 'x-ratelimit-limit': '60', 'x-ratelimit-remaining': '57' } assert r.links['next']['rel'] == 'next' def test_cookie_parameters(self): key = 'some_cookie' value = 'some_value' secure = True domain = 'test.com' rest = {'HttpOnly': True} jar = requests.cookies.RequestsCookieJar() jar.set(key, value, secure=secure, domain=domain, rest=rest) assert len(jar) == 1 assert 'some_cookie' in jar cookie = list(jar)[0] assert cookie.secure == secure assert cookie.domain == domain assert cookie._rest['HttpOnly'] == rest['HttpOnly'] def test_cookie_as_dict_keeps_len(self): key = 'some_cookie' value = 'some_value' key1 = 'some_cookie1' value1 = 'some_value1' jar = requests.cookies.RequestsCookieJar() jar.set(key, value) jar.set(key1, value1) d1 = dict(jar) d2 = dict(jar.iteritems()) d3 = dict(jar.items()) assert len(jar) == 2 assert len(d1) == 2 assert len(d2) == 2 assert len(d3) == 2 def test_cookie_as_dict_keeps_items(self): key = 'some_cookie' value = 'some_value' key1 = 'some_cookie1' value1 = 'some_value1' jar = requests.cookies.RequestsCookieJar() jar.set(key, value) jar.set(key1, value1) d1 = dict(jar) d2 = dict(jar.iteritems()) d3 = dict(jar.items()) assert d1['some_cookie'] == 'some_value' assert d2['some_cookie'] == 'some_value' assert d3['some_cookie1'] == 'some_value1' def test_cookie_as_dict_keys(self): key = 'some_cookie' value = 'some_value' key1 = 'some_cookie1' value1 = 'some_value1' jar = requests.cookies.RequestsCookieJar() jar.set(key, value) jar.set(key1, value1) keys = jar.keys() assert keys == list(keys) # make sure one can use keys multiple times assert list(keys) == list(keys) def test_cookie_as_dict_values(self): key = 'some_cookie' value = 'some_value' key1 = 'some_cookie1' value1 = 'some_value1' jar = requests.cookies.RequestsCookieJar() jar.set(key, value) jar.set(key1, value1) values = jar.values() assert values == list(values) # make sure one can use values multiple times assert list(values) == list(values) def test_cookie_as_dict_items(self): key = 'some_cookie' value = 'some_value' key1 = 'some_cookie1' value1 = 'some_value1' jar = requests.cookies.RequestsCookieJar() jar.set(key, value) jar.set(key1, value1) items = jar.items() assert items == list(items) # make sure one can use items multiple times assert list(items) == list(items) def test_cookie_duplicate_names_different_domains(self): key = 'some_cookie' value = 'some_value' domain1 = 'test1.com' domain2 = 'test2.com' jar = requests.cookies.RequestsCookieJar() jar.set(key, value, domain=domain1) jar.set(key, value, domain=domain2) assert key in jar items = jar.items() assert len(items) == 2 # Verify that CookieConflictError is raised if domain is not specified with pytest.raises(requests.cookies.CookieConflictError): jar.get(key) # Verify that CookieConflictError is not raised if domain is specified cookie = jar.get(key, domain=domain1) assert cookie == value def test_cookie_duplicate_names_raises_cookie_conflict_error(self): key = 'some_cookie' value = 'some_value' path = 'some_path' jar = requests.cookies.RequestsCookieJar() jar.set(key, value, path=path) jar.set(key, value) with pytest.raises(requests.cookies.CookieConflictError): jar.get(key) def test_cookie_policy_copy(self): class MyCookiePolicy(cookielib.DefaultCookiePolicy): pass jar = requests.cookies.RequestsCookieJar() jar.set_policy(MyCookiePolicy()) assert isinstance(jar.copy().get_policy(), MyCookiePolicy) def test_time_elapsed_blank(self, httpbin): r = requests.get(httpbin('get')) td = r.elapsed total_seconds = ((td.microseconds + (td.seconds + td.days * 24 * 3600) * 10**6) / 10**6) assert total_seconds > 0.0 def test_empty_response_has_content_none(self): r = requests.Response() assert r.content is None def test_response_is_iterable(self): r = requests.Response() io = StringIO.StringIO('abc') read_ = io.read def read_mock(amt, decode_content=None): return read_(amt) setattr(io, 'read', read_mock) r.raw = io assert next(iter(r)) io.close() def test_response_decode_unicode(self): """When called with decode_unicode, Response.iter_content should always return unicode. """ r = requests.Response() r._content_consumed = True r._content = b'the content' r.encoding = 'ascii' chunks = r.iter_content(decode_unicode=True) assert all(isinstance(chunk, str) for chunk in chunks) # also for streaming r = requests.Response() r.raw = io.BytesIO(b'the content') r.encoding = 'ascii' chunks = r.iter_content(decode_unicode=True) assert all(isinstance(chunk, str) for chunk in chunks) def test_response_reason_unicode(self): # check for unicode HTTP status r = requests.Response() r.url = u'unicode URL' r.reason = u'Komponenttia ei löydy'.encode('utf-8') r.status_code = 404 r.encoding = None assert not r.ok # old behaviour - crashes here def test_response_reason_unicode_fallback(self): # check raise_status falls back to ISO-8859-1 r = requests.Response() r.url = 'some url' reason = u'Komponenttia ei löydy' r.reason = reason.encode('latin-1') r.status_code = 500 r.encoding = None with pytest.raises(requests.exceptions.HTTPError) as e: r.raise_for_status() assert reason in e.value.args[0] def test_response_chunk_size_type(self): """Ensure that chunk_size is passed as None or an integer, otherwise raise a TypeError. """ r = requests.Response() r.raw = io.BytesIO(b'the content') chunks = r.iter_content(1) assert all(len(chunk) == 1 for chunk in chunks) r = requests.Response() r.raw = io.BytesIO(b'the content') chunks = r.iter_content(None) assert list(chunks) == [b'the content'] r = requests.Response() r.raw = io.BytesIO(b'the content') with pytest.raises(TypeError): chunks = r.iter_content("1024") def test_request_and_response_are_pickleable(self, httpbin): r = requests.get(httpbin('get')) # verify we can pickle the original request assert pickle.loads(pickle.dumps(r.request)) # verify we can pickle the response and that we have access to # the original request. pr = pickle.loads(pickle.dumps(r)) assert r.request.url == pr.request.url assert r.request.headers == pr.request.headers def test_prepared_request_is_pickleable(self, httpbin): p = requests.Request('GET', httpbin('get')).prepare() # Verify PreparedRequest can be pickled and unpickled r = pickle.loads(pickle.dumps(p)) assert r.url == p.url assert r.headers == p.headers assert r.body == p.body # Verify unpickled PreparedRequest sends properly s = requests.Session() resp = s.send(r) assert resp.status_code == 200 def test_prepared_request_with_file_is_pickleable(self, httpbin): files = {'file': open(__file__, 'rb')} r = requests.Request('POST', httpbin('post'), files=files) p = r.prepare() # Verify PreparedRequest can be pickled and unpickled r = pickle.loads(pickle.dumps(p)) assert r.url == p.url assert r.headers == p.headers assert r.body == p.body # Verify unpickled PreparedRequest sends properly s = requests.Session() resp = s.send(r) assert resp.status_code == 200 def test_prepared_request_with_hook_is_pickleable(self, httpbin): r = requests.Request('GET', httpbin('get'), hooks=default_hooks()) p = r.prepare() # Verify PreparedRequest can be pickled r = pickle.loads(pickle.dumps(p)) assert r.url == p.url assert r.headers == p.headers assert r.body == p.body assert r.hooks == p.hooks # Verify unpickled PreparedRequest sends properly s = requests.Session() resp = s.send(r) assert resp.status_code == 200 def test_cannot_send_unprepared_requests(self, httpbin): r = requests.Request(url=httpbin()) with pytest.raises(ValueError): requests.Session().send(r) def test_http_error(self): error = requests.exceptions.HTTPError() assert not error.response response = requests.Response() error = requests.exceptions.HTTPError(response=response) assert error.response == response error = requests.exceptions.HTTPError('message', response=response) assert str(error) == 'message' assert error.response == response def test_session_pickling(self, httpbin): r = requests.Request('GET', httpbin('get')) s = requests.Session() s = pickle.loads(pickle.dumps(s)) s.proxies = getproxies() r = s.send(r.prepare()) assert r.status_code == 200 def test_fixes_1329(self, httpbin): """Ensure that header updates are done case-insensitively.""" s = requests.Session() s.headers.update({'ACCEPT': 'BOGUS'}) s.headers.update({'accept': 'application/json'}) r = s.get(httpbin('get')) headers = r.request.headers assert headers['accept'] == 'application/json' assert headers['Accept'] == 'application/json' assert headers['ACCEPT'] == 'application/json' def test_uppercase_scheme_redirect(self, httpbin): parts = urlparse(httpbin('html')) url = "HTTP://" + parts.netloc + parts.path r = requests.get(httpbin('redirect-to'), params={'url': url}) assert r.status_code == 200 assert r.url.lower() == url.lower() def test_transport_adapter_ordering(self): s = requests.Session() order = ['https://', 'http://'] assert order == list(s.adapters) s.mount('http://git', HTTPAdapter()) s.mount('http://github', HTTPAdapter()) s.mount('http://github.com', HTTPAdapter()) s.mount('http://github.com/about/', HTTPAdapter()) order = [ 'http://github.com/about/', 'http://github.com', 'http://github', 'http://git', 'https://', 'http://', ] assert order == list(s.adapters) s.mount('http://gittip', HTTPAdapter()) s.mount('http://gittip.com', HTTPAdapter()) s.mount('http://gittip.com/about/', HTTPAdapter()) order = [ 'http://github.com/about/', 'http://gittip.com/about/', 'http://github.com', 'http://gittip.com', 'http://github', 'http://gittip', 'http://git', 'https://', 'http://', ] assert order == list(s.adapters) s2 = requests.Session() s2.adapters = {'http://': HTTPAdapter()} s2.mount('https://', HTTPAdapter()) assert 'http://' in s2.adapters assert 'https://' in s2.adapters def test_session_get_adapter_prefix_matching(self): prefix = 'https://example.com' more_specific_prefix = prefix + '/some/path' url_matching_only_prefix = prefix + '/another/path' url_matching_more_specific_prefix = more_specific_prefix + '/longer/path' url_not_matching_prefix = 'https://another.example.com/' s = requests.Session() prefix_adapter = HTTPAdapter() more_specific_prefix_adapter = HTTPAdapter() s.mount(prefix, prefix_adapter) s.mount(more_specific_prefix, more_specific_prefix_adapter) assert s.get_adapter(url_matching_only_prefix) is prefix_adapter assert s.get_adapter(url_matching_more_specific_prefix) is more_specific_prefix_adapter assert s.get_adapter(url_not_matching_prefix) not in (prefix_adapter, more_specific_prefix_adapter) def test_session_get_adapter_prefix_matching_mixed_case(self): mixed_case_prefix = 'hTtPs://eXamPle.CoM/MixEd_CAse_PREfix' url_matching_prefix = mixed_case_prefix + '/full_url' s = requests.Session() my_adapter = HTTPAdapter() s.mount(mixed_case_prefix, my_adapter) assert s.get_adapter(url_matching_prefix) is my_adapter def test_session_get_adapter_prefix_matching_is_case_insensitive(self): mixed_case_prefix = 'hTtPs://eXamPle.CoM/MixEd_CAse_PREfix' url_matching_prefix_with_different_case = 'HtTpS://exaMPLe.cOm/MiXeD_caSE_preFIX/another_url' s = requests.Session() my_adapter = HTTPAdapter() s.mount(mixed_case_prefix, my_adapter) assert s.get_adapter(url_matching_prefix_with_different_case) is my_adapter def test_header_remove_is_case_insensitive(self, httpbin): # From issue #1321 s = requests.Session() s.headers['foo'] = 'bar' r = s.get(httpbin('get'), headers={'FOO': None}) assert 'foo' not in r.request.headers def test_params_are_merged_case_sensitive(self, httpbin): s = requests.Session() s.params['foo'] = 'bar' r = s.get(httpbin('get'), params={'FOO': 'bar'}) assert r.json()['args'] == {'foo': 'bar', 'FOO': 'bar'} def test_long_authinfo_in_url(self): url = 'http://{}:{}@{}:9000/path?query#frag'.format( 'E8A3BE87-9E3F-4620-8858-95478E385B5B', 'EA770032-DA4D-4D84-8CE9-29C6D910BF1E', 'exactly-------------sixty-----------three------------characters', ) r = requests.Request('GET', url).prepare() assert r.url == url def test_header_keys_are_native(self, httpbin): headers = {u('unicode'): 'blah', 'byte'.encode('ascii'): 'blah'} r = requests.Request('GET', httpbin('get'), headers=headers) p = r.prepare() # This is testing that they are builtin strings. A bit weird, but there # we go. assert 'unicode' in p.headers.keys() assert 'byte' in p.headers.keys() def test_header_validation(self, httpbin): """Ensure prepare_headers regex isn't flagging valid header contents.""" headers_ok = {'foo': 'bar baz qux', 'bar': u'fbbq'.encode('utf8'), 'baz': '', 'qux': '1'} r = requests.get(httpbin('get'), headers=headers_ok) assert r.request.headers['foo'] == headers_ok['foo'] def test_header_value_not_str(self, httpbin): """Ensure the header value is of type string or bytes as per discussion in GH issue #3386 """ headers_int = {'foo': 3} headers_dict = {'bar': {'foo': 'bar'}} headers_list = {'baz': ['foo', 'bar']} # Test for int with pytest.raises(InvalidHeader) as excinfo: r = requests.get(httpbin('get'), headers=headers_int) assert 'foo' in str(excinfo.value) # Test for dict with pytest.raises(InvalidHeader) as excinfo: r = requests.get(httpbin('get'), headers=headers_dict) assert 'bar' in str(excinfo.value) # Test for list with pytest.raises(InvalidHeader) as excinfo: r = requests.get(httpbin('get'), headers=headers_list) assert 'baz' in str(excinfo.value) def test_header_no_return_chars(self, httpbin): """Ensure that a header containing return character sequences raise an exception. Otherwise, multiple headers are created from single string. """ headers_ret = {'foo': 'bar\r\nbaz: qux'} headers_lf = {'foo': 'bar\nbaz: qux'} headers_cr = {'foo': 'bar\rbaz: qux'} # Test for newline with pytest.raises(InvalidHeader): r = requests.get(httpbin('get'), headers=headers_ret) # Test for line feed with pytest.raises(InvalidHeader): r = requests.get(httpbin('get'), headers=headers_lf) # Test for carriage return with pytest.raises(InvalidHeader): r = requests.get(httpbin('get'), headers=headers_cr) def test_header_no_leading_space(self, httpbin): """Ensure headers containing leading whitespace raise InvalidHeader Error before sending. """ headers_space = {'foo': ' bar'} headers_tab = {'foo': ' bar'} # Test for whitespace with pytest.raises(InvalidHeader): r = requests.get(httpbin('get'), headers=headers_space) # Test for tab with pytest.raises(InvalidHeader): r = requests.get(httpbin('get'), headers=headers_tab) @pytest.mark.parametrize('files', ('foo', b'foo', bytearray(b'foo'))) def test_can_send_objects_with_files(self, httpbin, files): data = {'a': 'this is a string'} files = {'b': files} 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, httpbin): 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, httpbin): data = 'this is a string' length = '16' req = requests.Request('POST', httpbin('post'), data=data) p = req.prepare() assert p.headers['Content-Length'] == length def test_nonhttp_schemes_dont_check_URLs(self): test_urls = ( 'data:image/gif;base64,R0lGODlhAQABAHAAACH5BAUAAAAALAAAAAABAAEAAAICRAEAOw==', 'file:///etc/passwd', 'magnet:?xt=urn:btih:be08f00302bc2d1d3cfa3af02024fa647a271431', ) for test_url in test_urls: req = requests.Request('GET', test_url) preq = req.prepare() assert test_url == preq.url def test_auth_is_stripped_on_http_downgrade(self, httpbin, httpbin_secure, httpbin_ca_bundle): r = requests.get( httpbin_secure('redirect-to'), params={'url': httpbin('get')}, auth=('user', 'pass'), verify=httpbin_ca_bundle ) assert r.history[0].request.headers['Authorization'] assert 'Authorization' not in r.request.headers def test_auth_is_retained_for_redirect_on_host(self, httpbin): r = requests.get(httpbin('redirect/1'), auth=('user', 'pass')) h1 = r.history[0].request.headers['Authorization'] h2 = r.request.headers['Authorization'] assert h1 == h2 def test_should_strip_auth_host_change(self): s = requests.Session() assert s.should_strip_auth('http://example.com/foo', 'http://another.example.com/') def test_should_strip_auth_http_downgrade(self): s = requests.Session() assert s.should_strip_auth('https://example.com/foo', 'http://example.com/bar') def test_should_strip_auth_https_upgrade(self): s = requests.Session() assert not s.should_strip_auth('http://example.com/foo', 'https://example.com/bar') assert not s.should_strip_auth('http://example.com:80/foo', 'https://example.com/bar') assert not s.should_strip_auth('http://example.com/foo', 'https://example.com:443/bar') # Non-standard ports should trigger stripping assert s.should_strip_auth('http://example.com:8080/foo', 'https://example.com/bar') assert s.should_strip_auth('http://example.com/foo', 'https://example.com:8443/bar') def test_should_strip_auth_port_change(self): s = requests.Session() assert s.should_strip_auth('http://example.com:1234/foo', 'https://example.com:4321/bar') @pytest.mark.parametrize( 'old_uri, new_uri', ( ('https://example.com:443/foo', 'https://example.com/bar'), ('http://example.com:80/foo', 'http://example.com/bar'), ('https://example.com/foo', 'https://example.com:443/bar'), ('http://example.com/foo', 'http://example.com:80/bar') )) def test_should_strip_auth_default_port(self, old_uri, new_uri): s = requests.Session() assert not s.should_strip_auth(old_uri, new_uri) def test_manual_redirect_with_partial_body_read(self, httpbin): s = requests.Session() r1 = s.get(httpbin('redirect/2'), allow_redirects=False, stream=True) assert r1.is_redirect rg = s.resolve_redirects(r1, r1.request, stream=True) # read only the first eight bytes of the response body, # then follow the redirect r1.iter_content(8) r2 = next(rg) assert r2.is_redirect # read all of the response via iter_content, # then follow the redirect for _ in r2.iter_content(): pass r3 = next(rg) assert not r3.is_redirect def test_prepare_body_position_non_stream(self): data = b'the data' prep = requests.Request('GET', 'http://example.com', data=data).prepare() assert prep._body_position is None def test_rewind_body(self): data = io.BytesIO(b'the data') prep = requests.Request('GET', 'http://example.com', data=data).prepare() assert prep._body_position == 0 assert prep.body.read() == b'the data' # the data has all been read assert prep.body.read() == b'' # rewind it back requests.utils.rewind_body(prep) assert prep.body.read() == b'the data' def test_rewind_partially_read_body(self): data = io.BytesIO(b'the data') data.read(4) # read some data prep = requests.Request('GET', 'http://example.com', data=data).prepare() assert prep._body_position == 4 assert prep.body.read() == b'data' # the data has all been read assert prep.body.read() == b'' # rewind it back requests.utils.rewind_body(prep) assert prep.body.read() == b'data' def test_rewind_body_no_seek(self): class BadFileObj: def __init__(self, data): self.data = data def tell(self): return 0 def __iter__(self): return data = BadFileObj('the data') prep = requests.Request('GET', 'http://example.com', data=data).prepare() assert prep._body_position == 0 with pytest.raises(UnrewindableBodyError) as e: requests.utils.rewind_body(prep) assert 'Unable to rewind request body' in str(e) def test_rewind_body_failed_seek(self): class BadFileObj: def __init__(self, data): self.data = data def tell(self): return 0 def seek(self, pos, whence=0): raise OSError() def __iter__(self): return data = BadFileObj('the data') prep = requests.Request('GET', 'http://example.com', data=data).prepare() assert prep._body_position == 0 with pytest.raises(UnrewindableBodyError) as e: requests.utils.rewind_body(prep) assert 'error occurred when rewinding request body' in str(e) def test_rewind_body_failed_tell(self): class BadFileObj: def __init__(self, data): self.data = data def tell(self): raise OSError() def __iter__(self): return data = BadFileObj('the data') prep = requests.Request('GET', 'http://example.com', data=data).prepare() assert prep._body_position is not None with pytest.raises(UnrewindableBodyError) as e: requests.utils.rewind_body(prep) assert 'Unable to rewind request body' in str(e) def _patch_adapter_gzipped_redirect(self, session, url): adapter = session.get_adapter(url=url) org_build_response = adapter.build_response self._patched_response = False def build_response(*args, **kwargs): resp = org_build_response(*args, **kwargs) if not self._patched_response: resp.raw.headers['content-encoding'] = 'gzip' self._patched_response = True return resp adapter.build_response = build_response def test_redirect_with_wrong_gzipped_header(self, httpbin): s = requests.Session() url = httpbin('redirect/1') self._patch_adapter_gzipped_redirect(s, url) s.get(url) @pytest.mark.parametrize( 'username, password, auth_str', ( ('test', 'test', 'Basic dGVzdDp0ZXN0'), (u'имя'.encode('utf-8'), u'пароль'.encode('utf-8'), 'Basic 0LjQvNGPOtC/0LDRgNC+0LvRjA=='), )) def test_basic_auth_str_is_always_native(self, username, password, auth_str): s = _basic_auth_str(username, password) assert isinstance(s, builtin_str) assert s == auth_str def test_requests_history_is_saved(self, httpbin): 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 += 1 def test_json_param_post_content_type_works(self, httpbin): r = requests.post( httpbin('post'), json={'life': 42} ) assert r.status_code == 200 assert 'application/json' in r.request.headers['Content-Type'] assert {'life': 42} == r.json()['json'] def test_json_param_post_should_not_override_data_param(self, httpbin): r = requests.Request(method='POST', url=httpbin('post'), data={'stuff': 'elixr'}, json={'music': 'flute'}) prep = r.prepare() assert 'stuff=elixr' == prep.body def test_response_iter_lines(self, httpbin): r = requests.get(httpbin('stream/4'), stream=True) assert r.status_code == 200 it = r.iter_lines() next(it) assert len(list(it)) == 3 def test_response_context_manager(self, httpbin): with requests.get(httpbin('stream/4'), stream=True) as response: assert isinstance(response, requests.Response) assert response.raw.closed def test_unconsumed_session_response_closes_connection(self, httpbin): s = requests.session() with contextlib.closing(s.get(httpbin('stream/4'), stream=True)) as response: pass assert response._content_consumed is False assert response.raw.closed @pytest.mark.xfail def test_response_iter_lines_reentrant(self, httpbin): """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 def test_session_close_proxy_clear(self, mocker): proxies = { 'one': mocker.Mock(), 'two': mocker.Mock(), } session = requests.Session() mocker.patch.dict(session.adapters['http://'].proxy_manager, proxies) session.close() proxies['one'].clear.assert_called_once_with() proxies['two'].clear.assert_called_once_with() def test_proxy_auth(self): adapter = HTTPAdapter() headers = adapter.proxy_headers("http://user:pass@httpbin.org") assert headers == {'Proxy-Authorization': 'Basic dXNlcjpwYXNz'} def test_proxy_auth_empty_pass(self): adapter = HTTPAdapter() headers = adapter.proxy_headers("http://user:@httpbin.org") assert headers == {'Proxy-Authorization': 'Basic dXNlcjo='} def test_response_json_when_content_is_None(self, httpbin): r = requests.get(httpbin('/status/204')) # Make sure r.content is None r.status_code = 0 r._content = False r._content_consumed = False assert r.content is None with pytest.raises(ValueError): r.json() def test_response_without_release_conn(self): """Test `close` call for non-urllib3-like raw objects. Should work when `release_conn` attr doesn't exist on `response.raw`. """ resp = requests.Response() resp.raw = StringIO.StringIO('test') assert not resp.raw.closed resp.close() assert resp.raw.closed def test_empty_stream_with_auth_does_not_set_content_length_header(self, httpbin): """Ensure that a byte stream with size 0 will not set both a Content-Length and Transfer-Encoding header. """ auth = ('user', 'pass') url = httpbin('post') file_obj = io.BytesIO(b'') r = requests.Request('POST', url, auth=auth, data=file_obj) prepared_request = r.prepare() assert 'Transfer-Encoding' in prepared_request.headers assert 'Content-Length' not in prepared_request.headers def test_stream_with_auth_does_not_set_transfer_encoding_header(self, httpbin): """Ensure that a byte stream with size > 0 will not set both a Content-Length and Transfer-Encoding header. """ auth = ('user', 'pass') url = httpbin('post') file_obj = io.BytesIO(b'test data') r = requests.Request('POST', url, auth=auth, data=file_obj) prepared_request = r.prepare() assert 'Transfer-Encoding' not in prepared_request.headers assert 'Content-Length' in prepared_request.headers def test_chunked_upload_does_not_set_content_length_header(self, httpbin): """Ensure that requests with a generator body stream using Transfer-Encoding: chunked, not a Content-Length header. """ data = (i for i in [b'a', b'b', b'c']) url = httpbin('post') r = requests.Request('POST', url, data=data) prepared_request = r.prepare() assert 'Transfer-Encoding' in prepared_request.headers assert 'Content-Length' not in prepared_request.headers def test_custom_redirect_mixin(self, httpbin): """Tests a custom mixin to overwrite ``get_redirect_target``. Ensures a subclassed ``requests.Session`` can handle a certain type of malformed redirect responses. 1. original request receives a proper response: 302 redirect 2. following the redirect, a malformed response is given: status code = HTTP 200 location = alternate url 3. the custom session catches the edge case and follows the redirect """ url_final = httpbin('html') querystring_malformed = urlencode({'location': url_final}) url_redirect_malformed = httpbin('response-headers?%s' % querystring_malformed) querystring_redirect = urlencode({'url': url_redirect_malformed}) url_redirect = httpbin('redirect-to?%s' % querystring_redirect) urls_test = [url_redirect, url_redirect_malformed, url_final, ] class CustomRedirectSession(requests.Session): def get_redirect_target(self, resp): # default behavior if resp.is_redirect: return resp.headers['location'] # edge case - check to see if 'location' is in headers anyways location = resp.headers.get('location') if location and (location != resp.url): return location return None session = CustomRedirectSession() r = session.get(urls_test[0]) assert len(r.history) == 2 assert r.status_code == 200 assert r.history[0].status_code == 302 assert r.history[0].is_redirect assert r.history[1].status_code == 200 assert not r.history[1].is_redirect assert r.url == urls_test[2] class TestCaseInsensitiveDict: @pytest.mark.parametrize( 'cid', ( CaseInsensitiveDict({'Foo': 'foo', 'BAr': 'bar'}), CaseInsensitiveDict([('Foo', 'foo'), ('BAr', 'bar')]), CaseInsensitiveDict(FOO='foo', BAr='bar'), )) def test_init(self, cid): assert len(cid) == 2 assert 'foo' in cid assert 'bar' in cid def test_docstring_example(self): cid = CaseInsensitiveDict() cid['Accept'] = 'application/json' assert cid['aCCEPT'] == 'application/json' assert list(cid) == ['Accept'] def test_len(self): cid = CaseInsensitiveDict({'a': 'a', 'b': 'b'}) cid['A'] = 'a' assert len(cid) == 2 def test_getitem(self): cid = CaseInsensitiveDict({'Spam': 'blueval'}) assert cid['spam'] == 'blueval' assert cid['SPAM'] == 'blueval' def test_fixes_649(self): """__setitem__ should behave case-insensitively.""" cid = CaseInsensitiveDict() cid['spam'] = 'oneval' cid['Spam'] = 'twoval' cid['sPAM'] = 'redval' cid['SPAM'] = 'blueval' assert cid['spam'] == 'blueval' assert cid['SPAM'] == 'blueval' assert list(cid.keys()) == ['SPAM'] def test_delitem(self): cid = CaseInsensitiveDict() cid['Spam'] = 'someval' del cid['sPam'] assert 'spam' not in cid assert len(cid) == 0 def test_contains(self): cid = CaseInsensitiveDict() cid['Spam'] = 'someval' assert 'Spam' in cid assert 'spam' in cid assert 'SPAM' in cid assert 'sPam' in cid assert 'notspam' not in cid def test_get(self): cid = CaseInsensitiveDict() cid['spam'] = 'oneval' cid['SPAM'] = 'blueval' assert cid.get('spam') == 'blueval' assert cid.get('SPAM') == 'blueval' assert cid.get('sPam') == 'blueval' assert cid.get('notspam', 'default') == 'default' def test_update(self): cid = CaseInsensitiveDict() cid['spam'] = 'blueval' cid.update({'sPam': 'notblueval'}) assert cid['spam'] == 'notblueval' cid = CaseInsensitiveDict({'Foo': 'foo', 'BAr': 'bar'}) cid.update({'fOO': 'anotherfoo', 'bAR': 'anotherbar'}) assert len(cid) == 2 assert cid['foo'] == 'anotherfoo' assert cid['bar'] == 'anotherbar' def test_update_retains_unchanged(self): cid = CaseInsensitiveDict({'foo': 'foo', 'bar': 'bar'}) cid.update({'foo': 'newfoo'}) assert cid['bar'] == 'bar' def test_iter(self): cid = CaseInsensitiveDict({'Spam': 'spam', 'Eggs': 'eggs'}) keys = frozenset(['Spam', 'Eggs']) assert frozenset(iter(cid)) == keys def test_equality(self): cid = CaseInsensitiveDict({'SPAM': 'blueval', 'Eggs': 'redval'}) othercid = CaseInsensitiveDict({'spam': 'blueval', 'eggs': 'redval'}) assert cid == othercid del othercid['spam'] assert cid != othercid assert cid == {'spam': 'blueval', 'eggs': 'redval'} assert cid != object() def test_setdefault(self): cid = CaseInsensitiveDict({'Spam': 'blueval'}) assert cid.setdefault('spam', 'notblueval') == 'blueval' assert cid.setdefault('notspam', 'notblueval') == 'notblueval' def test_lower_items(self): cid = CaseInsensitiveDict({ 'Accept': 'application/json', 'user-Agent': 'requests', }) keyset = frozenset(lowerkey for lowerkey, v in cid.lower_items()) lowerkeyset = frozenset(['accept', 'user-agent']) assert keyset == lowerkeyset def test_preserve_key_case(self): cid = CaseInsensitiveDict({ 'Accept': 'application/json', 'user-Agent': 'requests', }) keyset = frozenset(['Accept', 'user-Agent']) assert frozenset(i[0] for i in cid.items()) == keyset assert frozenset(cid.keys()) == keyset assert frozenset(cid) == keyset def test_preserve_last_key_case(self): cid = CaseInsensitiveDict({ 'Accept': 'application/json', 'user-Agent': 'requests', }) cid.update({'ACCEPT': 'application/json'}) cid['USER-AGENT'] = 'requests' keyset = frozenset(['ACCEPT', 'USER-AGENT']) assert frozenset(i[0] for i in cid.items()) == keyset assert frozenset(cid.keys()) == keyset assert frozenset(cid) == keyset def test_copy(self): cid = CaseInsensitiveDict({ 'Accept': 'application/json', 'user-Agent': 'requests', }) cid_copy = cid.copy() assert cid == cid_copy cid['changed'] = True assert cid != cid_copy class TestMorselToCookieExpires: """Tests for morsel_to_cookie when morsel contains expires.""" def test_expires_valid_str(self): """Test case where we convert expires from string time.""" morsel = Morsel() morsel['expires'] = 'Thu, 01-Jan-1970 00:00:01 GMT' cookie = morsel_to_cookie(morsel) assert cookie.expires == 1 @pytest.mark.parametrize( 'value, exception', ( (100, TypeError), ('woops', ValueError), )) def test_expires_invalid_int(self, value, exception): """Test case where an invalid type is passed for expires.""" morsel = Morsel() morsel['expires'] = value with pytest.raises(exception): morsel_to_cookie(morsel) def test_expires_none(self): """Test case where expires is None.""" morsel = Morsel() morsel['expires'] = None cookie = morsel_to_cookie(morsel) assert cookie.expires is None class TestMorselToCookieMaxAge: """Tests for morsel_to_cookie when morsel contains max-age.""" def test_max_age_valid_int(self): """Test case where a valid max age in seconds is passed.""" morsel = Morsel() morsel['max-age'] = 60 cookie = morsel_to_cookie(morsel) assert isinstance(cookie.expires, int) def test_max_age_invalid_str(self): """Test case where a invalid max age is passed.""" morsel = Morsel() morsel['max-age'] = 'woops' with pytest.raises(TypeError): morsel_to_cookie(morsel) class TestTimeout: def test_stream_timeout(self, httpbin): try: requests.get(httpbin('delay/10'), timeout=2.0) except requests.exceptions.Timeout as e: assert 'Read timed out' in e.args[0].args[0] @pytest.mark.parametrize( 'timeout, error_text', ( ((3, 4, 5), '(connect, read)'), ('foo', 'must be an int, float or None'), )) def test_invalid_timeout(self, httpbin, timeout, error_text): with pytest.raises(ValueError) as e: requests.get(httpbin('get'), timeout=timeout) assert error_text in str(e) @pytest.mark.parametrize( 'timeout', ( None, Urllib3Timeout(connect=None, read=None) )) def test_none_timeout(self, httpbin, timeout): """Check that you can set None as a valid timeout value. To actually test this behavior, we'd want to check that setting the timeout to None actually lets the request block past the system default timeout. However, this would make the test suite unbearably slow. Instead we verify that setting the timeout to None does not prevent the request from succeeding. """ r = requests.get(httpbin('get'), timeout=timeout) assert r.status_code == 200 @pytest.mark.parametrize( 'timeout', ( (None, 0.1), Urllib3Timeout(connect=None, read=0.1) )) def test_read_timeout(self, httpbin, timeout): try: requests.get(httpbin('delay/10'), timeout=timeout) pytest.fail('The recv() request should time out.') except ReadTimeout: pass @pytest.mark.parametrize( 'timeout', ( (0.1, None), Urllib3Timeout(connect=0.1, read=None) )) def test_connect_timeout(self, timeout): try: requests.get(TARPIT, timeout=timeout) pytest.fail('The connect() request should time out.') except ConnectTimeout as e: assert isinstance(e, ConnectionError) assert isinstance(e, Timeout) @pytest.mark.parametrize( 'timeout', ( (0.1, 0.1), Urllib3Timeout(connect=0.1, read=0.1) )) def test_total_timeout_connect(self, timeout): try: requests.get(TARPIT, timeout=timeout) pytest.fail('The connect() request should time out.') except ConnectTimeout: pass def test_encoded_methods(self, httpbin): """See: https://github.com/psf/requests/issues/2316""" r = requests.request(b'GET', httpbin('get')) assert r.ok SendCall = collections.namedtuple('SendCall', ('args', 'kwargs')) class RedirectSession(SessionRedirectMixin): def __init__(self, order_of_redirects): self.redirects = order_of_redirects self.calls = [] self.max_redirects = 30 self.cookies = {} self.trust_env = False def send(self, *args, **kwargs): self.calls.append(SendCall(args, kwargs)) return self.build_response() def build_response(self): request = self.calls[-1].args[0] r = requests.Response() try: r.status_code = int(self.redirects.pop(0)) except IndexError: r.status_code = 200 r.headers = CaseInsensitiveDict({'Location': '/'}) r.raw = self._build_raw() r.request = request return r def _build_raw(self): string = StringIO.StringIO('') setattr(string, 'release_conn', lambda *args: args) return string def test_json_encodes_as_bytes(): # urllib3 expects bodies as bytes-like objects body = {"key": "value"} p = PreparedRequest() p.prepare( method='GET', url='https://www.example.com/', json=body ) assert isinstance(p.body, bytes) def test_requests_are_updated_each_time(httpbin): session = RedirectSession([303, 307]) prep = requests.Request('POST', httpbin('post')).prepare() r0 = session.send(prep) assert r0.request.method == 'POST' assert session.calls[-1] == SendCall((r0.request,), {}) redirect_generator = session.resolve_redirects(r0, prep) default_keyword_args = { 'stream': False, 'verify': True, 'cert': None, 'timeout': None, 'allow_redirects': False, 'proxies': {}, } for response in redirect_generator: assert response.request.method == 'GET' send_call = SendCall((response.request,), default_keyword_args) assert session.calls[-1] == send_call @pytest.mark.parametrize("var,url,proxy", [ ('http_proxy', 'http://example.com', 'socks5://proxy.com:9876'), ('https_proxy', 'https://example.com', 'socks5://proxy.com:9876'), ('all_proxy', 'http://example.com', 'socks5://proxy.com:9876'), ('all_proxy', 'https://example.com', 'socks5://proxy.com:9876'), ]) def test_proxy_env_vars_override_default(var, url, proxy): session = requests.Session() prep = PreparedRequest() prep.prepare(method='GET', url=url) kwargs = { var: proxy } scheme = urlparse(url).scheme with override_environ(**kwargs): proxies = session.rebuild_proxies(prep, {}) assert scheme in proxies assert proxies[scheme] == proxy @pytest.mark.parametrize( 'data', ( (('a', 'b'), ('c', 'd')), (('c', 'd'), ('a', 'b')), (('a', 'b'), ('c', 'd'), ('e', 'f')), )) def test_data_argument_accepts_tuples(data): """Ensure that the data argument will accept tuples of strings and properly encode them. """ p = PreparedRequest() p.prepare( method='GET', url='http://www.example.com', data=data, hooks=default_hooks() ) assert p.body == urlencode(data) @pytest.mark.parametrize( 'kwargs', ( None, { 'method': 'GET', 'url': 'http://www.example.com', 'data': 'foo=bar', 'hooks': default_hooks() }, { 'method': 'GET', 'url': 'http://www.example.com', 'data': 'foo=bar', 'hooks': default_hooks(), 'cookies': {'foo': 'bar'} }, { 'method': 'GET', 'url': u('http://www.example.com/üniçø∂é') }, )) def test_prepared_copy(kwargs): p = PreparedRequest() if kwargs: p.prepare(**kwargs) copy = p.copy() for attr in ('method', 'url', 'headers', '_cookies', 'body', 'hooks'): assert getattr(p, attr) == getattr(copy, attr) def test_urllib3_retries(httpbin): from 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_urllib3_pool_connection_closed(httpbin): s = requests.Session() s.mount('http://', HTTPAdapter(pool_connections=0, pool_maxsize=0)) try: s.get(httpbin('status/200')) except ConnectionError as e: assert u"Pool is closed." in str(e) class TestPreparingURLs(object): @pytest.mark.parametrize( 'url,expected', ( ('http://google.com', 'http://google.com/'), (u'http://ジェーピーニック.jp', u'http://xn--hckqz9bzb1cyrb.jp/'), (u'http://xn--n3h.net/', u'http://xn--n3h.net/'), ( u'http://ジェーピーニック.jp'.encode('utf-8'), u'http://xn--hckqz9bzb1cyrb.jp/' ), ( u'http://straße.de/straße', u'http://xn--strae-oqa.de/stra%C3%9Fe' ), ( u'http://straße.de/straße'.encode('utf-8'), u'http://xn--strae-oqa.de/stra%C3%9Fe' ), ( u'http://Königsgäßchen.de/straße', u'http://xn--knigsgchen-b4a3dun.de/stra%C3%9Fe' ), ( u'http://Königsgäßchen.de/straße'.encode('utf-8'), u'http://xn--knigsgchen-b4a3dun.de/stra%C3%9Fe' ), ( b'http://xn--n3h.net/', u'http://xn--n3h.net/' ), ( b'http://[1200:0000:ab00:1234:0000:2552:7777:1313]:12345/', u'http://[1200:0000:ab00:1234:0000:2552:7777:1313]:12345/' ), ( u'http://[1200:0000:ab00:1234:0000:2552:7777:1313]:12345/', u'http://[1200:0000:ab00:1234:0000:2552:7777:1313]:12345/' ) ) ) def test_preparing_url(self, url, expected): def normalize_percent_encode(x): # Helper function that normalizes equivalent # percent-encoded bytes before comparisons for c in re.findall(r'%[a-fA-F0-9]{2}', x): x = x.replace(c, c.upper()) return x r = requests.Request('GET', url=url) p = r.prepare() assert normalize_percent_encode(p.url) == expected @pytest.mark.parametrize( 'url', ( b"http://*.google.com", b"http://*", u"http://*.google.com", u"http://*", u"http://☃.net/" ) ) def test_preparing_bad_url(self, url): r = requests.Request('GET', url=url) with pytest.raises(requests.exceptions.InvalidURL): r.prepare() @pytest.mark.parametrize( 'url, exception', ( ('http://localhost:-1', InvalidURL), ) ) def test_redirecting_to_bad_url(self, httpbin, url, exception): with pytest.raises(exception): r = requests.get(httpbin('redirect-to'), params={'url': url}) @pytest.mark.parametrize( 'input, expected', ( ( b"http+unix://%2Fvar%2Frun%2Fsocket/path%7E", u"http+unix://%2Fvar%2Frun%2Fsocket/path~", ), ( u"http+unix://%2Fvar%2Frun%2Fsocket/path%7E", u"http+unix://%2Fvar%2Frun%2Fsocket/path~", ), ( b"mailto:user@example.org", u"mailto:user@example.org", ), ( u"mailto:user@example.org", u"mailto:user@example.org", ), ( b"data:SSDimaUgUHl0aG9uIQ==", u"data:SSDimaUgUHl0aG9uIQ==", ) ) ) def test_url_mutation(self, input, expected): """ This test validates that we correctly exclude some URLs from preparation, and that we handle others. Specifically, it tests that any URL whose scheme doesn't begin with "http" is left alone, and those whose scheme *does* begin with "http" are mutated. """ r = requests.Request('GET', url=input) p = r.prepare() assert p.url == expected @pytest.mark.parametrize( 'input, params, expected', ( ( b"http+unix://%2Fvar%2Frun%2Fsocket/path", {"key": "value"}, u"http+unix://%2Fvar%2Frun%2Fsocket/path?key=value", ), ( u"http+unix://%2Fvar%2Frun%2Fsocket/path", {"key": "value"}, u"http+unix://%2Fvar%2Frun%2Fsocket/path?key=value", ), ( b"mailto:user@example.org", {"key": "value"}, u"mailto:user@example.org", ), ( u"mailto:user@example.org", {"key": "value"}, u"mailto:user@example.org", ), ) ) def test_parameters_for_nonstandard_schemes(self, input, params, expected): """ Setting parameters for nonstandard schemes is allowed if those schemes begin with "http", and is forbidden otherwise. """ r = requests.Request('GET', url=input, params=params) p = r.prepare() assert p.url == expected