diff --git a/mechanize/_urllib2_fork.py b/mechanize/_urllib2_fork.py
index d0cfe38..c53b46d 100644
--- a/mechanize/_urllib2_fork.py
+++ b/mechanize/_urllib2_fork.py
@@ -1109,7 +1109,13 @@ def do_open(self, http_class, req):
set_tunnel = h._set_tunnel
else:
set_tunnel = h.set_tunnel
- set_tunnel(req._tunnel_host)
+ tunnel_headers = {}
+ proxy_auth_hdr = "Proxy-Authorization"
+ if proxy_auth_hdr in headers:
+ tunnel_headers[proxy_auth_hdr] = headers[proxy_auth_hdr]
+ # Proxy-Authorization should not be sent to origin server.
+ del headers[proxy_auth_hdr]
+ set_tunnel(req._tunnel_host, headers=tunnel_headers)
try:
h.request(req.get_method(), req.get_selector(), req.data, headers)
diff --git a/test/test_urllib2.py b/test/test_urllib2.py
index abe2624..7532ecf 100644
--- a/test/test_urllib2.py
+++ b/test/test_urllib2.py
@@ -21,9 +21,9 @@
from mechanize._http import parse_head
from mechanize._response import test_response
from mechanize import HTTPRedirectHandler, \
- HTTPEquivProcessor, HTTPRefreshProcessor, \
- HTTPCookieProcessor, HTTPRefererProcessor, \
- HTTPErrorProcessor, HTTPHandler
+ HTTPEquivProcessor, HTTPRefreshProcessor, \
+ HTTPCookieProcessor, HTTPRefererProcessor, \
+ HTTPErrorProcessor, HTTPHandler
from mechanize import OpenerDirector, build_opener, Request
from mechanize._urllib2_fork import AbstractHTTPHandler
from mechanize._util import write_file
@@ -45,7 +45,6 @@ def __cmp__(self, other):
class TrivialTests(mechanize._testcase.TestCase):
-
def test_trivial(self):
# A couple trivial tests
@@ -61,6 +60,7 @@ def test_trivial(self):
fname = '/' + fname.replace(':', '/')
elif os.name == 'riscos':
import string
+
fname = os.expand(fname)
fname = fname.translate(string.maketrans("/.", "./"))
@@ -110,6 +110,7 @@ def test_request_headers_dict():
"""
+
def test_request_headers_methods():
"""
Note the case normalization of header names here, to .capitalize()-case.
@@ -255,19 +256,26 @@ def test_password_manager_default_port(self):
"""
+
class MockOpener:
addheaders = []
+
def open(self, req, data=None,
timeout=_sockettimeout._GLOBAL_DEFAULT_TIMEOUT):
- self.req, self.data, self.timeout = req, data, timeout
+ self.req, self.data, self.timeout = req, data, timeout
+
def error(self, proto, *args):
self.proto, self.args = proto, args
+
class MockFile:
def read(self, count=None): pass
+
def readline(self, count=None): pass
+
def close(self): pass
+
def http_message(mapping):
"""
>>> http_message({"Content-Type": "text/html"}).items()
@@ -281,41 +289,104 @@ def http_message(mapping):
msg = httplib.HTTPMessage(StringIO.StringIO("\r\n".join(f)))
return msg
+
class MockResponse(StringIO.StringIO):
def __init__(self, code, msg, headers, data, url=None):
StringIO.StringIO.__init__(self, data)
self.code, self.msg, self.headers, self.url = code, msg, headers, url
+
def info(self):
return self.headers
+
def geturl(self):
return self.url
+
class MockCookieJar:
def add_cookie_header(self, request, unverifiable=False):
self.ach_req, self.ach_u = request, unverifiable
+
def extract_cookies(self, response, request, unverifiable=False):
self.ec_req, self.ec_r, self.ec_u = request, response, unverifiable
+
class FakeMethod:
def __init__(self, meth_name, action, handle):
self.meth_name = meth_name
self.handle = handle
self.action = action
+
def __call__(self, *args):
return self.handle(self.meth_name, self.action, *args)
+
+class MockHTTPResponse:
+ def __init__(self, fp, msg, status, reason):
+ self.fp = fp
+ self.msg = msg
+ self.status = status
+ self.reason = reason
+
+ def read(self):
+ return ''
+
+
+class MockHTTPClass:
+ def __init__(self):
+ self.req_headers = []
+ self.data = None
+ self.raise_on_endheaders = False
+ self._tunnel_headers = {}
+
+ def __call__(self, host, timeout=_sockettimeout._GLOBAL_DEFAULT_TIMEOUT):
+ self.host = host
+ self.timeout = timeout
+ return self
+
+ def set_debuglevel(self, level):
+ self.level = level
+
+ def set_tunnel(self, host, port=None, headers=None):
+ self._tunnel_host = host
+ self._tunnel_port = port
+ if headers:
+ self._tunnel_headers = headers
+ else:
+ self._tunnel_headers.clear()
+
+ def request(self, method, url, body=None, headers={}):
+ self.method = method
+ self.selector = url
+ self.req_headers += headers.items()
+ self.req_headers.sort()
+ if body:
+ self.data = body
+ if self.raise_on_endheaders:
+ import socket
+
+ raise socket.error()
+
+ def getresponse(self):
+ return MockHTTPResponse(MockFile(), {}, 200, "OK")
+
+
class MockHandler:
# useful for testing handler machinery
# see add_ordered_mock_handlers() docstring
handler_order = 500
+
def __init__(self, methods):
self._define_methods(methods)
+
def _define_methods(self, methods):
for spec in methods:
- if len(spec) == 2: name, action = spec
- else: name, action = spec, None
+ if len(spec) == 2:
+ name, action = spec
+ else:
+ name, action = spec, None
meth = FakeMethod(name, action, self.handle)
setattr(self.__class__, name, meth)
+
def handle(self, fn_name, action, *args, **kwds):
self.parent.calls.append((self, fn_name, args, kwds))
if action is None:
@@ -328,7 +399,7 @@ def handle(self, fn_name, action, *args, **kwds):
elif action == "return request":
return Request("http://blah/")
elif action.startswith("error"):
- code = action[action.rfind(" ")+1:]
+ code = action[action.rfind(" ") + 1:]
try:
code = int(code)
except ValueError:
@@ -338,10 +409,14 @@ def handle(self, fn_name, action, *args, **kwds):
elif action == "raise":
raise mechanize.URLError("blah")
assert False
- def close(self): pass
+
+ def close(self):
+ pass
+
def add_parent(self, parent):
self.parent = parent
self.parent.calls = []
+
def __lt__(self, other):
if not hasattr(other, "handler_order"):
# Try to preserve the old behavior of having custom classes
@@ -350,6 +425,16 @@ def __lt__(self, other):
return True
return self.handler_order < other.handler_order
+
+class MockHTTPSHandler(AbstractHTTPHandler):
+ # Useful for testing the Proxy-Authorization request by verifying the
+ # properties of httpcon
+ httpconn = MockHTTPClass()
+
+ def https_open(self, req):
+ return self.do_open(self.httpconn, req)
+
+
def add_ordered_mock_handlers(opener, meth_spec):
"""Create MockHandlers and add them to an OpenerDirector.
@@ -373,6 +458,7 @@ def add_ordered_mock_handlers(opener, meth_spec):
count = 0
for meths in meth_spec:
class MockHandlerSubclass(MockHandler): pass
+
h = MockHandlerSubclass(meths)
h.handler_order += count
h.add_parent(opener)
@@ -381,12 +467,14 @@ class MockHandlerSubclass(MockHandler): pass
opener.add_handler(h)
return handlers
+
def build_test_opener(*handler_instances):
opener = OpenerDirector()
for h in handler_instances:
opener.add_handler(h)
return opener
+
class MockHTTPHandler(mechanize.BaseHandler):
# useful for testing redirections and auth
# sends supplied headers and code as first response
@@ -395,12 +483,15 @@ def __init__(self, code, headers):
self.code = code
self.headers = headers
self.reset()
+
def reset(self):
self._count = 0
self.requests = []
+
def http_open(self, req):
import mimetools, copy
from StringIO import StringIO
+
self.requests.append(copy.deepcopy(req))
if self._count == 0:
self._count = self._count + 1
@@ -412,12 +503,14 @@ def http_open(self, req):
self.req = req
return test_response("", [], req.get_full_url())
+
class MockPasswordManager:
def add_password(self, realm, uri, user, password):
self.realm = realm
self.url = uri
self.user = user
self.password = password
+
def find_user_password(self, realm, authuri):
self.target_realm = realm
self.target_url = authuri
@@ -425,10 +518,10 @@ def find_user_password(self, realm, authuri):
class OpenerDirectorTests(unittest.TestCase):
-
def test_add_non_handler(self):
class NonHandler(object):
pass
+
self.assertRaises(TypeError,
OpenerDirector().add_handler, NonHandler())
@@ -447,11 +540,11 @@ def test_badly_named_methods(self):
meth_spec = [
[("do_open", "return self"), ("proxy_open", "return self")],
[("redirect_request", "return self")],
- ]
+ ]
handlers = add_ordered_mock_handlers(o, meth_spec)
o.add_handler(mechanize.UnknownHandler())
for scheme in "do", "proxy", "redirect":
- self.assertRaises(URLError, o.open, scheme+"://example.com/")
+ self.assertRaises(URLError, o.open, scheme + "://example.com/")
def test_handled(self):
# handler returning non-None means no more handlers will be called
@@ -461,7 +554,7 @@ def test_handled(self):
["ftp_open"],
[("http_open", "return self")],
[("http_open", "return self")],
- ]
+ ]
handlers = add_ordered_mock_handlers(o, meth_spec)
req = Request("http://example.com/")
@@ -482,24 +575,34 @@ def test_handled(self):
def test_reindex_handlers(self):
o = OpenerDirector()
+
class MockHandler:
def add_parent(self, parent): pass
- def close(self):pass
+
+ def close(self): pass
+
def __lt__(self, other):
return self.handler_order < other.handler_order
+
# this first class is here as an obscure regression test for bug
# encountered during development: if something manages to get through
# to _maybe_reindex_handlers, make sure it's properly removed and
# doesn't affect adding of subsequent handlers
class NonHandler(MockHandler):
handler_order = 1
+
class Handler(MockHandler):
handler_order = 2
+
def http_open(self): pass
+
class Processor(MockHandler):
handler_order = 3
+
def any_response(self): pass
+
def http_response(self): pass
+
o.add_handler(NonHandler())
h = Handler()
o.add_handler(h)
@@ -518,8 +621,9 @@ def test_handler_order(self):
for meths, handler_order in [
([("http_open", "return self")], 500),
(["http_open"], 0),
- ]:
+ ]:
class MockHandlerSubclass(MockHandler): pass
+
h = MockHandlerSubclass(meths)
h.handler_order = handler_order
handlers.append(h)
@@ -536,16 +640,16 @@ def test_raise(self):
meth_spec = [
[("http_open", "raise")],
[("http_open", "return self")],
- ]
+ ]
handlers = add_ordered_mock_handlers(o, meth_spec)
req = Request("http://example.com/")
self.assertRaises(mechanize.URLError, o.open, req)
self.assertEqual(o.calls, [(handlers[0], "http_open", (req,), {})])
-## def test_error(self):
-## # XXX this doesn't actually seem to be used in standard library,
-## # but should really be tested anyway...
+ ## def test_error(self):
+ ## # XXX this doesn't actually seem to be used in standard library,
+ ## # but should really be tested anyway...
def test_http_error(self):
# XXX http_error_default
@@ -557,7 +661,7 @@ def test_http_error(self):
[("http_error_302", "return response"), "http_error_303",
"http_error"],
[("http_error_302")],
- ]
+ ]
handlers = add_ordered_mock_handlers(o, meth_spec)
req = Request("http://example.com/")
@@ -576,12 +680,15 @@ def test_http_error_raised(self):
# XXX it worries me that this is the only test that excercises the else
# branch in HTTPDefaultErrorHandler
from mechanize import _response
+
o = mechanize.OpenerDirector()
o.add_handler(mechanize.HTTPErrorProcessor())
o.add_handler(mechanize.HTTPDefaultErrorHandler())
+
class HTTPHandler(AbstractHTTPHandler):
def http_open(self, req):
return _response.test_response(code=302)
+
o.add_handler(HTTPHandler())
self.assertRaises(mechanize.HTTPError, o.open, "http://example.com/")
@@ -593,7 +700,7 @@ def test_processors(self):
("http_response", "return response")],
[("http_request", "return request"),
("http_response", "return response")],
- ]
+ ]
handlers = add_ordered_mock_handlers(o, meth_spec)
req = Request("http://example.com/")
@@ -619,19 +726,19 @@ def test_processors(self):
# response from opener.open is None, because there's no
# handler that defines http_open to handle it
self.assertTrue(args[1] is None or
- isinstance(args[1], MockResponse))
+ isinstance(args[1], MockResponse))
def test_any(self):
# XXXXX two handlers case: ordering
o = OpenerDirector()
meth_spec = [[
- ("http_request", "return request"),
- ("http_response", "return response"),
- ("ftp_request", "return request"),
- ("ftp_response", "return response"),
- ("any_request", "return request"),
- ("any_response", "return response"),
- ]]
+ ("http_request", "return request"),
+ ("http_response", "return response"),
+ ("ftp_request", "return request"),
+ ("ftp_response", "return response"),
+ ("any_request", "return request"),
+ ("any_response", "return response"),
+ ]]
handlers = add_ordered_mock_handlers(o, meth_spec)
handler = handlers[0]
@@ -644,10 +751,10 @@ def test_any(self):
(handler, ("%s_request" % scheme)),
(handler, "any_response"),
(handler, ("%s_response" % scheme)),
- ]
+ ]
self.assertEqual(len(o.calls), len(calls))
for i, ((handler, name, args, kwds), calls) in (
- enumerate(zip(o.calls, calls))):
+ enumerate(zip(o.calls, calls))):
if i < 2:
# *_request
self.assert_((handler, name) == calls)
@@ -666,6 +773,7 @@ def test_any(self):
def sanepathname2url(path):
import urllib
+
urlpath = urllib.pathname2url(path)
if os.name == "nt" and urlpath.startswith("///"):
urlpath = urlpath[2:]
@@ -677,45 +785,56 @@ class MockRobotFileParserClass:
def __init__(self):
self.calls = []
self._can_fetch = True
+
def clear(self):
self.calls = []
+
def __call__(self):
self.calls.append("__call__")
return self
+
def set_url(self, url):
self.calls.append(("set_url", url))
+
def set_timeout(self, timeout):
self.calls.append(("set_timeout", timeout))
+
def set_opener(self, opener):
self.calls.append(("set_opener", opener))
+
def read(self):
self.calls.append("read")
+
def can_fetch(self, ua, url):
self.calls.append(("can_fetch", ua, url))
return self._can_fetch
+
class MockPasswordManager:
def add_password(self, realm, uri, user, password):
self.realm = realm
self.url = uri
self.user = user
self.password = password
+
def find_user_password(self, realm, authuri):
self.target_realm = realm
self.target_url = authuri
return self.user, self.password
-class HandlerTests(mechanize._testcase.TestCase):
+class HandlerTests(mechanize._testcase.TestCase):
def test_ftp(self):
class MockFTPWrapper:
def __init__(self, data): self.data = data
+
def retrfile(self, filename, filetype):
self.filename, self.filetype = filename, filetype
return StringIO.StringIO(self.data), len(self.data)
class NullFTPHandler(mechanize.FTPHandler):
def __init__(self, data): self.data = data
+
def connect_ftp(self, user, passwd, host, port, dirs, timeout):
self.user, self.passwd = user, passwd
self.host, self.port = host, port
@@ -725,6 +844,7 @@ def connect_ftp(self, user, passwd, host, port, dirs, timeout):
return self.ftpwrapper
import ftplib, socket
+
data = "rheum rhaponicum"
h = NullFTPHandler(data)
o = h.parent = MockOpener()
@@ -742,7 +862,7 @@ def connect_ftp(self, user, passwd, host, port, dirs, timeout):
"localhost", ftplib.FTP_PORT, "A",
[], _sockettimeout._GLOBAL_DEFAULT_TIMEOUT,
"baz.gif", None), # TODO: really this should guess image/gif
- ]:
+ ]:
req = Request(url, timeout=timeout)
r = h.ftp_open(req)
# ftp authentication not yet implemented by FTPHandler
@@ -760,6 +880,7 @@ def connect_ftp(self, user, passwd, host, port, dirs, timeout):
def test_file(self):
import rfc822, socket
+
h = mechanize.FileHandler()
o = h.parent = MockOpener()
@@ -771,11 +892,11 @@ def test_file(self):
except socket.gaierror:
fqdn = "localhost"
for url in [
- "file://localhost%s" % urlpath,
- "file://%s" % urlpath,
- "file://%s%s" % (socket.gethostbyname('localhost'), urlpath),
- "file://%s%s" % (fqdn, urlpath)
- ]:
+ "file://localhost%s" % urlpath,
+ "file://%s" % urlpath,
+ "file://%s%s" % (socket.gethostbyname('localhost'), urlpath),
+ "file://%s%s" % (fqdn, urlpath)
+ ]:
write_file(temp_file, towrite)
r = h.file_open(Request(url))
try:
@@ -792,21 +913,21 @@ def test_file(self):
self.assertEqual(headers["Last-modified"], modified)
for url in [
- "file://localhost:80%s" % urlpath,
- "file:///file_does_not_exist.txt",
- "file://%s:80%s/%s" % (socket.gethostbyname('localhost'),
- os.getcwd(), temp_file),
- "file://somerandomhost.ontheinternet.com%s/%s" %
- (os.getcwd(), temp_file),
- ]:
+ "file://localhost:80%s" % urlpath,
+ "file:///file_does_not_exist.txt",
+ "file://%s:80%s/%s" % (socket.gethostbyname('localhost'),
+ os.getcwd(), temp_file),
+ "file://somerandomhost.ontheinternet.com%s/%s" %
+ (os.getcwd(), temp_file),
+ ]:
write_file(temp_file, towrite)
self.assertRaises(mechanize.URLError, h.file_open, Request(url))
h = mechanize.FileHandler()
o = h.parent = MockOpener()
# XXXX why does // mean ftp (and /// mean not ftp!), and where
- # is file: scheme specified? I think this is really a bug, and
- # what was intended was to distinguish between URLs like:
+ # is file: scheme specified? I think this is really a bug, and
+ # what was intended was to distinguish between URLs like:
# file:/blah.txt (a file)
# file://localhost/blah.txt (a file)
# file:///blah.txt (a file)
@@ -814,9 +935,9 @@ def test_file(self):
for url, ftp in [
("file://ftp.example.com//foo.txt", True),
("file://ftp.example.com///foo.txt", False),
-# XXXX bug: fails with OSError, should be URLError
+ # XXXX bug: fails with OSError, should be URLError
("file://ftp.example.com/foo.txt", False),
- ]:
+ ]:
req = Request(url)
try:
h.file_open(req)
@@ -828,39 +949,6 @@ def test_file(self):
self.assertEqual(req.type, "ftp")
def test_http(self):
- class MockHTTPResponse:
- def __init__(self, fp, msg, status, reason):
- self.fp = fp
- self.msg = msg
- self.status = status
- self.reason = reason
- def read(self):
- return ''
- class MockHTTPClass:
- def __init__(self):
- self.req_headers = []
- self.data = None
- self.raise_on_endheaders = False
- def __call__(self, host,
- timeout=_sockettimeout._GLOBAL_DEFAULT_TIMEOUT):
- self.host = host
- self.timeout = timeout
- return self
- def set_debuglevel(self, level):
- self.level = level
- def request(self, method, url, body=None, headers={}):
- self.method = method
- self.selector = url
- self.req_headers += headers.items()
- self.req_headers.sort()
- if body:
- self.data = body
- if self.raise_on_endheaders:
- import socket
- raise socket.error()
- def getresponse(self):
- return MockHTTPResponse(MockFile(), {}, 200, "OK")
-
h = AbstractHTTPHandler()
o = h.parent = MockOpener()
@@ -872,11 +960,14 @@ def getresponse(self):
r = h.do_open(http, req)
# result attributes
- r.read; r.readline # wrapped MockFile methods
- r.info; r.geturl # addinfourl methods
+ r.read;
+ r.readline # wrapped MockFile methods
+ r.info;
+ r.geturl # addinfourl methods
r.code, r.msg == 200, "OK" # added from MockHTTPClass.getreply()
hdrs = r.info()
- hdrs.get; hdrs.has_key # r.info() gives dict from .getreply()
+ hdrs.get;
+ hdrs.has_key # r.info() gives dict from .getreply()
self.assertEqual(r.geturl(), url)
self.assertEqual(http.host, "example.com")
@@ -904,7 +995,7 @@ def getresponse(self):
else: # POST
self.assertEqual(req.unredirected_hdrs["Content-length"], "0")
self.assertEqual(req.unredirected_hdrs["Content-type"],
- "application/x-www-form-urlencoded")
+ "application/x-www-form-urlencoded")
# XXX the details of Host could be better tested
self.assertEqual(req.unredirected_hdrs["Host"], "example.com")
self.assertEqual(req.unredirected_hdrs["Spam"], "eggs")
@@ -940,12 +1031,12 @@ def test_http_double_slash(self):
# Check whether host is determined correctly if there is no proxy
np_ds_req = h.do_request_(ds_req)
- self.assertEqual(np_ds_req.unredirected_hdrs["Host"],"example.com")
+ self.assertEqual(np_ds_req.unredirected_hdrs["Host"], "example.com")
# Check whether host is determined correctly if there is a proxy
- ds_req.set_proxy("someproxy:3128",None)
+ ds_req.set_proxy("someproxy:3128", None)
p_ds_req = h.do_request_(ds_req)
- self.assertEqual(p_ds_req.unredirected_hdrs["Host"],"example.com")
+ self.assertEqual(p_ds_req.unredirected_hdrs["Host"], "example.com")
def test_errors(self):
h = HTTPErrorProcessor()
@@ -996,9 +1087,12 @@ def test_raise_http_errors(self):
# HTTPDefaultErrorHandler should raise HTTPError if no error handler
# handled the error response
from mechanize import _response
+
h = mechanize.HTTPDefaultErrorHandler()
- url = "http://example.com"; code = 500; msg = "Error"
+ url = "http://example.com";
+ code = 500;
+ msg = "Error"
request = mechanize.Request(url)
response = _response.test_response(url=url, code=code, msg=msg)
@@ -1026,6 +1120,7 @@ def test_raise_http_errors(self):
def test_robots(self):
# XXX useragent
from mechanize import HTTPRobotRulesProcessor
+
opener = OpenerDirector()
rfpc = MockRobotFileParserClass()
h = HTTPRobotRulesProcessor(rfpc)
@@ -1034,7 +1129,7 @@ def test_robots(self):
url = "http://example.com:80/foo/bar.html"
req = Request(url)
# first time: initialise and set up robots.txt parser before checking
- # whether OK to fetch URL
+ # whether OK to fetch URL
h.http_request(req)
self.assertEquals(rfpc.calls, [
"__call__",
@@ -1043,14 +1138,14 @@ def test_robots(self):
("set_timeout", _sockettimeout._GLOBAL_DEFAULT_TIMEOUT),
"read",
("can_fetch", "", url),
- ])
+ ])
# second time: just use existing parser
rfpc.clear()
req = Request(url)
h.http_request(req)
self.assert_(rfpc.calls == [
("can_fetch", "", url),
- ])
+ ])
# different URL on same server: same again
rfpc.clear()
url = "http://example.com:80/blah.html"
@@ -1058,7 +1153,7 @@ def test_robots(self):
h.http_request(req)
self.assert_(rfpc.calls == [
("can_fetch", "", url),
- ])
+ ])
# disallowed URL
rfpc.clear()
rfpc._can_fetch = False
@@ -1070,8 +1165,8 @@ def test_robots(self):
self.assert_(e.request == req)
self.assert_(e.code == 403)
# new host: reload robots.txt (even though the host and port are
- # unchanged, we treat this as a new host because
- # "example.com" != "example.com:80")
+ # unchanged, we treat this as a new host because
+ # "example.com" != "example.com:80")
rfpc.clear()
rfpc._can_fetch = True
url = "http://example.com/rhubarb.html"
@@ -1084,7 +1179,7 @@ def test_robots(self):
("set_timeout", _sockettimeout._GLOBAL_DEFAULT_TIMEOUT),
"read",
("can_fetch", "", url),
- ])
+ ])
# https url -> should fetch robots.txt from https url too
rfpc.clear()
url = "https://example.org/rhubarb.html"
@@ -1097,7 +1192,7 @@ def test_robots(self):
("set_timeout", _sockettimeout._GLOBAL_DEFAULT_TIMEOUT),
"read",
("can_fetch", "", url),
- ])
+ ])
# non-HTTP URL -> ignore robots.txt
rfpc.clear()
url = "ftp://example.com/"
@@ -1110,15 +1205,17 @@ def test_redirected_robots_txt(self):
# robots.txt fetch to check the redirection is allowed!
import mechanize
from mechanize import build_opener, HTTPHandler, \
- HTTPDefaultErrorHandler, HTTPRedirectHandler, \
- HTTPRobotRulesProcessor
+ HTTPDefaultErrorHandler, HTTPRedirectHandler, \
+ HTTPRobotRulesProcessor
class MockHTTPHandler(mechanize.BaseHandler):
def __init__(self):
self.requests = []
+
def http_open(self, req):
import mimetools, httplib, copy
from StringIO import StringIO
+
self.requests.append(copy.deepcopy(req))
if req.get_full_url() == "http://example.com/robots.txt":
hdr = "Location: http://example.com/en/robots.txt\r\n\r\n"
@@ -1138,7 +1235,7 @@ def http_open(self, req):
["http://example.com/robots.txt",
"http://example.com/en/robots.txt",
"http://example.com/",
- ])
+ ])
def test_cookies(self):
cj = MockCookieJar()
@@ -1163,11 +1260,11 @@ def test_http_equiv(self):
data = (''
''
''
- )
+ )
headers = [("Foo", "Bar"),
("Content-type", "text/html"),
("Refresh", "blah"),
- ]
+ ]
url = "http://example.com/"
req = Request(url)
r = mechanize._response.make_response(data, headers, url, 200, "OK")
@@ -1188,7 +1285,7 @@ def test_refresh(self):
("2", True),
# in the past, this failed with UnboundLocalError
('0; "http://example.com/foo/"', False),
- ]:
+ ]:
o = h.parent = MockOpener()
req = Request("http://example.com/")
headers = http_message({"refresh": val})
@@ -1206,14 +1303,19 @@ def __init__(self, test, seconds):
seconds = None # don't expect a sleep for 0 seconds
self._expected = seconds
self._got = None
+
def sleep(self, seconds):
self._got = seconds
+
def verify(self):
self._test.assertEqual(self._expected, self._got)
+
class Opener:
called = False
+
def error(self, *args, **kwds):
self.called = True
+
def test(rp, header, refresh_after):
expect_refresh = refresh_after is not None
opener = Opener()
@@ -1222,7 +1324,7 @@ def test(rp, header, refresh_after):
rp._sleep = st.sleep
rp.http_response(Request("http://example.com"),
test_response(headers=[("Refresh", header)]),
- )
+ )
self.assertEqual(expect_refresh, opener.called)
st.verify()
@@ -1279,9 +1381,11 @@ def test_redirect(self):
# loop detection
req = Request(from_url)
+
def redirect(h, req, url=to_url):
h.http_error_302(req, MockFile(), 302, "Blah",
http_message({"location": url}))
+
# Note that the *original* request shares the same record of
# redirections with the sub-requests caused by the redirections.
@@ -1309,6 +1413,7 @@ def redirect(h, req, url=to_url):
def test_redirect_bad_uri(self):
# bad URIs should be cleaned up before redirection
from mechanize._response import test_html_response
+
from_url = "http://example.com/a.html"
bad_to_url = "http://example.com/b. |html"
good_to_url = "http://example.com/b.%20%7Chtml"
@@ -1319,12 +1424,13 @@ def test_redirect_bad_uri(self):
req = Request(from_url)
h.http_error_302(req, test_html_response(), 302, "Blah",
http_message({"location": bad_to_url}),
- )
+ )
self.assertEqual(o.req.get_full_url(), good_to_url)
def test_refresh_bad_uri(self):
# bad URIs should be cleaned up before redirection
from mechanize._response import test_html_response
+
from_url = "http://example.com/a.html"
bad_to_url = "http://example.com/b. |html"
good_to_url = "http://example.com/b.%20%7Chtml"
@@ -1343,8 +1449,8 @@ def test_cookie_redirect(self):
# cookies shouldn't leak into redirected requests
import mechanize
from mechanize import CookieJar, build_opener, HTTPHandler, \
- HTTPCookieProcessor, HTTPError, HTTPDefaultErrorHandler, \
- HTTPRedirectHandler
+ HTTPCookieProcessor, HTTPError, HTTPDefaultErrorHandler, \
+ HTTPRedirectHandler
from test_cookies import interact_netscape
@@ -1364,7 +1470,7 @@ def test_proxy(self):
o.add_handler(ph)
meth_spec = [
[("http_open", "return response")]
- ]
+ ]
handlers = add_ordered_mock_handlers(o, meth_spec)
o._maybe_reindex_handlers()
@@ -1396,16 +1502,20 @@ def test_proxy_no_proxy(self):
def test_proxy_custom_proxy_bypass(self):
self.monkey_patch_environ("no_proxy",
mechanize._testcase.MonkeyPatcher.Unset)
+
def proxy_bypass(hostname):
return hostname == "noproxy.com"
+
o = OpenerDirector()
ph = mechanize.ProxyHandler(dict(http="proxy.example.com"),
proxy_bypass=proxy_bypass)
+
def is_proxied(url):
o.add_handler(ph)
req = Request(url)
o.open(req)
return req.has_proxy()
+
self.assertTrue(is_proxied("http://example.com"))
self.assertFalse(is_proxied("http://noproxy.com"))
@@ -1414,7 +1524,7 @@ def test_proxy_https(self):
ph = mechanize.ProxyHandler(dict(https='proxy.example.com:3128'))
o.add_handler(ph)
meth_spec = [
- [("https_open","return response")]
+ [("https_open", "return response")]
]
handlers = add_ordered_mock_handlers(o, meth_spec)
req = Request("https://www.example.com/")
@@ -1424,6 +1534,29 @@ def test_proxy_https(self):
self.assertEqual([(handlers[0], "https_open")],
[tup[0:2] for tup in o.calls])
+ def test_proxy_https_proxy_authorization(self):
+ o = OpenerDirector()
+ ph = mechanize.ProxyHandler(dict(https='proxy.example.com:3128'))
+ o.add_handler(ph)
+ https_handler = MockHTTPSHandler()
+ o.add_handler(https_handler)
+ req = Request("https://www.example.com/")
+ req.add_header("Proxy-Authorization", "FooBar")
+ req.add_header("User-Agent", "Grail")
+ self.assertEqual(req.get_host(), "www.example.com")
+ self.assertIsNone(req._tunnel_host)
+ r = o.open(req)
+ # Verify Proxy-Authorization gets tunneled to request.
+ # httpsconn req_headers do not have the Proxy-Authorization header but
+ # the req will have.
+ self.assertFalse(("Proxy-Authorization", "FooBar") in
+ https_handler.httpconn.req_headers)
+ self.assertTrue(("User-Agent", "Grail") in
+ https_handler.httpconn.req_headers)
+ self.assertIsNotNone(req._tunnel_host)
+ self.assertEqual(req.get_host(), "proxy.example.com:3128")
+ self.assertEqual(req.get_header("Proxy-authorization"), "FooBar")
+
def test_basic_auth(self, quote_char='"'):
opener = OpenerDirector()
password_manager = MockPasswordManager()
@@ -1431,14 +1564,14 @@ def test_basic_auth(self, quote_char='"'):
realm = "ACME Widget Store"
http_handler = MockHTTPHandler(
401, 'WWW-Authenticate: Basic realm=%s%s%s\r\n\r\n' %
- (quote_char, realm, quote_char) )
+ (quote_char, realm, quote_char))
opener.add_handler(auth_handler)
opener.add_handler(http_handler)
self._test_basic_auth(opener, auth_handler, "Authorization",
realm, http_handler, password_manager,
"http://acme.example.com/protected",
"http://acme.example.com/protected",
- )
+ )
def test_basic_auth_with_single_quoted_realm(self):
self.test_basic_auth(quote_char="'")
@@ -1458,7 +1591,7 @@ def test_proxy_basic_auth(self):
realm, http_handler, password_manager,
"http://acme.example.com:3128/protected",
"proxy.example.com:3128",
- )
+ )
def test_basic_and_digest_auth_handlers(self):
# HTTPDigestAuthHandler threw an exception if it couldn't handle a 40*
@@ -1473,18 +1606,21 @@ class RecordingOpenerDirector(OpenerDirector):
def __init__(self):
OpenerDirector.__init__(self)
self.recorded = []
+
def record(self, info):
self.recorded.append(info)
+
class TestDigestAuthHandler(mechanize.HTTPDigestAuthHandler):
def http_error_401(self, *args, **kwds):
self.parent.record("digest")
mechanize.HTTPDigestAuthHandler.http_error_401(self,
- *args, **kwds)
+ *args, **kwds)
+
class TestBasicAuthHandler(mechanize.HTTPBasicAuthHandler):
def http_error_401(self, *args, **kwds):
self.parent.record("basic")
mechanize.HTTPBasicAuthHandler.http_error_401(self,
- *args, **kwds)
+ *args, **kwds)
opener = RecordingOpenerDirector()
password_manager = MockPasswordManager()
@@ -1503,15 +1639,16 @@ def http_error_401(self, *args, **kwds):
realm, http_handler, password_manager,
"http://acme.example.com/protected",
"http://acme.example.com/protected",
- )
+ )
# check digest was tried before basic (twice, because
# _test_basic_auth called .open() twice)
- self.assertEqual(opener.recorded, ["digest", "basic"]*2)
+ self.assertEqual(opener.recorded, ["digest", "basic"] * 2)
def _test_basic_auth(self, opener, auth_handler, auth_header,
realm, http_handler, password_manager,
request_url, protected_url):
import base64
+
user, password = "wile", "coyote"
# .add_password() fed through to password manager
@@ -1531,7 +1668,7 @@ def _test_basic_auth(self, opener, auth_handler, auth_header,
self.assertEqual(len(http_handler.requests), 2)
self.assertFalse(http_handler.requests[0].has_header(auth_header))
userpass = '%s:%s' % (user, password)
- auth_hdr_value = 'Basic '+base64.encodestring(userpass).strip()
+ auth_hdr_value = 'Basic ' + base64.encodestring(userpass).strip()
self.assertEqual(http_handler.requests[1].get_header(auth_header),
auth_hdr_value)
@@ -1545,14 +1682,14 @@ def _test_basic_auth(self, opener, auth_handler, auth_header,
class HeadParserTests(unittest.TestCase):
-
def test(self):
# XXX XHTML
from mechanize import HeadParser
+
htmls = [
("""
""",
- [("refresh", "1; http://example.com/")]
+ [("refresh", "1; http://example.com/")]
),
("""
@@ -1567,29 +1704,38 @@ def test(self):
("""
""",
[])
- ]
+ ]
for html, result in htmls:
self.assertEqual(parse_head(StringIO.StringIO(html), HeadParser()), result)
-
class A:
def a(self): pass
+
+
class B(A):
def a(self): pass
+
def b(self): pass
+
+
class C(A):
def c(self): pass
+
+
class D(C, B):
def a(self): pass
+
def d(self): pass
-class FunctionTests(unittest.TestCase):
+class FunctionTests(unittest.TestCase):
def test_build_opener(self):
class MyHTTPHandler(HTTPHandler): pass
+
class FooHandler(mechanize.BaseHandler):
def foo_open(self): pass
+
class BarHandler(mechanize.BaseHandler):
def bar_open(self): pass
@@ -1617,6 +1763,7 @@ def bar_open(self): pass
# Issue2670: multiple handlers sharing the same base class
class MyOtherHTTPHandler(HTTPHandler): pass
+
o = build_opener(MyHTTPHandler, MyOtherHTTPHandler)
self.opener_has_handler(o, MyHTTPHandler)
self.opener_has_handler(o, MyOtherHTTPHandler)
@@ -1628,13 +1775,13 @@ def opener_has_handler(self, opener, handler_class):
else:
self.assertTrue(False)
-class RequestTests(unittest.TestCase):
+class RequestTests(unittest.TestCase):
def setUp(self):
self.get = Request("http://www.python.org/~jeremy/")
self.post = Request("http://www.python.org/~jeremy/",
- "data",
- headers={"X-Test": "test"})
+ "data",
+ headers={"X-Test": "test"})
def test_method(self):
self.assertEqual("POST", self.post.get_method())
@@ -1676,5 +1823,6 @@ def test_proxy(self):
if __name__ == "__main__":
import doctest
+
doctest.testmod()
unittest.main()