|
|
@@ -1,2527 +0,0 @@
|
|
|
-# -*- 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('Pipfile') 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('Pipfile') 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('Pipfile') as f:
|
|
|
- pytest.raises(ValueError, "requests.post(url, data='[{\"some\": \"data\"}]', files={'some': f})")
|
|
|
- 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
|