| # -*- 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, |
| SKIP_HEADER, |
| ) |
| 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) |
| if not SKIP_HEADER: |
| assert "Accept-Encoding" not in prep.headers |
| else: |
| assert SKIP_HEADER == prep.headers["Accept-Encoding"] |
| |
| 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': ('<https://api.github.com/users/kennethreitz/repos?' |
| 'page=2&per_page=10>; rel="next", <https://api.github.' |
| 'com/users/kennethreitz/repos?page=7&per_page=10>; ' |
| ' 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 |