kennethreitz/requests
1
0
Fork 0
mirror of https://github.com/kennethreitz/requests.git synced 2026-06-05 22:50:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'develop'

Browse Source
This commit is contained in:
Kenneth Reitz
2011-10-23 15:21:42 -04:00
parent 5b5b3def00 02cf3dce37
commit 6b9cb059f6
12 changed files with 396 additions and 347 deletions
Download Patch File Download Diff File Expand all files Collapse all files
HISTORY.rst
+8
View File
@@ -1,6 +1,14 @@
History
-------
0.7.1 (2011-10-23)
++++++++++++++++++
* Move away from urllib2 authentication handling.
* Fully Remove AuthManager, AuthObject, &c.
* New tuple-based auth system with handler callbacks.
0.7.0 (2011-10-22)
++++++++++++++++++
docs/user/advanced.rst
+33
View File
@@ -157,6 +157,39 @@ And give it a try::
}
Custom Authentication
---------------------
Requests allows you to use specify your own authentication mechanism.
When you pass our authentication tuple to a request method, the first
string is the type of authentication. 'basic' is inferred if none is
provided.
You can pass in a callable object instead of a string for the first item
in the tuple, and it will be used in place of the built in authentication
callbacks.
Let's pretend that we have a web service that will only respond if the
``X-Pizza`` header is set to a password value. Unlikely, but just go with it.
We simply need to define a callback function that will be used to update the
Request object, right before it is dispatched.
::
def pizza_auth(r, username):
"""Attaches HTTP Pizza Authentication to the given Request object.
"""
r.headers['X-Pizza'] = username
return r
Then, we can make a request using our Pizza Auth::
>>> requests.get('http://pizzabin.org/admin', auth=(pizza_auth, 'kenneth'))
<Response [200]>
Verbose Logging
---------------
docs/user/quickstart.rst
+29
View File
@@ -136,3 +136,32 @@ parameter::
>>> r = requests.get(url, cookies=cookies)
>>> r.content
'{"cookies": {"cookies_are": "working"}}'
Basic Authentication
--------------------
Most web services require authentication. There many different types of
authentication, but the most common is called HTTP Basic Auth.
Making requests with Basic Auth is easy, with Requests::
>>> requests.get('https://api.github.com/user', auth=('user', 'pass'))
<Response [200]>
Digest Authentication
---------------------
Another popular form of protecting web service is Digest Authentication.
Requests supports it!::
>>> url = 'http://httpbin.org/digest-auth/auth/user/pass'
>>> requests.get(url, auth=('digest', 'user', 'pass'))
<Response [200]>
-----------------------
Ready for more? Check out the advanced_ section.
requests/__init__.py
+2 -2
View File
@@ -15,8 +15,8 @@ requests
"""
__title__ = 'requests'
__version__ = '0.7.0'
__build__ = 0x000700
__version__ = '0.7.1'
__build__ = 0x000701
__author__ = 'Kenneth Reitz'
__license__ = 'ISC'
__copyright__ = 'Copyright 2011 Kenneth Reitz'
requests/async.py
+10 -16
View File
@@ -28,7 +28,7 @@ __all__ = (
)
def _patched(f):
def patched(f):
"""Patches a given API function to not send."""
def wrapped(*args, **kwargs):
@@ -37,7 +37,7 @@ def _patched(f):
return wrapped
def _send(r, pools=None):
def send(r, pools=None):
"""Sends a given Request object."""
if pools:
@@ -45,23 +45,17 @@ def _send(r, pools=None):
r.send()
# Post-request hook.
r = dispatch_hook('post_request', r.hooks, r)
# Response manipulation hook.
r.response = dispatch_hook('response', r.hooks, r.response)
return r.response
# Patched requests.api functions.
get = _patched(api.get)
head = _patched(api.head)
post = _patched(api.post)
put = _patched(api.put)
patch = _patched(api.patch)
delete = _patched(api.delete)
request = _patched(api.request)
get = patched(api.get)
head = patched(api.head)
post = patched(api.post)
put = patched(api.put)
patch = patched(api.patch)
delete = patched(api.delete)
request = patched(api.request)
def map(requests, prefetch=True):
@@ -71,7 +65,7 @@ def map(requests, prefetch=True):
:param prefetch: If False, the content will not be downloaded immediately.
"""
jobs = [gevent.spawn(_send, r) for r in requests]
jobs = [gevent.spawn(send, r) for r in requests]
gevent.joinall(jobs)
if prefetch:
requests/auth.py
+151
View File
@@ -0,0 +1,151 @@
# -*- coding: utf-8 -*-
"""
requests.auth
~~~~~~~~~~~~~
This module contains the authentication handlers for Requests.
"""
import time
import hashlib
from base64 import b64encode
from urlparse import urlparse
from .utils import randombytes, parse_dict_header
def http_basic(r, username, password):
"""Attaches HTTP Basic Authentication to the given Request object.
Arguments should be considered non-positional.
"""
username = str(username)
password = str(password)
auth_s = b64encode('%s:%s' % (username, password))
r.headers['Authorization'] = ('Basic %s' % auth_s)
return r
def http_digest(r, username, password):
"""Attaches HTTP Digest Authentication to the given Request object.
Arguments should be considered non-positional.
"""
def handle_401(r):
"""Takes the given response and tries digest-auth, if needed."""
s_auth = r.headers.get('www-authenticate', '')
if 'digest' in s_auth.lower():
last_nonce = ''
nonce_count = 0
chal = parse_dict_header(s_auth.replace('Digest ', ''))
realm = chal['realm']
nonce = chal['nonce']
qop = chal.get('qop')
algorithm = chal.get('algorithm', 'MD5')
opaque = chal.get('opaque', None)
algorithm = algorithm.upper()
# lambdas assume digest modules are imported at the top level
if algorithm == 'MD5':
H = lambda x: hashlib.md5(x).hexdigest()
elif algorithm == 'SHA':
H = lambda x: hashlib.sha1(x).hexdigest()
# XXX MD5-sess
KD = lambda s, d: H("%s:%s" % (s, d))
if H is None:
return None
# XXX not implemented yet
entdig = None
path = urlparse(r.request.url).path
A1 = "%s:%s:%s" % (username, realm, password)
A2 = "%s:%s" % (r.request.method, path)
if qop == 'auth':
if nonce == last_nonce:
nonce_count += 1
else:
nonce_count = 1
last_nonce = nonce
ncvalue = '%08x' % nonce_count
cnonce = (hashlib.sha1("%s:%s:%s:%s" % (
nonce_count, nonce, time.ctime(), randombytes(8)))
.hexdigest()[:16]
)
noncebit = "%s:%s:%s:%s:%s" % (nonce, ncvalue, cnonce, qop, H(A2))
respdig = KD(H(A1), noncebit)
elif qop is None:
respdig = KD(H(A1), "%s:%s" % (nonce, H(A2)))
else:
# XXX handle auth-int.
return None
# XXX should the partial digests be encoded too?
base = 'username="%s", realm="%s", nonce="%s", uri="%s", ' \
'response="%s"' % (username, realm, nonce, path, respdig)
if opaque:
base += ', opaque="%s"' % opaque
if entdig:
base += ', digest="%s"' % entdig
base += ', algorithm="%s"' % algorithm
if qop:
base += ', qop=auth, nc=%s, cnonce="%s"' % (ncvalue, cnonce)
r.request.headers['Authorization'] = 'Digest %s' % (base)
r.request.send(anyway=True)
_r = r.request.response
_r.history.append(r)
print _r.status_code
# r.request.response
print locals()
print _r.headers
return _r
return r
r.hooks['response'] = handle_401
return r
def dispatch(t):
"""Given an auth tuple, return an expanded version."""
if not t:
return t
else:
t = list(t)
# Make sure they're passing in something.
assert len(t) >= 2
# If only two items are passed in, assume HTTPBasic.
if (len(t) == 2):
t.insert(0, 'basic')
# Allow built-in string referenced auths.
if isinstance(t[0], basestring):
if t[0] in ('basic', 'forced_basic'):
t[0] = http_basic
elif t[0] in ('digest',):
t[0] = http_digest
# Return a custom callable.
return (t[0], tuple(t[1:]))
requests/defaults.py
+1 -1
View File
@@ -25,7 +25,7 @@ defaults = dict()
defaults['base_headers'] = {
'User-Agent': 'python-requests/%s' % __version__,
'Accept-Encoding': ', '.join([ 'identity', 'deflate', 'compress', 'gzip' ]),
'Accept-Encoding': ', '.join(('identity', 'deflate', 'compress', 'gzip')),
}
defaults['proxies'] = {}
requests/models.py
+23 -202
View File
@@ -16,6 +16,7 @@ from urllib2 import HTTPError
from urlparse import urlparse, urlunparse, urljoin
from datetime import datetime
from .hooks import dispatch_hook
from .structures import CaseInsensitiveDict
from .packages.poster.encode import multipart_encode
from .packages.poster.streaminghttp import register_openers, get_handlers
@@ -23,10 +24,9 @@ from .utils import (dict_from_cookiejar, get_unicode_from_response, stream_decod
from .status_codes import codes
from .exceptions import Timeout, URLRequired, TooManyRedirects
from .monkeys import Request as _Request
from .monkeys import (
HTTPBasicAuthHandler, HTTPForcedBasicAuthHandler,
HTTPDigestAuthHandler, HTTPRedirectHandler)
from .monkeys import HTTPRedirectHandler
from .auth import dispatch as auth_dispatch
REDIRECT_STATI = (codes.moved, codes.found, codes.other, codes.temporary_moved)
@@ -94,13 +94,8 @@ class Request(object):
#: content and metadata of HTTP Response, once :attr:`sent <send>`.
self.response = Response()
if isinstance(auth, (list, tuple)):
auth = AuthObject(*auth)
if not auth:
auth = auth_manager.get_auth(self.url)
#: :class:`AuthObject` to attach to :class:`Request <Request>`.
self.auth = auth
#: Authentication tuple to attach to :class:`Request <Request>`.
self.auth = auth_dispatch(auth)
#: CookieJar to attach to :class:`Request <Request>`.
self.cookies = cookies
@@ -125,6 +120,10 @@ class Request(object):
self.headers = headers
# Pre-request hook.
r = dispatch_hook('pre_request', hooks, self)
self.__dict__.update(r.__dict__)
def __repr__(self):
return '<Request [%s]>' % (self.method)
@@ -138,22 +137,6 @@ class Request(object):
if self.cookies is not None:
_handlers.append(urllib2.HTTPCookieProcessor(self.cookies))
if self.auth:
if not isinstance(self.auth.handler,
(urllib2.AbstractBasicAuthHandler,
urllib2.AbstractDigestAuthHandler)):
# TODO: REMOVE THIS COMPLETELY
auth_manager.add_password(
self.auth.realm, self.url,
self.auth.username,
self.auth.password)
self.auth.handler = self.auth.handler(auth_manager)
auth_manager.add_auth(self.url, self.auth)
_handlers.append(self.auth.handler)
if self.proxies:
_handlers.append(urllib2.ProxyHandler(self.proxies))
@@ -348,6 +331,13 @@ class Request(object):
data = self._enc_data
headers = {}
if self.auth:
auth_func, auth_args = self.auth
r = auth_func(self, *auth_args)
self.__dict__.update(r.__dict__)
# Build the Urllib2 Request.
req = _Request(url, data=data, headers=headers, method=self.method)
@@ -399,6 +389,13 @@ class Request(object):
self.sent = self.response.ok
# Response manipulation hook.
self.response = dispatch_hook('response', self.hooks, self.response)
# Post-request hook.
r = dispatch_hook('post_request', self.hooks, self)
self.__dict__.update(r.__dict__)
return self.sent
@@ -520,179 +517,3 @@ class Response(object):
if self.error:
raise self.error
class AuthManager(object):
"""Requests Authentication Manager."""
def __new__(cls):
singleton = cls.__dict__.get('__singleton__')
if singleton is not None:
return singleton
cls.__singleton__ = singleton = object.__new__(cls)
return singleton
def __init__(self):
self.passwd = {}
self._auth = {}
def __repr__(self):
return '<AuthManager [%s]>' % (self.method)
def add_auth(self, uri, auth):
"""Registers AuthObject to AuthManager."""
uri = self.reduce_uri(uri, False)
# try to make it an AuthObject
if not isinstance(auth, AuthObject):
try:
auth = AuthObject(*auth)
except TypeError:
pass
self._auth[uri] = auth
def add_password(self, realm, uri, user, passwd):
"""Adds password to AuthManager."""
# uri could be a single URI or a sequence
if isinstance(uri, basestring):
uri = [uri]
reduced_uri = tuple([self.reduce_uri(u, False) for u in uri])
if reduced_uri not in self.passwd:
self.passwd[reduced_uri] = {}
self.passwd[reduced_uri] = (user, passwd)
def find_user_password(self, realm, authuri):
for uris, authinfo in self.passwd.iteritems():
reduced_authuri = self.reduce_uri(authuri, False)
for uri in uris:
if self.is_suburi(uri, reduced_authuri):
return authinfo
return (None, None)
def get_auth(self, uri):
(in_domain, in_path) = self.reduce_uri(uri, False)
for domain, path, authority in (
(i[0][0], i[0][1], i[1]) for i in self._auth.iteritems()
):
if in_domain == domain:
if path in in_path:
return authority
def reduce_uri(self, uri, default_port=True):
"""Accept authority or URI and extract only the authority and path."""
# note HTTP URLs do not have a userinfo component
parts = urllib2.urlparse.urlsplit(uri)
if parts[1]:
# URI
scheme = parts[0]
authority = parts[1]
path = parts[2] or '/'
else:
# host or host:port
scheme = None
authority = uri
path = '/'
host, port = urllib2.splitport(authority)
if default_port and port is None and scheme is not None:
dport = {"http": 80,
"https": 443,
}.get(scheme)
if dport is not None:
authority = "%s:%d" % (host, dport)
return authority, path
def is_suburi(self, base, test):
"""Check if test is below base in a URI tree
Both args must be URIs in reduced form.
"""
if base == test:
return True
if base[0] != test[0]:
return False
common = urllib2.posixpath.commonprefix((base[1], test[1]))
if len(common) == len(base[1]):
return True
return False
def empty(self):
self.passwd = {}
def remove(self, uri, realm=None):
# uri could be a single URI or a sequence
if isinstance(uri, basestring):
uri = [uri]
for default_port in True, False:
reduced_uri = tuple([self.reduce_uri(u, default_port) for u in uri])
del self.passwd[reduced_uri][realm]
def __contains__(self, uri):
# uri could be a single URI or a sequence
if isinstance(uri, basestring):
uri = [uri]
uri = tuple([self.reduce_uri(u, False) for u in uri])
if uri in self.passwd:
return True
return False
auth_manager = AuthManager()
class AuthObject(object):
"""The :class:`AuthObject` is a simple HTTP Authentication token. When
given to a Requests function, it enables Basic HTTP Authentication for that
Request. You can also enable Authorization for domain realms with AutoAuth.
See AutoAuth for more details.
:param username: Username to authenticate with.
:param password: Password for given username.
:param realm: (optional) the realm this auth applies to
:param handler: (optional) basic || digest || proxy_basic || proxy_digest
"""
_handlers = {
'basic': HTTPBasicAuthHandler,
'forced_basic': HTTPForcedBasicAuthHandler,
'digest': HTTPDigestAuthHandler,
'proxy_basic': urllib2.ProxyBasicAuthHandler,
'proxy_digest': urllib2.ProxyDigestAuthHandler
}
def __init__(self, username, password, handler='forced_basic', realm=None):
self.username = username
self.password = password
self.realm = realm
if isinstance(handler, basestring):
self.handler = self._handlers.get(handler.lower(), HTTPForcedBasicAuthHandler)
else:
self.handler = handler
requests/monkeys.py
+5 -114
View File
@@ -29,120 +29,11 @@ class Request(urllib2.Request):
class HTTPRedirectHandler(urllib2.HTTPRedirectHandler):
"""HTTP Redirect handler."""
def http_error_301(self, req, fp, code, msg, headers):
def _pass(self, req, fp, code, msg, headers):
pass
http_error_302 = http_error_303 = http_error_307 = http_error_301
http_error_302 = _pass
http_error_303 = _pass
http_error_307 = _pass
http_error_301 = _pass
class HTTPBasicAuthHandler(urllib2.HTTPBasicAuthHandler):
"""HTTP Basic Auth Handler with authentication loop fixes."""
def __init__(self, *args, **kwargs):
urllib2.HTTPBasicAuthHandler.__init__(self, *args, **kwargs)
self.retried_req = None
self.retried = 0
def reset_retry_count(self):
# Python 2.6.5 will call this on 401 or 407 errors and thus loop
# forever. We disable reset_retry_count completely and reset in
# http_error_auth_reqed instead.
pass
def http_error_auth_reqed(self, auth_header, host, req, headers):
# Reset the retry counter once for each request.
if req is not self.retried_req:
self.retried_req = req
self.retried = 0
return urllib2.HTTPBasicAuthHandler.http_error_auth_reqed(
self, auth_header, host, req, headers
)
class HTTPForcedBasicAuthHandler(HTTPBasicAuthHandler):
"""HTTP Basic Auth Handler with forced Authentication."""
auth_header = 'Authorization'
rx = re.compile('(?:.*,)*[ \t]*([^ \t]+)[ \t]+'
'realm=(["\'])(.*?)\\2', re.I)
def __init__(self, *args, **kwargs):
HTTPBasicAuthHandler.__init__(self, *args, **kwargs)
def http_error_401(self, req, fp, code, msg, headers):
url = req.get_full_url()
response = self._http_error_auth_reqed('www-authenticate', url, req, headers)
self.reset_retry_count()
return response
http_error_404 = http_error_401
def _http_error_auth_reqed(self, authreq, host, req, headers):
authreq = headers.get(authreq, None)
if self.retried > 5:
# retry sending the username:password 5 times before failing.
raise urllib2.HTTPError(req.get_full_url(), 401, "basic auth failed",
headers, None)
else:
self.retried += 1
if authreq:
mo = self.rx.search(authreq)
if mo:
scheme, quote, realm = mo.groups()
if scheme.lower() == 'basic':
response = self.retry_http_basic_auth(host, req, realm)
if response and response.code not in (401, 404):
self.retried = 0
return response
else:
response = self.retry_http_basic_auth(host, req, 'Realm')
if response and response.code not in (401, 404):
self.retried = 0
return response
class HTTPDigestAuthHandler(urllib2.HTTPDigestAuthHandler):
def __init__(self, *args, **kwargs):
urllib2.HTTPDigestAuthHandler.__init__(self, *args, **kwargs)
self.retried_req = None
def reset_retry_count(self):
# Python 2.6.5 will call this on 401 or 407 errors and thus loop
# forever. We disable reset_retry_count completely and reset in
# http_error_auth_reqed instead.
pass
def http_error_auth_reqed(self, auth_header, host, req, headers):
# Reset the retry counter once for each request.
if req is not self.retried_req:
self.retried_req = req
self.retried = 0
# In python < 2.5 AbstractDigestAuthHandler raises a ValueError if
# it doesn't know about the auth type requested. This can happen if
# somebody is using BasicAuth and types a bad password.
try:
return urllib2.HTTPDigestAuthHandler.http_error_auth_reqed(
self, auth_header, host, req, headers)
except ValueError, inst:
arg = inst.args[0]
if arg.startswith("AbstractDigestAuthHandler doesn't know "):
return
raise
requests/sessions.py
+1 -9
View File
@@ -112,7 +112,7 @@ class Session(object):
:param headers: (optional) Dictionary of HTTP Headers to send with the :class:`Request`.
:param cookies: (optional) Dict or CookieJar object to send with the :class:`Request`.
:param files: (optional) Dictionary of 'filename': file-like-objects for multipart encoding upload.
:param auth: (optional) AuthObject to enable Basic HTTP Auth.
:param auth: (optional) Auth typle to enable Basic/Digest/Custom HTTP Auth.
:param timeout: (optional) Float describing the timeout of the request.
:param allow_redirects: (optional) Boolean. Set to True if POST/PUT/DELETE redirect following is allowed.
:param proxies: (optional) Dictionary mapping protocol to the URL of the proxy.
@@ -162,9 +162,6 @@ class Session(object):
r = Request(**args)
# Pre-request hook.
r = dispatch_hook('pre_request', hooks, r)
# Don't send if asked nicely.
if not return_response:
return r
@@ -172,11 +169,6 @@ class Session(object):
# Send the HTTP Request.
r.send()
# Post-request hook.
r = dispatch_hook('post_request', hooks, r)
# Response manipulation hook.
r.response = dispatch_hook('response', hooks, r.response)
return r.response
requests/utils.py
+109
View File
@@ -12,9 +12,103 @@ that are also useful for external consumption.
import cgi
import codecs
import cookielib
import os
import random
import re
import zlib
from urllib2 import parse_http_list as _parse_list_header
# From mitsuhiko/werkzeug (used with permission).
def parse_list_header(value):
"""Parse lists as described by RFC 2068 Section 2.
In particular, parse comma-separated lists where the elements of
the list may include quoted-strings. A quoted-string could
contain a comma. A non-quoted string could have quotes in the
middle. Quotes are removed automatically after parsing.
It basically works like :func:`parse_set_header` just that items
may appear multiple times and case sensitivity is preserved.
The return value is a standard :class:`list`:
>>> parse_list_header('token, "quoted value"')
['token', 'quoted value']
To create a header from the :class:`list` again, use the
:func:`dump_header` function.
:param value: a string with a list header.
:return: :class:`list`
"""
result = []
for item in _parse_list_header(value):
if item[:1] == item[-1:] == '"':
item = unquote_header_value(item[1:-1])
result.append(item)
return result
# From mitsuhiko/werkzeug (used with permission).
def parse_dict_header(value):
"""Parse lists of key, value pairs as described by RFC 2068 Section 2 and
convert them into a python dict:
>>> d = parse_dict_header('foo="is a fish", bar="as well"')
>>> type(d) is dict
True
>>> sorted(d.items())
[('bar', 'as well'), ('foo', 'is a fish')]
If there is no value for a key it will be `None`:
>>> parse_dict_header('key_without_value')
{'key_without_value': None}
To create a header from the :class:`dict` again, use the
:func:`dump_header` function.
:param value: a string with a dict header.
:return: :class:`dict`
"""
result = {}
for item in _parse_list_header(value):
if '=' not in item:
result[item] = None
continue
name, value = item.split('=', 1)
if value[:1] == value[-1:] == '"':
value = unquote_header_value(value[1:-1])
result[name] = value
return result
# From mitsuhiko/werkzeug (used with permission).
def unquote_header_value(value, is_filename=False):
r"""Unquotes a header value. (Reversal of :func:`quote_header_value`).
This does not use the real unquoting but what browsers are actually
using for quoting.
:param value: the header value to unquote.
"""
if value and value[0] == value[-1] == '"':
# this is not the real unquoting, but fixing this so that the
# RFC is met will result in bugs with internet explorer and
# probably some other browsers as well. IE for example is
# uploading files with "C:\foo\bar.txt" as filename
value = value[1:-1]
# if this is a filename and the starting characters look like
# a UNC path, then just return the value without quotes. Using the
# replace sequence below on a UNC path has the effect of turning
# the leading double slash into a single slash and then
# _fix_ie_filename() doesn't work correctly. See #458.
if not is_filename or value[:2] != '\\\\':
return value.replace('\\\\', '\\').replace('\\"', '"')
return value
def header_expand(headers):
"""Returns an HTTP Header value string from a dictionary.
@@ -63,6 +157,21 @@ def header_expand(headers):
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)
def dict_from_cookiejar(cj):
"""Returns a key/value dictionary from a CookieJar.
test_requests.py
+24 -3
View File
@@ -139,7 +139,7 @@ class RequestsTestSuite(unittest.TestCase):
self.assertEqual(r.status_code, 200)
def test_AUTH_HTTP_200_OK_GET(self):
def test_BASICAUTH_HTTP_200_OK_GET(self):
for service in SERVICES:
@@ -147,14 +147,35 @@ class RequestsTestSuite(unittest.TestCase):
url = service('basic-auth', 'user', 'pass')
r = requests.get(url, auth=auth)
# print r.__dict__
self.assertEqual(r.status_code, 200)
r = requests.get(url)
self.assertEqual(r.status_code, 401)
s = requests.session(auth=auth)
r = s.get(url)
self.assertEqual(r.status_code, 200)
def test_DIGESTAUTH_HTTP_200_OK_GET(self):
for service in SERVICES:
auth = ('digest', 'user', 'pass')
url = service('digest-auth', 'auth', 'user', 'pass')
r = requests.get(url, auth=auth)
self.assertEqual(r.status_code, 200)
r = requests.get(url)
self.assertEqual(r.status_code, 401)
s = requests.session(auth=auth)
r = s.get(url)
self.assertEqual(r.status_code, 200)
def test_POSTBIN_GET_POST_FILES(self):
for service in SERVICES: