This module contains the authentication handlers for Requests.
"""
+from __future__ import unicode_literals
+
import time
import hashlib
from base64 import b64encode
-from urlparse import urlparse
+# from urlparse import urlparse
+from urllib.parse import urlparse
from .utils import randombytes, parse_dict_header
def _basic_auth_str(username, password):
"""Returns a Basic Auth string."""
- return 'Basic %s' % b64encode('%s:%s' % (username, password))
+
+ return 'Basic ' + b64encode(("%s:%s" % (username, password)).encode('utf-8')).strip().decode('utf-8')
class AuthBase(object):
class HTTPBasicAuth(AuthBase):
"""Attaches HTTP Basic Authentication to the given Request object."""
def __init__(self, username, password):
- self.username = str(username)
- self.password = str(password)
+ self.username = username
+ self.password = password
def __call__(self, r):
r.headers['Authorization'] = _basic_auth_str(self.username, self.password)
last_nonce = nonce
ncvalue = '%08x' % nonce_count
- cnonce = (hashlib.sha1("%s:%s:%s:%s" % (
- nonce_count, nonce, time.ctime(), randombytes(8)))
- .hexdigest()[:16]
- )
+ s = str(nonce_count).encode('utf-8')
+ s += nonce.encode('utf-8')
+ s += time.ctime().encode('utf-8')
+ s += randombytes(8)
+
+ cnonce = (hashlib.sha1(s).hexdigest()[:16])
noncebit = "%s:%s:%s:%s:%s" % (nonce, ncvalue, cnonce, qop, H(A2))
respdig = KD(H(A1), noncebit)
elif qop is None:
"""
import os
-import urllib
+import urllib.request, urllib.parse, urllib.error
-from urlparse import urlparse, urlunparse, urljoin, urlsplit
+# from urlparse import urlparse, urlunparse, urljoin, urlsplit
+from urllib.parse import urlparse, urlunparse, urljoin, urlsplit, urlencode
from datetime import datetime
from .hooks import dispatch_hook, HOOKS
from .structures import CaseInsensitiveDict
from .status_codes import codes
-from .packages import oreos
+# from .packages import oreos
from .auth import HTTPBasicAuth, HTTPProxyAuth
from .packages.urllib3.response import HTTPResponse
from .packages.urllib3.exceptions import MaxRetryError
URLRequired, SSLError)
from .utils import (
get_encoding_from_headers, stream_decode_response_unicode,
- stream_decompress, guess_filename, requote_path)
+ stream_decompress, guess_filename, requote_path, dict_from_string)
# Import chardet if it is available.
try:
self.timeout = timeout
#: Request URL.
+
+ # if isinstance(url, str):
+ # url = url.encode('utf-8')
+ # print(dir(url))
+
self.url = url
#: Dictionary of HTTP Headers to attach to the :class:`Request <Request>`.
hooks = hooks or {}
- for (k, v) in hooks.items():
+ for (k, v) in list(hooks.items()):
self.register_hook(event=k, hook=v)
#: Session.
headers = CaseInsensitiveDict()
# Add configured base headers.
- for (k, v) in self.config.get('base_headers', {}).items():
+ for (k, v) in list(self.config.get('base_headers', {}).items()):
if k not in headers:
headers[k] = v
# Add new cookies from the server.
if 'set-cookie' in response.headers:
cookie_header = response.headers['set-cookie']
- cookies = oreos.dict_from_string(cookie_header)
+ cookies = dict_from_string(cookie_header)
# Save cookies in Response.
response.cookies = cookies
# Save original response for later.
response.raw = resp
- response.url = self.full_url.decode('utf-8')
+ response.url = self.full_url
return response
returns it twice.
"""
- if hasattr(data, '__iter__'):
+ if hasattr(data, '__iter__') and not isinstance(data, str):
data = dict(data)
+
if hasattr(data, 'items'):
result = []
- for k, vs in data.items():
+ for k, vs in list(data.items()):
for v in isinstance(vs, list) and vs or [vs]:
- result.append((k.encode('utf-8') if isinstance(k, unicode) else k,
- v.encode('utf-8') if isinstance(v, unicode) else v))
- return result, urllib.urlencode(result, doseq=True)
+ result.append((k.encode('utf-8') if isinstance(k, str) else k,
+ v.encode('utf-8') if isinstance(v, str) else v))
+ return result, urlencode(result, doseq=True)
else:
return data, data
if not self.url:
raise URLRequired()
+ url = self.url
+
# Support for unicode domain names and paths.
- scheme, netloc, path, params, query, fragment = urlparse(self.url)
+ scheme, netloc, path, params, query, fragment = urlparse(url)
+
if not scheme:
- raise ValueError("Invalid URL %r: No schema supplied" %self.url)
+ raise ValueError("Invalid URL %r: No schema supplied" % url)
- netloc = netloc.encode('idna')
+ netloc = netloc.encode('idna').decode('utf-8')
- if isinstance(path, unicode):
- path = path.encode('utf-8')
+ # if isinstance(path, str):
+ # path = path.encode('utf-8')
- path = requote_path(path)
+ # path = requote_path(path)
- url = str(urlunparse([ scheme, netloc, path, params, query, fragment ]))
+ # print([ scheme, netloc, path, params, query, fragment ])
+ # print('---------------------')
+
+ url = (urlunparse([ scheme, netloc, path, params, query, fragment ]))
if self._enc_params:
if urlparse(url).query:
path = p.path
if not path:
path = '/'
+
+ from urllib.parse import quote, unquote
+
+ path = quote(path.encode('utf-8'))
url.append(path)
query = p.query
url.append('?')
url.append(query)
+ # print(url)
+
return ''.join(url)
# Multi-part file uploads.
if self.files:
- if not isinstance(self.data, basestring):
+ if not isinstance(self.data, str):
try:
fields = self.data.copy()
except AttributeError:
fields = dict(self.data)
- for (k, v) in self.files.items():
+ for (k, v) in list(self.files.items()):
# support for explicit filename
if isinstance(v, (tuple, list)):
fn, fp = v
if self.data:
body = self._enc_data
- if isinstance(self.data, basestring):
+ if isinstance(self.data, str):
content_type = None
else:
content_type = 'application/x-www-form-urlencoded'
if 'cookie' not in self.headers:
# Simple cookie with our dict.
- c = oreos.monkeys.SimpleCookie()
- for (k, v) in self.cookies.items():
+ # c = oreos.monkeys.SimpleCookie()
+ from http.cookies import SimpleCookie
+ c = SimpleCookie()
+ for (k, v) in list(self.cookies.items()):
c[k] = v
# Turn it into a header.
)
self.sent = True
- except MaxRetryError, e:
+ except MaxRetryError as e:
raise ConnectionError(e)
- except (_SSLError, _HTTPError), e:
+ except (_SSLError, _HTTPError) as e:
if self.verify and isinstance(e, _SSLError):
raise SSLError(e)
raise Timeout('Request timed out.')
- except RequestException, e:
+ except RequestException as e:
if self.config.get('safe_mode', False):
# In safe mode, catch the exception and attach it to
# a blank urllib3.HTTPResponse object.
def __repr__(self):
return '<Response [%s]>' % (self.status_code)
- def __nonzero__(self):
+ def __bool__(self):
"""Returns true if :attr:`status_code` is 'OK'."""
return self.ok
# Decode unicode from given encoding.
try:
- content = unicode(self.content, encoding)
- except UnicodeError, TypeError:
+ content = str(self.content, encoding)
+ except (UnicodeError, TypeError):
pass
# Try to fall back:
if not content:
try:
- content = unicode(content, encoding, errors='replace')
- except UnicodeError, TypeError:
+ content = str(content, encoding, errors='replace')
+ except (UnicodeError, TypeError):
pass
'\375' : '\\375', '\376' : '\\376', '\377' : '\\377'
}
-_idmap = ''.join(chr(x) for x in xrange(256))
+_idmap = ''.join(chr(x) for x in range(256))
def _quote(str, LegalChars=_LegalChars,
idmap=_idmap, translate=string.translate):
from http.client import HTTPConnection, HTTPSConnection, HTTPException
from http.client import HTTP_PORT, HTTPS_PORT
except ImportError:
- from httplib import HTTPConnection, HTTPSConnection, HTTPException
- from httplib import HTTP_PORT, HTTPS_PORT
+ from http.client import HTTPConnection, HTTPSConnection, HTTPException
+ from http.client import HTTP_PORT, HTTPS_PORT
try: # Python 3
from queue import Queue, Empty, Full
except ImportError:
- from Queue import Queue, Empty, Full
+ from queue import Queue, Empty, Full
try: # Compiled with SSL?
import ssl
EmptyPoolError,
)
-from urllib3.packages.ssl_match_hostname import match_hostname, CertificateError
-from urllib3.packages import six
+from .packages.ssl_match_hostname import match_hostname, CertificateError
+from .packages import six
xrange = six.moves.xrange
self.headers = headers or {}
# Fill the queue up so that doing get() on it will block properly
- for _ in xrange(maxsize):
+ for _ in range(maxsize):
self.pool.put(None)
# These are mostly for testing and debugging purposes.
try:
from http.client import HTTPSConnection
except ImportError:
- from httplib import HTTPSConnection
+ from http.client import HTTPSConnection
from logging import getLogger
from ntlm import ntlm
-from __future__ import absolute_import
+
from . import ssl_match_hostname
MAXSIZE = sys.maxsize
else:
- string_types = basestring,
- integer_types = (int, long)
- class_types = (type, types.ClassType)
- text_type = unicode
+ string_types = str,
+ integer_types = (int, int)
+ class_types = (type, type)
+ text_type = str
binary_type = str
# It's possible to have sizeof(long) != sizeof(Py_ssize_t).
return any("__call__" in klass.__dict__ for klass in type(obj).__mro__)
else:
def get_unbound_function(unbound):
- return unbound.im_func
+ return unbound.__func__
def advance_iterator(it):
- return it.next()
+ return next(it)
callable = callable
_add_doc(get_unbound_function,
def b(s):
return s
def u(s):
- return unicode(s, "unicode_escape")
+ return str(s, "unicode_escape")
int2byte = chr
- import StringIO
- StringIO = BytesIO = StringIO.StringIO
+ import io
+ StringIO = BytesIO = io.StringIO
_add_doc(b, """Byte literal""")
_add_doc(u, """Text literal""")
if fp is None:
return
def write(data):
- if not isinstance(data, basestring):
+ if not isinstance(data, str):
data = str(data)
fp.write(data)
want_unicode = False
sep = kwargs.pop("sep", None)
if sep is not None:
- if isinstance(sep, unicode):
+ if isinstance(sep, str):
want_unicode = True
elif not isinstance(sep, str):
raise TypeError("sep must be None or a string")
end = kwargs.pop("end", None)
if end is not None:
- if isinstance(end, unicode):
+ if isinstance(end, str):
want_unicode = True
elif not isinstance(end, str):
raise TypeError("end must be None or a string")
raise TypeError("invalid keyword arguments to print()")
if not want_unicode:
for arg in args:
- if isinstance(arg, unicode):
+ if isinstance(arg, str):
want_unicode = True
break
if want_unicode:
- newline = unicode("\n")
- space = unicode(" ")
+ newline = str("\n")
+ space = str(" ")
else:
newline = "\n"
space = " "
try:
from urllib.parse import urlencode
except ImportError:
- from urllib import urlencode
+ from urllib.parse import urlencode
from .filepost import encode_multipart_formdata
self.strict = strict
self._decode_content = decode_content
- self._body = body if body and isinstance(body, basestring) else None
+ self._body = body if body and isinstance(body, str) else None
self._fp = None
self._original_response = original_response
if default_kwarg is None:
return local_kwarg
- if isinstance(local_kwarg, basestring):
+ if isinstance(local_kwarg, str):
return local_kwarg
if local_kwarg is None:
kwargs.update(local_kwarg)
# Remove keys that are set to None.
- for (k,v) in local_kwarg.items():
+ for (k,v) in list(local_kwarg.items()):
if v is None:
del kwargs[k]
self.config = config or {}
self.verify = verify
- for (k, v) in defaults.items():
+ for (k, v) in list(defaults.items()):
self.config.setdefault(k, v)
self.poolmanager = PoolManager(
verify = self.verify
# use session's hooks as defaults
- for key, cb in self.hooks.iteritems():
+ for key, cb in list(self.hooks.items()):
hooks.setdefault(key, cb)
# Expand header values.
if headers:
- for k, v in headers.items() or {}:
+ for k, v in list(headers.items()) or {}:
headers[k] = header_expand(v)
args = dict(
codes = LookupDict(name='status_codes')
-for (code, titles) in _codes.items():
+for (code, titles) in list(_codes.items()):
for title in titles:
setattr(codes, title, code)
if not title.startswith('\\'):
@property
def lower_keys(self):
if not hasattr(self, '_lower_keys') or not self._lower_keys:
- self._lower_keys = dict((k.lower(), k) for k in self.iterkeys())
+ self._lower_keys = dict((k.lower(), k) for k in list(self.keys()))
return self._lower_keys
def _clear_lower_keys(self):
return self.__dict__.get(key, None)
def get(self, key, default=None):
- return self.__dict__.get(key, default)
\ No newline at end of file
+ return self.__dict__.get(key, default)
import cgi
import codecs
-import cookielib
+# import cookielib
+from http import cookiejar as cookielib
import os
import random
import re
import zlib
-import urllib
+import urllib.request, urllib.parse, urllib.error
-from urllib2 import parse_http_list as _parse_list_header
+from urllib.parse import quote, unquote
+
+# from urllib2 import parse_http_list as _parse_list_header
+from urllib.request import parse_http_list as _parse_list_header
+
+from http.cookies import SimpleCookie
+
+def dict_from_string(s):
+ """Returns a MultiDict with Cookies."""
+
+ cookies = dict()
+
+ c = SimpleCookie()
+ c.load(s)
+
+ for k,v in list(c.items()):
+ cookies.update({k: v.value})
+
+ return cookies
def guess_filename(obj):
"""Tries to guess the filename of the given object."""
collector = []
if isinstance(headers, dict):
- headers = headers.items()
+ headers = list(headers.items())
- elif isinstance(headers, basestring):
+ elif isinstance(headers, str):
return headers
for i, (value, params) in enumerate(headers):
_params = []
- for (p_k, p_v) in params.items():
+ for (p_k, p_v) in list(params.items()):
_params.append('%s=%s' % (p_k, p_v))
def randombytes(n):
"""Return n random bytes."""
- # Use /dev/urandom if it is available. Fall back to random module
- # if not. It might be worthwhile to extend this function to use
- # other platform-specific mechanisms for getting random bytes.
- if os.path.exists("/dev/urandom"):
- f = open("/dev/urandom")
- s = f.read(n)
- f.close()
- return s
- else:
- L = [chr(random.randrange(0, 256)) for i in range(n)]
- return "".join(L)
+ L = [chr(random.randrange(0, 256)).encode('utf-8') for i in range(n)]
+ return b"".join(L)
def dict_from_cookiejar(cj):
cookie_dict = {}
- for _, cookies in cj._cookies.items():
- for _, cookies in cookies.items():
- for cookie in cookies.values():
+ for _, cookies in list(cj._cookies.items()):
+ for _, cookies in list(cookies.items()):
+ for cookie in list(cookies.values()):
# print cookie
cookie_dict[cookie.name] = cookie.value
:param cookie_dict: Dict of key/values to insert into CookieJar.
"""
- for k, v in cookie_dict.items():
+ for k, v in list(cookie_dict.items()):
cookie = cookielib.Cookie(
version=0,
for encoding in encodings:
try:
- return unicode(content, encoding)
+ return str(content, encoding)
except (UnicodeError, TypeError):
pass
if encoding:
try:
- return unicode(r.content, encoding)
+ return str(r.content, encoding)
except UnicodeError:
tried_encodings.append(encoding)
# Fall back:
try:
- return unicode(r.content, encoding, errors='replace')
+ return str(r.content, encoding, errors='replace')
except TypeError:
return r.content
This function passes the given path through an unquote/quote cycle to
ensure that it is fully and consistently quoted.
"""
- parts = path.split("/")
- parts = (urllib.quote(urllib.unquote(part), safe="") for part in parts)
- return "/".join(parts)
+ parts = path.split(b"/")
+ parts = (quote(unquote(part), safe=b"") for part in parts)
+ return b"/".join(parts)
-from __future__ import with_statement
-import StringIO
+
+import io
import time
import os
import sys
import unittest
import requests
-import envoy
+# import envoy
from requests import HTTPError
from requests.auth import HTTPBasicAuth, HTTPDigestAuth
global _httpbin
if (not 'HTTPBIN_URL' in os.environ) and not _httpbin:
- c = envoy.connect('httpbin %s' % (PORT))
- time.sleep(1)
+ # c = envoy.connect('httpbin %s' % (PORT))
+ # time.sleep(1)
_httpbin = True
r = requests.get(httpbin('user-agent'), headers=heads)
- assert heads['User-agent'] in r.content
+ assert heads['User-agent'] in r.text
self.assertEqual(r.status_code, 200)
}
r = requests.get(httpbin('user-agent'), headers=heads);
- self.assertTrue(heads['User-agent'] in r.content)
+ self.assertTrue(heads['User-agent'] in r.text)
heads = {
'user-agent':
}
r = requests.get(httpbin('user-agent'), headers=heads);
- self.assertTrue(heads['user-agent'] in r.content)
+ self.assertTrue(heads['user-agent'] in r.text)
def test_HTTP_200_OK_HEAD(self):
response = requests.get(url)
- self.assertIsInstance(response.url, unicode)
+ self.assertIsInstance(response.url, str)
def test_unicode_get(self):
url = service('/get')
- requests.get(url, params={'foo': u'føø'})
- requests.get(url, params={u'føø': u'føø'})
+ requests.get(url, params={'foo': 'føø'})
+ requests.get(url, params={'føø': 'føø'})
requests.get(url, params={'føø': 'føø'})
- requests.get(url, params={'foo': u'foo'})
- requests.get(service('ø'), params={'foo': u'foo'})
+ requests.get(url, params={'foo': 'foo'})
+ # print('\n')
+ # print('\n')
+ # print('\n')
+ # print('\n')
+ # print(service('ø'))
+ # print('\n')
+ # print('\n')
+ # print('\n')
+ # print('\n')
+ # print('\n')
+ requests.get(service('ø'), params={'foo': 'foo'})
def test_httpauth_recursion(self):
for service in SERVICES:
r = requests.get(service('basic-auth', 'user', 'pass'), auth=http_auth)
- self.assertEquals(r.status_code, 401)
+ self.assertEqual(r.status_code, 401)
def test_urlencoded_post_data(self):
r = requests.post(service('post'), data=dict(test='fooaowpeuf'))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
- self.assertEquals(r.url, service('post'))
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.url, service('post'))
- rbody = json.loads(r.content)
+ rbody = json.loads(r.text)
- self.assertEquals(rbody.get('form'), dict(test='fooaowpeuf'))
- self.assertEquals(rbody.get('data'), '')
+ self.assertEqual(rbody.get('form'), dict(test='fooaowpeuf'))
+ self.assertEqual(rbody.get('data'), '')
def test_nonurlencoded_post_data(self):
r = requests.post(service('post'), data='fooaowpeuf')
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
- self.assertEquals(r.url, service('post'))
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.url, service('post'))
- rbody = json.loads(r.content)
+ rbody = json.loads(r.text)
# Body wasn't valid url encoded data, so the server returns None as
# "form" and the raw body as "data".
- self.assertEquals(rbody.get('form'), {})
- self.assertEquals(rbody.get('data'), 'fooaowpeuf')
+ self.assertEqual(rbody.get('form'), {})
+ self.assertEqual(rbody.get('data'), 'fooaowpeuf')
def test_urlencoded_post_querystring(self):
r = requests.post(service('post'), params=dict(test='fooaowpeuf'))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
- self.assertEquals(r.url, service('post?test=fooaowpeuf'))
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.url, service('post?test=fooaowpeuf'))
- rbody = json.loads(r.content)
- self.assertEquals(rbody.get('form'), {}) # No form supplied
- self.assertEquals(rbody.get('data'), '')
+ rbody = json.loads(r.text)
+ self.assertEqual(rbody.get('form'), {}) # No form supplied
+ self.assertEqual(rbody.get('data'), '')
def test_urlencoded_post_query_and_data(self):
params=dict(test='fooaowpeuf'),
data=dict(test2="foobar"))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
- self.assertEquals(r.url, service('post?test=fooaowpeuf'))
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.url, service('post?test=fooaowpeuf'))
- rbody = json.loads(r.content)
- self.assertEquals(rbody.get('form'), dict(test2='foobar'))
- self.assertEquals(rbody.get('data'), '')
+ rbody = json.loads(r.text)
+ self.assertEqual(rbody.get('form'), dict(test2='foobar'))
+ self.assertEqual(rbody.get('data'), '')
def test_nonurlencoded_postdata(self):
r = requests.post(service('post'), data="foobar")
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
- rbody = json.loads(r.content)
+ rbody = json.loads(r.text)
- self.assertEquals(rbody.get('form'), {})
- self.assertEquals(rbody.get('data'), 'foobar')
+ self.assertEqual(rbody.get('form'), {})
+ self.assertEqual(rbody.get('data'), 'foobar')
# def test_idna(self):
for service in SERVICES:
r = requests.get(service('get'), params=dict(test=['foo','baz']))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.url, service('get?test=foo&test=baz'))
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.url, service('get?test=foo&test=baz'))
def test_urlencoded_post_querystring_multivalued(self):
for service in SERVICES:
r = requests.post(service('post'), params=dict(test=['foo','baz']))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
- self.assertEquals(r.url, service('post?test=foo&test=baz'))
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.url, service('post?test=foo&test=baz'))
- rbody = json.loads(r.content)
- self.assertEquals(rbody.get('form'), {}) # No form supplied
- self.assertEquals(rbody.get('data'), '')
+ rbody = json.loads(r.text)
+ self.assertEqual(rbody.get('form'), {}) # No form supplied
+ self.assertEqual(rbody.get('data'), '')
def test_urlencoded_post_query_multivalued_and_data(self):
params=dict(test=['foo','baz']),
data=dict(test2="foobar",test3=['foo','baz']))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(r.headers['content-type'], 'application/json')
- self.assertEquals(r.url, service('post?test=foo&test=baz'))
- rbody = json.loads(r.content)
- self.assertEquals(rbody.get('form'), dict(test2='foobar',test3=['foo','baz']))
- self.assertEquals(rbody.get('data'), '')
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(r.headers['content-type'], 'application/json')
+ self.assertEqual(r.url, service('post?test=foo&test=baz'))
+
+ # print(r.text)
+ # print('-----------------------')
+
+ rbody = json.loads(r.text)
+ self.assertEqual(rbody.get('form'), dict(test2='foobar',test3=['foo','baz']))
+ self.assertEqual(rbody.get('data'), '')
def test_GET_no_redirect(self):
for service in SERVICES:
r = requests.get(service('redirect', '3'), allow_redirects=False)
- self.assertEquals(r.status_code, 302)
- self.assertEquals(len(r.history), 0)
+ self.assertEqual(r.status_code, 302)
+ self.assertEqual(len(r.history), 0)
def test_HEAD_no_redirect(self):
for service in SERVICES:
r = requests.head(service('redirect', '3'), allow_redirects=False)
- self.assertEquals(r.status_code, 302)
- self.assertEquals(len(r.history), 0)
+ self.assertEqual(r.status_code, 302)
+ self.assertEqual(len(r.history), 0)
def test_redirect_history(self):
for service in SERVICES:
r = requests.get(service('redirect', '3'))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(len(r.history), 3)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(len(r.history), 3)
def test_relative_redirect_history(self):
for service in SERVICES:
r = requests.get(service('relative-redirect', '3'))
- self.assertEquals(r.status_code, 200)
- self.assertEquals(len(r.history), 3)
+ self.assertEqual(r.status_code, 200)
+ self.assertEqual(len(r.history), 3)
def test_session_HTTP_200_OK_GET(self):
# Make 2 requests from Session object, should send header both times
r1 = s.get(httpbin('user-agent'))
- assert heads['User-agent'] in r1.content
+ assert heads['User-agent'] in r1.text
r2 = s.get(httpbin('user-agent'))
- assert heads['User-agent'] in r2.content
+ assert heads['User-agent'] in r2.text
new_heads = {'User-agent': 'blah'}
r3 = s.get(httpbin('user-agent'), headers=new_heads)
- assert new_heads['User-agent'] in r3.content
+ assert new_heads['User-agent'] in r3.text
self.assertEqual(r2.status_code, 200)
}
)
- assert 'foo' in response.content
+ assert 'foo' in response.text
def test_multiple_hooks(self):
}
)
- assert 'foo' in response.content
- assert 'bar' in response.content
+ assert 'foo' in response.text
+ assert 'bar' in response.text
def test_session_persistent_cookies(self):
r = s.get(httpbin('cookies'))
# Those cookies persist transparently.
- c = json.loads(r.content).get('cookies')
+ c = json.loads(r.text).get('cookies')
assert c == _c
# Double check.
r = s.get(httpbin('cookies'), cookies={})
- c = json.loads(r.content).get('cookies')
+ c = json.loads(r.text).get('cookies')
assert c == _c
# Remove a cookie by setting it's value to None.
r = s.get(httpbin('cookies'), cookies={'bessie': None})
- c = json.loads(r.content).get('cookies')
+ c = json.loads(r.text).get('cookies')
del _c['bessie']
assert c == _c
# Test session-level cookies.
s = requests.session(cookies=_c)
r = s.get(httpbin('cookies'))
- c = json.loads(r.content).get('cookies')
+ c = json.loads(r.text).get('cookies')
assert c == _c
# Have the server set a cookie.
r = s.get(httpbin('cookies', 'set', 'k', 'v'), allow_redirects=True)
- c = json.loads(r.content).get('cookies')
+ c = json.loads(r.text).get('cookies')
assert 'k' in c
# And server-set cookie persistience.
r = s.get(httpbin('cookies'))
- c = json.loads(r.content).get('cookies')
+ c = json.loads(r.text).get('cookies')
assert 'k' in c
# Make 2 requests from Session object, should send header both times
r1 = s.get(httpbin('get'))
- assert params['a'] in r1.content
+ assert params['a'] in r1.text
params2 = {'b': 'b_test'}
r2 = s.get(httpbin('get'), params=params2)
- assert params['a'] in r2.content
- assert params2['b'] in r2.content
+ assert params['a'] in r2.text
+ assert params2['b'] in r2.text
params3 = {'b': 'b_test', 'a': None, 'c': 'c_test'}
r3 = s.get(httpbin('get'), params=params3)
- assert not params['a'] in r3.content
- assert params3['b'] in r3.content
- assert params3['c'] in r3.content
+ assert not params['a'] in r3.text
+ assert params3['b'] in r3.text
+ assert params3['c'] in r3.text
def test_invalid_content(self):
# WARNING: if you're using a terrible DNS provider (comcast),
def test_cached_response(self):
r1 = requests.get(httpbin('get'), prefetch=False)
+ assert not r1._content
assert r1.content
- assert r1.content
+ assert r1.text
r2 = requests.get(httpbin('get'), prefetch=True)
assert r2._content
assert r2.content
+ assert r2.text
def test_iter_lines(self):
# Make a request and monkey-patch its contents
r = requests.get(httpbin('get'))
- r.raw = StringIO.StringIO(quote)
+ r.raw = io.StringIO(quote)
# Make sure iter_lines doesn't chop the trailing bit
lines = '\n'.join(r.iter_lines())
# Safe mode creates empty responses for failed requests.
# Iterating on these responses should produce empty sequences
r = safe.get('http://_/')
- self.assertEquals(list(r.iter_lines()), [])
+ self.assertEqual(list(r.iter_lines()), [])
self.assertIsInstance(r.error, requests.exceptions.ConnectionError)
r = safe.get('http://_/')
- self.assertEquals(list(r.iter_content()), [])
+ self.assertEqual(list(r.iter_content()), [])
self.assertIsInstance(r.error, requests.exceptions.ConnectionError)
# When not in safe mode, should raise Timeout exception
projects / services / python-requests.git / commitdiff