summaryrefslogtreecommitdiffstats
path: root/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py
diff options
context:
space:
mode:
authoryum <yum.food.vr@gmail.com>2023-01-01 21:05:27 -0800
committeryum <yum.food.vr@gmail.com>2023-01-01 21:44:45 -0800
commite25bdba3a3a53b09be5269d8b065c13b73ab55c3 (patch)
tree1d1dc1d94cde92c2f4f8ce86017395054787515d /FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py
parent0d408cc812a094a708edbe4baf536e928731cfc3 (diff)
Embed git in package
package.ps1 fetches PortableGit and embeds it in the package. This eliminates all but one runtime dependency (MSVC++ Redistributable). * Move Python into a new FOSS folder.
Diffstat (limited to 'FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py')
-rw-r--r--FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py1569
1 files changed, 1569 insertions, 0 deletions
diff --git a/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py b/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py
new file mode 100644
index 0000000..2d69dad
--- /dev/null
+++ b/FOSS/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py
@@ -0,0 +1,1569 @@
+from __future__ import absolute_import, division, unicode_literals
+import os
+import io
+import socket
+import array
+import sys
+
+import http.client
+from future.standard_library import install_aliases
+from future.backports.test import support
+import future.backports.urllib.request as urllib_request
+# The proxy bypass method imported below has logic specific to the OSX
+# proxy config data structure but is testable on all platforms.
+from future.backports.urllib.request import Request, OpenerDirector, _proxy_bypass_macosx_sysconf
+import future.backports.urllib.error as urllib_error
+from future.tests.base import unittest, skip26
+from future.builtins import bytes, dict, int, open, str, zip
+from future.utils import text_to_native_str
+
+install_aliases() # for base64.encodebytes on Py2
+
+# from future.tests.test_http_cookiejar import interact_netscape
+
+class FakeResponse(object):
+ def __init__(self, headers=[], url=None):
+ """
+ headers: list of RFC822-style 'Key: value' strings
+ """
+ import email
+ # The email.message_from_string is available on both Py2.7 and Py3.3
+ self._headers = email.message_from_string("\n".join(headers))
+ self._url = url
+ def info(self): return self._headers
+
+
+def interact_netscape(cookiejar, url, *set_cookie_hdrs):
+ return _interact(cookiejar, url, set_cookie_hdrs, "Set-Cookie")
+
+def _interact(cookiejar, url, set_cookie_hdrs, hdr_name):
+ """Perform a single request / response cycle, returning Cookie: header."""
+ req = urllib_request.Request(url)
+ cookiejar.add_cookie_header(req)
+ cookie_hdr = req.get_header("Cookie", "")
+ headers = []
+ for hdr in set_cookie_hdrs:
+ headers.append("%s: %s" % (hdr_name, hdr))
+ res = FakeResponse(headers, url)
+ cookiejar.extract_cookies(res, req)
+ return cookie_hdr
+
+
+# XXX
+# Request
+# CacheFTPHandler (hard to write)
+# parse_keqv_list, parse_http_list, HTTPDigestAuthHandler
+
+class TrivialTests(unittest.TestCase):
+
+ def test___all__(self):
+ # Verify which names are exposed
+ for module in 'request', 'response', 'parse', 'error', 'robotparser':
+ context = {}
+ exec('from future.backports.urllib.%s import *' % module, context)
+ del context['__builtins__']
+ if module == 'request' and os.name == 'nt':
+ u, p = context.pop('url2pathname'), context.pop('pathname2url')
+ self.assertEqual(u.__module__, 'nturl2path')
+ self.assertEqual(p.__module__, 'nturl2path')
+ for k, v in context.items():
+ self.assertEqual(v.__module__, 'future.backports.urllib.%s' % module,
+ "%r is exposed in 'future.backports.urllib.%s' but defined in %r" %
+ (k, module, v.__module__))
+
+ def test_trivial(self):
+ # A couple trivial tests
+
+ self.assertRaises(ValueError, urllib_request.urlopen, 'bogus url')
+
+ # XXX Name hacking to get this to work on Windows.
+ fname = os.path.abspath(urllib_request.__file__).replace('\\', '/')
+
+ if os.name == 'nt':
+ file_url = "file:///%s" % fname
+ else:
+ file_url = "file://%s" % fname
+
+ f = urllib_request.urlopen(file_url)
+
+ f.read()
+ f.close()
+
+ def test_parse_http_list(self):
+ tests = [
+ ('a,b,c', ['a', 'b', 'c']),
+ ('path"o,l"og"i"cal, example', ['path"o,l"og"i"cal', 'example']),
+ ('a, b, "c", "d", "e,f", g, h',
+ ['a', 'b', '"c"', '"d"', '"e,f"', 'g', 'h']),
+ ('a="b\\"c", d="e\\,f", g="h\\\\i"',
+ ['a="b"c"', 'd="e,f"', 'g="h\\i"'])]
+ for string, list in tests:
+ self.assertEqual(urllib_request.parse_http_list(string), list)
+
+ def test_URLError_reasonstr(self):
+ err = urllib_error.URLError('reason')
+ self.assertIn(err.reason, str(err))
+
+class RequestHdrsTests(unittest.TestCase):
+
+ def test_request_headers_dict(self):
+ """
+ The Request.headers dictionary is not a documented interface. It
+ should stay that way, because the complete set of headers are only
+ accessible through the .get_header(), .has_header(), .header_items()
+ interface. However, .headers pre-dates those methods, and so real code
+ will be using the dictionary.
+
+ The introduction in 2.4 of those methods was a mistake for the same
+ reason: code that previously saw all (urllib2 user)-provided headers in
+ .headers now sees only a subset.
+
+ """
+ url = "http://example.com"
+ self.assertEqual(Request(url,
+ headers={"Spam-eggs": "blah"}
+ ).headers["Spam-eggs"], "blah")
+ self.assertEqual(Request(url,
+ headers={"spam-EggS": "blah"}
+ ).headers["Spam-eggs"], "blah")
+
+ def test_request_headers_methods(self):
+ """
+ Note the case normalization of header names here, to
+ .capitalize()-case. This should be preserved for
+ backwards-compatibility. (In the HTTP case, normalization to
+ .title()-case is done by urllib2 before sending headers to
+ http.client).
+
+ Note that e.g. r.has_header("spam-EggS") is currently False, and
+ r.get_header("spam-EggS") returns None, but that could be changed in
+ future.
+
+ Method r.remove_header should remove items both from r.headers and
+ r.unredirected_hdrs dictionaries
+ """
+ url = "http://example.com"
+ req = Request(url, headers={"Spam-eggs": "blah"})
+ self.assertTrue(req.has_header("Spam-eggs"))
+ self.assertEqual(req.header_items(), [('Spam-eggs', 'blah')])
+
+ req.add_header("Foo-Bar", "baz")
+ self.assertEqual(sorted(req.header_items()),
+ [('Foo-bar', 'baz'), ('Spam-eggs', 'blah')])
+ self.assertFalse(req.has_header("Not-there"))
+ self.assertIsNone(req.get_header("Not-there"))
+ self.assertEqual(req.get_header("Not-there", "default"), "default")
+
+
+ def test_password_manager(self):
+ mgr = urllib_request.HTTPPasswordMgr()
+ add = mgr.add_password
+ find_user_pass = mgr.find_user_password
+ add("Some Realm", "http://example.com/", "joe", "password")
+ add("Some Realm", "http://example.com/ni", "ni", "ni")
+ add("c", "http://example.com/foo", "foo", "ni")
+ add("c", "http://example.com/bar", "bar", "nini")
+ add("b", "http://example.com/", "first", "blah")
+ add("b", "http://example.com/", "second", "spam")
+ add("a", "http://example.com", "1", "a")
+ add("Some Realm", "http://c.example.com:3128", "3", "c")
+ add("Some Realm", "d.example.com", "4", "d")
+ add("Some Realm", "e.example.com:3128", "5", "e")
+
+ self.assertEqual(find_user_pass("Some Realm", "example.com"),
+ ('joe', 'password'))
+
+ #self.assertEqual(find_user_pass("Some Realm", "http://example.com/ni"),
+ # ('ni', 'ni'))
+
+ self.assertEqual(find_user_pass("Some Realm", "http://example.com"),
+ ('joe', 'password'))
+ self.assertEqual(find_user_pass("Some Realm", "http://example.com/"),
+ ('joe', 'password'))
+ self.assertEqual(
+ find_user_pass("Some Realm", "http://example.com/spam"),
+ ('joe', 'password'))
+ self.assertEqual(
+ find_user_pass("Some Realm", "http://example.com/spam/spam"),
+ ('joe', 'password'))
+ self.assertEqual(find_user_pass("c", "http://example.com/foo"),
+ ('foo', 'ni'))
+ self.assertEqual(find_user_pass("c", "http://example.com/bar"),
+ ('bar', 'nini'))
+ self.assertEqual(find_user_pass("b", "http://example.com/"),
+ ('second', 'spam'))
+
+ # No special relationship between a.example.com and example.com:
+
+ self.assertEqual(find_user_pass("a", "http://example.com/"),
+ ('1', 'a'))
+ self.assertEqual(find_user_pass("a", "http://a.example.com/"),
+ (None, None))
+
+ # Ports:
+
+ self.assertEqual(find_user_pass("Some Realm", "c.example.com"),
+ (None, None))
+ self.assertEqual(find_user_pass("Some Realm", "c.example.com:3128"),
+ ('3', 'c'))
+ self.assertEqual(
+ find_user_pass("Some Realm", "http://c.example.com:3128"),
+ ('3', 'c'))
+ self.assertEqual(find_user_pass("Some Realm", "d.example.com"),
+ ('4', 'd'))
+ self.assertEqual(find_user_pass("Some Realm", "e.example.com:3128"),
+ ('5', 'e'))
+
+ def test_password_manager_default_port(self):
+ """
+ The point to note here is that we can't guess the default port if
+ there's no scheme. This applies to both add_password and
+ find_user_password.
+ """
+ mgr = urllib_request.HTTPPasswordMgr()
+ add = mgr.add_password
+ find_user_pass = mgr.find_user_password
+ add("f", "http://g.example.com:80", "10", "j")
+ add("g", "http://h.example.com", "11", "k")
+ add("h", "i.example.com:80", "12", "l")
+ add("i", "j.example.com", "13", "m")
+ self.assertEqual(find_user_pass("f", "g.example.com:100"),
+ (None, None))
+ self.assertEqual(find_user_pass("f", "g.example.com:80"),
+ ('10', 'j'))
+ self.assertEqual(find_user_pass("f", "g.example.com"),
+ (None, None))
+ self.assertEqual(find_user_pass("f", "http://g.example.com:100"),
+ (None, None))
+ self.assertEqual(find_user_pass("f", "http://g.example.com:80"),
+ ('10', 'j'))
+ self.assertEqual(find_user_pass("f", "http://g.example.com"),
+ ('10', 'j'))
+ self.assertEqual(find_user_pass("g", "h.example.com"), ('11', 'k'))
+ self.assertEqual(find_user_pass("g", "h.example.com:80"), ('11', 'k'))
+ self.assertEqual(find_user_pass("g", "http://h.example.com:80"),
+ ('11', 'k'))
+ self.assertEqual(find_user_pass("h", "i.example.com"), (None, None))
+ self.assertEqual(find_user_pass("h", "i.example.com:80"), ('12', 'l'))
+ self.assertEqual(find_user_pass("h", "http://i.example.com:80"),
+ ('12', 'l'))
+ self.assertEqual(find_user_pass("i", "j.example.com"), ('13', 'm'))
+ self.assertEqual(find_user_pass("i", "j.example.com:80"),
+ (None, None))
+ self.assertEqual(find_user_pass("i", "http://j.example.com"),
+ ('13', 'm'))
+ self.assertEqual(find_user_pass("i", "http://j.example.com:80"),
+ (None, None))
+
+
+class MockOpener(object):
+ addheaders = []
+ def open(self, req, data=None, timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
+ self.req, self.data, self.timeout = req, data, timeout
+ def error(self, proto, *args):
+ self.proto, self.args = proto, args
+
+class MockFile(object):
+ def read(self, count=None): pass
+ def readline(self, count=None): pass
+ def close(self): pass
+
+class MockHeaders(dict):
+ def getheaders(self, name):
+ return list(self.values())
+
+class MockResponse(io.StringIO):
+ def __init__(self, code, msg, headers, data, url=None):
+ io.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(object):
+ def add_cookie_header(self, request):
+ self.ach_req = request
+ def extract_cookies(self, response, request):
+ self.ec_req, self.ec_r = request, response
+
+class FakeMethod(object):
+ 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(io.IOBase):
+ def __init__(self, fp, msg, status, reason):
+ self.fp = fp
+ self.msg = msg
+ self.status = status
+ self.reason = reason
+ self.code = 200
+
+ def read(self):
+ return ''
+
+ def info(self):
+ return {}
+
+ def geturl(self):
+ return self.url
+
+
+class MockHTTPClass(object):
+ def __init__(self):
+ self.level = 0
+ self.req_headers = []
+ self.data = None
+ self.raise_on_endheaders = False
+ self.sock = None
+ self._tunnel_headers = {}
+
+ def __call__(self, host, timeout=socket._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=None):
+ self.method = method
+ self.selector = url
+ if headers is not None:
+ 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")
+
+ def close(self):
+ pass
+
+class MockHandler(object):
+ # 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
+ 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:
+ return None
+ elif action == "return self":
+ return self
+ elif action == "return response":
+ res = MockResponse(200, "OK", {}, "")
+ return res
+ elif action == "return request":
+ return Request("http://blah/")
+ elif action.startswith("error"):
+ code = action[action.rfind(" ")+1:]
+ try:
+ code = int(code)
+ except ValueError:
+ pass
+ res = MockResponse(200, "OK", {}, "")
+ return self.parent.error("http", args[0], res, code, "", {})
+ elif action == "raise":
+ raise urllib_error.URLError("blah")
+ assert False
+ 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"):
+ # No handler_order, leave in original order. Yuck.
+ return True
+ return self.handler_order < other.handler_order
+
+def add_ordered_mock_handlers(opener, meth_spec):
+ """Create MockHandlers and add them to an OpenerDirector.
+
+ meth_spec: list of lists of tuples and strings defining methods to define
+ on handlers. eg:
+
+ [["http_error", "ftp_open"], ["http_open"]]
+
+ defines methods .http_error() and .ftp_open() on one handler, and
+ .http_open() on another. These methods just record their arguments and
+ return None. Using a tuple instead of a string causes the method to
+ perform some action (see MockHandler.handle()), eg:
+
+ [["http_error"], [("http_open", "return request")]]
+
+ defines .http_error() on one handler (which simply returns None), and
+ .http_open() on another handler, which returns a Request object.
+
+ """
+ handlers = []
+ count = 0
+ for meths in meth_spec:
+ class MockHandlerSubclass(MockHandler): pass
+ h = MockHandlerSubclass(meths)
+ h.handler_order += count
+ h.add_parent(opener)
+ count = count + 1
+ handlers.append(h)
+ 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(urllib_request.BaseHandler):
+ # useful for testing redirections and auth
+ # sends supplied headers and code as first response
+ # sends 200 OK as second response
+ 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 future.backports.email as email
+ import copy
+ self.requests.append(copy.deepcopy(req))
+ if self._count == 0:
+ self._count = self._count + 1
+ name = http.client.responses[self.code]
+ msg = email.message_from_string(self.headers)
+ return self.parent.error(
+ "http", req, MockFile(), self.code, name, msg)
+ else:
+ self.req = req
+ msg = email.message_from_string("\r\n\r\n")
+ return MockResponse(200, "OK", msg, "", req.get_full_url())
+
+class MockHTTPSHandler(urllib_request.AbstractHTTPHandler):
+ # Useful for testing the Proxy-Authorization request by verifying the
+ # properties of httpcon
+
+ def __init__(self):
+ urllib_request.AbstractHTTPHandler.__init__(self)
+ self.httpconn = MockHTTPClass()
+
+ def https_open(self, req):
+ return self.do_open(self.httpconn, req)
+
+class MockPasswordManager(object):
+ 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 OpenerDirectorTests(unittest.TestCase):
+
+ def test_add_non_handler(self):
+ class NonHandler(object):
+ pass
+ self.assertRaises(TypeError,
+ OpenerDirector().add_handler, NonHandler())
+
+ def test_badly_named_methods(self):
+ # test work-around for three methods that accidentally follow the
+ # naming conventions for handler methods
+ # (*_open() / *_request() / *_response())
+
+ # These used to call the accidentally-named methods, causing a
+ # TypeError in real code; here, returning self from these mock
+ # methods would either cause no exception, or AttributeError.
+
+ from future.backports.urllib.error import URLError
+
+ o = OpenerDirector()
+ meth_spec = [
+ [("do_open", "return self"), ("proxy_open", "return self")],
+ [("redirect_request", "return self")],
+ ]
+ add_ordered_mock_handlers(o, meth_spec)
+ o.add_handler(urllib_request.UnknownHandler())
+ for scheme in "do", "proxy", "redirect":
+ self.assertRaises(URLError, o.open, scheme+"://example.com/")
+
+ def test_handled(self):
+ # handler returning non-None means no more handlers will be called
+ o = OpenerDirector()
+ meth_spec = [
+ ["http_open", "ftp_open", "http_error_302"],
+ ["ftp_open"],
+ [("http_open", "return self")],
+ [("http_open", "return self")],
+ ]
+ handlers = add_ordered_mock_handlers(o, meth_spec)
+
+ req = Request("http://example.com/")
+ r = o.open(req)
+ # Second .http_open() gets called, third doesn't, since second returned
+ # non-None. Handlers without .http_open() never get any methods called
+ # on them.
+ # In fact, second mock handler defining .http_open() returns self
+ # (instead of response), which becomes the OpenerDirector's return
+ # value.
+ self.assertEqual(r, handlers[2])
+ calls = [(handlers[0], "http_open"), (handlers[2], "http_open")]
+ for expected, got in zip(calls, o.calls):
+ handler, name, args, kwds = got
+ self.assertEqual((handler, name), expected)
+ self.assertEqual(args, (req,))
+
+ def test_handler_order(self):
+ o = OpenerDirector()
+ handlers = []
+ 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)
+ o.add_handler(h)
+
+ o.open("http://example.com/")
+ # handlers called in reverse order, thanks to their sort order
+ self.assertEqual(o.calls[0][0], handlers[1])
+ self.assertEqual(o.calls[1][0], handlers[0])
+
+ def test_raise(self):
+ # raising URLError stops processing of request
+ o = OpenerDirector()
+ meth_spec = [
+ [("http_open", "raise")],
+ [("http_open", "return self")],
+ ]
+ handlers = add_ordered_mock_handlers(o, meth_spec)
+
+ req = Request("http://example.com/")
+ self.assertRaises(urllib_error.URLError, o.open, req)
+ self.assertEqual(o.calls, [(handlers[0], "http_open", (req,), {})])
+
+ def test_http_error(self):
+ # XXX http_error_default
+ # http errors are a special case
+ o = OpenerDirector()
+ meth_spec = [
+ [("http_open", "error 302")],
+ [("http_error_400", "raise"), "http_open"],
+ [("http_error_302", "return response"), "http_error_303",
+ "http_error"],
+ [("http_error_302")],
+ ]
+ handlers = add_ordered_mock_handlers(o, meth_spec)
+
+ class Unknown(object):
+ def __eq__(self, other): return True
+
+ req = Request("http://example.com/")
+ o.open(req)
+ assert len(o.calls) == 2
+ calls = [(handlers[0], "http_open", (req,)),
+ (handlers[2], "http_error_302",
+ (req, Unknown(), 302, "", {}))]
+ for expected, got in zip(calls, o.calls):
+ handler, method_name, args = expected
+ self.assertEqual((handler, method_name), got[:2])
+ self.assertEqual(args, got[2])
+
+
+ def test_processors(self):
+ # *_request / *_response methods get called appropriately
+ o = OpenerDirector()
+ meth_spec = [
+ [("http_request", "return request"),
+ ("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/")
+ o.open(req)
+ # processor methods are called on *all* handlers that define them,
+ # not just the first handler that handles the request
+ calls = [
+ (handlers[0], "http_request"), (handlers[1], "http_request"),
+ (handlers[0], "http_response"), (handlers[1], "http_response")]
+
+ for i, (handler, name, args, kwds) in enumerate(o.calls):
+ if i < 2:
+ # *_request
+ self.assertEqual((handler, name), calls[i])
+ self.assertEqual(len(args), 1)
+ self.assertIsInstance(args[0], Request)
+ else:
+ # *_response
+ self.assertEqual((handler, name), calls[i])
+ self.assertEqual(len(args), 2)
+ self.assertIsInstance(args[0], Request)
+ # 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))
+
+ def test_method_deprecations(self):
+ req = Request("http://www.example.com")
+
+ with self.assertWarns(DeprecationWarning):
+ req.add_data("data")
+ with self.assertWarns(DeprecationWarning):
+ req.get_data()
+ with self.assertWarns(DeprecationWarning):
+ req.has_data()
+ with self.assertWarns(DeprecationWarning):
+ req.get_host()
+ with self.assertWarns(DeprecationWarning):
+ req.get_selector()
+ with self.assertWarns(DeprecationWarning):
+ req.is_unverifiable()
+ with self.assertWarns(DeprecationWarning):
+ req.get_origin_req_host()
+ with self.assertWarns(DeprecationWarning):
+ req.get_type()
+
+
+def sanepathname2url(path):
+ try:
+ path.encode("utf-8")
+ except UnicodeEncodeError:
+ raise unittest.SkipTest("path is not encodable to utf8")
+ urlpath = urllib_request.pathname2url(path)
+ if os.name == "nt" and urlpath.startswith("///"):
+ urlpath = urlpath[2:]
+ # XXX don't ask me about the mac...
+ return urlpath
+
+class HandlerTests(unittest.TestCase):
+
+ def test_ftp(self):
+ class MockFTPWrapper(object):
+ def __init__(self, data): self.data = data
+ def retrfile(self, filename, filetype):
+ self.filename, self.filetype = filename, filetype
+ return io.StringIO(self.data), len(self.data)
+ def close(self): pass
+
+ class NullFTPHandler(urllib_request.FTPHandler):
+ def __init__(self, data): self.data = data
+ def connect_ftp(self, user, passwd, host, port, dirs,
+ timeout=socket._GLOBAL_DEFAULT_TIMEOUT):
+ self.user, self.passwd = user, passwd
+ self.host, self.port = host, port
+ self.dirs = dirs
+ self.ftpwrapper = MockFTPWrapper(self.data)
+ return self.ftpwrapper
+
+ import ftplib
+ data = "rheum rhaponicum"
+ h = NullFTPHandler(data)
+ h.parent = MockOpener()
+
+ # MIME guessing works in Python 3.8!
+ guessed_mime = None
+ if sys.hexversion >= 0x03080000:
+ guessed_mime = "image/gif"
+ for url, host, port, user, passwd, type_, dirs, filename, mimetype in [
+ ("ftp://localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://parrot@localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "parrot", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://%25parrot@localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "%parrot", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://%2542parrot@localhost/foo/bar/baz.html",
+ "localhost", ftplib.FTP_PORT, "%42parrot", "", "I",
+ ["foo", "bar"], "baz.html", "text/html"),
+ ("ftp://localhost:80/foo/bar/",
+ "localhost", 80, "", "", "D",
+ ["foo", "bar"], "", None),
+ ("ftp://localhost/baz.gif;type=a",
+ "localhost", ftplib.FTP_PORT, "", "", "A",
+ [], "baz.gif", guessed_mime),
+ ]:
+ req = Request(url)
+ req.timeout = None
+ r = h.ftp_open(req)
+ # ftp authentication not yet implemented by FTPHandler
+ self.assertEqual(h.user, user)
+ self.assertEqual(h.passwd, passwd)
+ self.assertEqual(h.host, socket.gethostbyname(host))
+ self.assertEqual(h.port, port)
+ self.assertEqual(h.dirs, dirs)
+ self.assertEqual(h.ftpwrapper.filename, filename)
+ self.assertEqual(h.ftpwrapper.filetype, type_)
+ headers = r.info()
+ self.assertEqual(headers.get("Content-type"), mimetype)
+ self.assertEqual(int(headers["Content-length"]), len(data))
+
+ def test_file(self):
+ import future.backports.email.utils as email_utils
+ import socket
+ h = urllib_request.FileHandler()
+ o = h.parent = MockOpener()
+
+ TESTFN = support.TESTFN
+ urlpath = sanepathname2url(os.path.abspath(TESTFN))
+ towrite = b"hello, world\n"
+ urls = [
+ "file://localhost%s" % urlpath,
+ "file://%s" % urlpath,
+ "file://%s%s" % (socket.gethostbyname('localhost'), urlpath),
+ ]
+ try:
+ localaddr = socket.gethostbyname(socket.gethostname())
+ except socket.gaierror:
+ localaddr = ''
+ if localaddr:
+ urls.append("file://%s%s" % (localaddr, urlpath))
+
+ for url in urls:
+ f = open(TESTFN, "wb")
+ try:
+ try:
+ f.write(towrite)
+ finally:
+ f.close()
+
+ r = h.file_open(Request(url))
+ try:
+ data = r.read()
+ headers = r.info()
+ respurl = r.geturl()
+ finally:
+ r.close()
+ stats = os.stat(TESTFN)
+ modified = email_utils.formatdate(stats.st_mtime, usegmt=True)
+ finally:
+ os.remove(TESTFN)
+ self.assertEqual(data, towrite)
+ self.assertEqual(headers["Content-type"], "text/plain")
+ self.assertEqual(headers["Content-length"], "13")
+ self.assertEqual(headers["Last-modified"], modified)
+ self.assertEqual(respurl, url)
+
+ for url in [
+ "file://localhost:80%s" % urlpath,
+ "file:///file_does_not_exist.txt",
+ "file://%s:80%s/%s" % (socket.gethostbyname('localhost'),
+ os.getcwd(), TESTFN),
+ "file://somerandomhost.ontheinternet.com%s/%s" %
+ (os.getcwd(), TESTFN),
+ ]:
+ try:
+ f = open(TESTFN, "wb")
+ try:
+ f.write(towrite)
+ finally:
+ f.close()
+
+ self.assertRaises(urllib_error.URLError,
+ h.file_open, Request(url))
+ finally:
+ os.remove(TESTFN)
+
+ h = urllib_request.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:
+ # file:/blah.txt (a file)
+ # file://localhost/blah.txt (a file)
+ # file:///blah.txt (a file)
+ # file://ftp.example.com/blah.txt (an ftp URL)
+ for url, ftp in [
+ ("file://ftp.example.com//foo.txt", False),
+ ("file://ftp.example.com///foo.txt", False),
+# XXXX bug: fails with OSError, should be URLError
+ ("file://ftp.example.com/foo.txt", False),
+ ("file://somehost//foo/something.txt", False),
+ ("file://localhost//foo/something.txt", False),
+ ]:
+ req = Request(url)
+ try:
+ h.file_open(req)
+ # XXXX remove OSError when bug fixed
+ except (urllib_error.URLError, OSError):
+ self.assertFalse(ftp)
+ else:
+ self.assertIs(o.req, req)
+ self.assertEqual(req.type, "ftp")
+ self.assertEqual(req.type == "ftp", ftp)
+
+ @skip26
+ def test_http(self):
+
+ h = urllib_request.AbstractHTTPHandler()
+ o = h.parent = MockOpener()
+
+ url = "http://example.com/"
+ for method, data in [("GET", None), ("POST", b"blah")]:
+ req = Request(url, data, {"Foo": "bar"})
+ req.timeout = None
+ req.add_unredirected_header("Spam", "eggs")
+ http = MockHTTPClass()
+ r = h.do_open(http, req)
+
+ # result attributes
+ 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.__contains__ # r.info() gives dict from .getreply()
+ self.assertEqual(r.geturl(), url)
+
+ self.assertEqual(http.host, "example.com")
+ self.assertEqual(http.level, 0)
+ self.assertEqual(http.method, method)
+ self.assertEqual(http.selector, "/")
+ self.assertEqual(http.req_headers,
+ [("Connection", "close"),
+ ("Foo", "bar"), ("Spam", "eggs")])
+ self.assertEqual(http.data, data)
+
+ # check socket.error converted to URLError
+ http.raise_on_endheaders = True
+ self.assertRaises(urllib_error.URLError, h.do_open, http, req)
+
+ # Check for TypeError on POST data which is str.
+ req = Request("http://example.com/","badpost")
+ self.assertRaises(TypeError, h.do_request_, req)
+
+ # check adding of standard headers
+ o.addheaders = [("Spam", "eggs")]
+ for data in b"", None: # POST, GET
+ req = Request("http://example.com/", data)
+ r = MockResponse(200, "OK", {}, "")
+ newreq = h.do_request_(req)
+ if data is None: # GET
+ self.assertNotIn("Content-length", req.unredirected_hdrs)
+ self.assertNotIn("Content-type", req.unredirected_hdrs)
+ else: # POST
+ self.assertEqual(req.unredirected_hdrs["Content-length"], "0")
+ self.assertEqual(req.unredirected_hdrs["Content-type"],
+ "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")
+
+ # don't clobber existing headers
+ req.add_unredirected_header("Content-length", "foo")
+ req.add_unredirected_header("Content-type", "bar")
+ req.add_unredirected_header("Host", "baz")
+ req.add_unredirected_header("Spam", "foo")
+ newreq = h.do_request_(req)
+ self.assertEqual(req.unredirected_hdrs["Content-length"], "foo")
+ self.assertEqual(req.unredirected_hdrs["Content-type"], "bar")
+ self.assertEqual(req.unredirected_hdrs["Host"], "baz")
+ self.assertEqual(req.unredirected_hdrs["Spam"], "foo")
+
+ # Check iterable body support
+ def iterable_body():
+ yield b"one"
+ yield b"two"
+ yield b"three"
+
+ for headers in {}, {"Content-Length": 11}:
+ req = Request("http://example.com/", iterable_body(), headers)
+ if not headers:
+ # Having an iterable body without a Content-Length should
+ # raise an exception
+ self.assertRaises(ValueError, h.do_request_, req)
+ else:
+ newreq = h.do_request_(req)
+
+ # A file object.
+ # Test only Content-Length attribute of request.
+
+ file_obj = io.BytesIO()
+ file_obj.write(b"Something\nSomething\nSomething\n")
+
+ for headers in {}, {"Content-Length": 30}:
+ req = Request("http://example.com/", file_obj, headers)
+ if not headers:
+ # Having an iterable body without a Content-Length should
+ # raise an exception
+ self.assertRaises(ValueError, h.do_request_, req)
+ else:
+ newreq = h.do_request_(req)
+ self.assertEqual(int(newreq.get_header('Content-length')),30)
+
+ file_obj.close()
+
+ # array.array Iterable - Content Length is calculated
+
+ iterable_array = array.array(text_to_native_str("I"),
+ [1,2,3,4])
+
+ for headers in {}, {"Content-Length": 16}:
+ req = Request("http://example.com/", iterable_array, headers)
+ newreq = h.do_request_(req)
+ self.assertEqual(int(newreq.get_header('Content-length')),16)
+
+ @skip26
+ def test_http_doubleslash(self):
+ # Checks the presence of any unnecessary double slash in url does not
+ # break anything. Previously, a double slash directly after the host
+ # could cause incorrect parsing.
+ h = urllib_request.AbstractHTTPHandler()
+ h.parent = MockOpener()
+
+ data = b""
+ ds_urls = [
+ "http://example.com/foo/bar/baz.html",
+ "http://example.com//foo/bar/baz.html",
+ "http://example.com/foo//bar/baz.html",
+ "http://example.com/foo/bar//baz.html"
+ ]
+
+ for ds_url in ds_urls:
+ ds_req = Request(ds_url, data)
+
+ # 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")
+
+ # Check whether host is determined correctly if there is a proxy
+ 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")
+
+ def test_fixpath_in_weirdurls(self):
+ # Issue4493: urllib2 to supply '/' when to urls where path does not
+ # start with'/'
+
+ h = urllib_request.AbstractHTTPHandler()
+ h.parent = MockOpener()
+
+ weird_url = 'http://www.python.org?getspam'
+ req = Request(weird_url)
+ newreq = h.do_request_(req)
+ self.assertEqual(newreq.host,'www.python.org')
+ self.assertEqual(newreq.selector,'/?getspam')
+
+ url_without_path = 'http://www.python.org'
+ req = Request(url_without_path)
+ newreq = h.do_request_(req)
+ self.assertEqual(newreq.host,'www.python.org')
+ self.assertEqual(newreq.selector,'')
+
+
+ def test_errors(self):
+ h = urllib_request.HTTPErrorProcessor()
+ o = h.parent = MockOpener()
+
+ url = "http://example.com/"
+ req = Request(url)
+ # all 2xx are passed through
+ r = MockResponse(200, "OK", {}, "", url)
+ newr = h.http_response(req, r)
+ self.assertIs(r, newr)
+ self.assertFalse(hasattr(o, "proto")) # o.error not called
+ r = MockResponse(202, "Accepted", {}, "", url)
+ newr = h.http_response(req, r)
+ self.assertIs(r, newr)
+ self.assertFalse(hasattr(o, "proto")) # o.error not called
+ r = MockResponse(206, "Partial content", {}, "", url)
+ newr = h.http_response(req, r)
+ self.assertIs(r, newr)
+ self.assertFalse(hasattr(o, "proto")) # o.error not called
+ # anything else calls o.error (and MockOpener returns None, here)
+ r = MockResponse(502, "Bad gateway", {}, "", url)
+ self.assertIsNone(h.http_response(req, r))
+ self.assertEqual(o.proto, "http") # o.error called
+ self.assertEqual(o.args, (req, r, 502, "Bad gateway", {}))
+
+ def test_cookies(self):
+ cj = MockCookieJar()
+ h = urllib_request.HTTPCookieProcessor(cj)
+ h.parent = MockOpener()
+
+ req = Request("http://example.com/")
+ r = MockResponse(200, "OK", {}, "")
+ newreq = h.http_request(req)
+ self.assertIs(cj.ach_req, req)
+ self.assertIs(cj.ach_req, newreq)
+ self.assertEqual(req.origin_req_host, "example.com")
+ self.assertFalse(req.unverifiable)
+ newr = h.http_response(req, r)
+ self.assertIs(cj.ec_req, req)
+ self.assertIs(cj.ec_r, r)
+ self.assertIs(r, newr)
+
+ def test_redirect(self):
+ from_url = "http://example.com/a.html"
+ to_url = "http://example.com/b.html"
+ h = urllib_request.HTTPRedirectHandler()
+ o = h.parent = MockOpener()
+
+ # ordinary redirect behaviour
+ for code in 301, 302, 303, 307:
+ for data in None, "blah\nblah\n":
+ method = getattr(h, "http_error_%s" % code)
+ req = Request(from_url, data)
+ req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
+ req.add_header("Nonsense", "viking=withhold")
+ if data is not None:
+ req.add_header("Content-Length", str(len(data)))
+ req.add_unredirected_header("Spam", "spam")
+ try:
+ method(req, MockFile(), code, "Blah",
+ MockHeaders({"location": to_url}))
+ except urllib_error.HTTPError:
+ # 307 in response to POST requires user OK
+ self.assertTrue(code == 307 and data is not None)
+ self.assertEqual(o.req.get_full_url(), to_url)
+ try:
+ self.assertEqual(o.req.get_method(), "GET")
+ except AttributeError:
+ self.assertFalse(o.req.data)
+
+ # now it's a GET, there should not be headers regarding content
+ # (possibly dragged from before being a POST)
+ headers = [x.lower() for x in o.req.headers]
+ self.assertNotIn("content-length", headers)
+ self.assertNotIn("content-type", headers)
+
+ self.assertEqual(o.req.headers["Nonsense"],
+ "viking=withhold")
+ self.assertNotIn("Spam", o.req.headers)
+ self.assertNotIn("Spam", o.req.unredirected_hdrs)
+
+ # loop detection
+ req = Request(from_url)
+ req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
+ def redirect(h, req, url=to_url):
+ h.http_error_302(req, MockFile(), 302, "Blah",
+ MockHeaders({"location": url}))
+ # Note that the *original* request shares the same record of
+ # redirections with the sub-requests caused by the redirections.
+
+ # detect infinite loop redirect of a URL to itself
+ req = Request(from_url, origin_req_host="example.com")
+ count = 0
+ req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
+ try:
+ while 1:
+ redirect(h, req, "http://example.com/")
+ count = count + 1
+ except urllib_error.HTTPError:
+ # don't stop until max_repeats, because cookies may introduce state
+ self.assertEqual(count, urllib_request.HTTPRedirectHandler.max_repeats)
+
+ # detect endless non-repeating chain of redirects
+ req = Request(from_url, origin_req_host="example.com")
+ count = 0
+ req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
+ try:
+ while 1:
+ redirect(h, req, "http://example.com/%d" % count)
+ count = count + 1
+ except urllib_error.HTTPError:
+ self.assertEqual(count,
+ urllib_request.HTTPRedirectHandler.max_redirections)
+
+
+ def test_invalid_redirect(self):
+ from_url = "http://example.com/a.html"
+ valid_schemes = ['http','https','ftp']
+ invalid_schemes = ['file','imap','ldap']
+ schemeless_url = "example.com/b.html"
+ h = urllib_request.HTTPRedirectHandler()
+ o = h.parent = MockOpener()
+ req = Request(from_url)
+ req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
+
+ for scheme in invalid_schemes:
+ invalid_url = scheme + '://' + schemeless_url
+ self.assertRaises(urllib_error.HTTPError, h.http_error_302,
+ req, MockFile(), 302, "Security Loophole",
+ MockHeaders({"location": invalid_url}))
+
+ for scheme in valid_schemes:
+ valid_url = scheme + '://' + schemeless_url
+ h.http_error_302(req, MockFile(), 302, "That's fine",
+ MockHeaders({"location": valid_url}))
+ self.assertEqual(o.req.get_full_url(), valid_url)
+
+ def test_relative_redirect(self):
+ from future.backports.urllib import parse as urllib_parse
+ from_url = "http://example.com/a.html"
+ relative_url = "/b.html"
+ h = urllib_request.HTTPRedirectHandler()
+ o = h.parent = MockOpener()
+ req = Request(from_url)
+ req.timeout = socket._GLOBAL_DEFAULT_TIMEOUT
+
+ valid_url = urllib_parse.urljoin(from_url,relative_url)
+ h.http_error_302(req, MockFile(), 302, "That's fine",
+ MockHeaders({"location": valid_url}))
+ self.assertEqual(o.req.get_full_url(), valid_url)
+
+ def test_cookie_redirect(self):
+ # cookies shouldn't leak into redirected requests
+ from future.backports.http.cookiejar import CookieJar
+
+ cj = CookieJar()
+ interact_netscape(cj, "http://www.example.com/", "spam=eggs")
+ hh = MockHTTPHandler(302, "Location: http://www.cracker.com/\r\n\r\n")
+ hdeh = urllib_request.HTTPDefaultErrorHandler()
+ hrh = urllib_request.HTTPRedirectHandler()
+ cp = urllib_request.HTTPCookieProcessor(cj)
+ o = build_test_opener(hh, hdeh, hrh, cp)
+ o.open("http://www.example.com/")
+ self.assertFalse(hh.req.has_header("Cookie"))
+
+ def test_redirect_fragment(self):
+ redirected_url = 'http://www.example.com/index.html#OK\r\n\r\n'
+ hh = MockHTTPHandler(302, 'Location: ' + redirected_url)
+ hdeh = urllib_request.HTTPDefaultErrorHandler()
+ hrh = urllib_request.HTTPRedirectHandler()
+ o = build_test_opener(hh, hdeh, hrh)
+ fp = o.open('http://www.example.com')
+ self.assertEqual(fp.geturl(), redirected_url.strip())
+
+ def test_proxy(self):
+ o = OpenerDirector()
+ ph = urllib_request.ProxyHandler(dict(http="proxy.example.com:3128"))
+ o.add_handler(ph)
+ meth_spec = [
+ [("http_open", "return response")]
+ ]
+ handlers = add_ordered_mock_handlers(o, meth_spec)
+
+ req = Request("http://acme.example.com/")
+ self.assertEqual(req.host, "acme.example.com")
+ o.open(req)
+ self.assertEqual(req.host, "proxy.example.com:3128")
+
+ self.assertEqual([(handlers[0], "http_open")],
+ [tup[0:2] for tup in o.calls])
+
+ def test_proxy_no_proxy(self):
+ os.environ['no_proxy'] = 'python.org'
+ o = OpenerDirector()
+ ph = urllib_request.ProxyHandler(dict(http="proxy.example.com"))
+ o.add_handler(ph)
+ req = Request("http://www.perl.org/")
+ self.assertEqual(req.host, "www.perl.org")
+ o.open(req)
+ self.assertEqual(req.host, "proxy.example.com")
+ req = Request("http://www.python.org")
+ self.assertEqual(req.host, "www.python.org")
+ o.open(req)
+ self.assertEqual(req.host, "www.python.org")
+ del os.environ['no_proxy']
+
+ def test_proxy_no_proxy_all(self):
+ os.environ['no_proxy'] = '*'
+ o = OpenerDirector()
+ ph = urllib_request.ProxyHandler(dict(http="proxy.example.com"))
+ o.add_handler(ph)
+ req = Request("http://www.python.org")
+ self.assertEqual(req.host, "www.python.org")
+ o.open(req)
+ self.assertEqual(req.host, "www.python.org")
+ del os.environ['no_proxy']
+
+
+ def test_proxy_https(self):
+ o = OpenerDirector()
+ ph = urllib_request.ProxyHandler(dict(https="proxy.example.com:3128"))
+ o.add_handler(ph)
+ meth_spec = [
+ [("https_open", "return response")]
+ ]
+ handlers = add_ordered_mock_handlers(o, meth_spec)
+
+ req = Request("https://www.example.com/")
+ self.assertEqual(req.host, "www.example.com")
+ o.open(req)
+ self.assertEqual(req.host, "proxy.example.com:3128")
+ self.assertEqual([(handlers[0], "https_open")],
+ [tup[0:2] for tup in o.calls])
+
+ def test_proxy_https_proxy_authorization(self):
+ o = OpenerDirector()
+ ph = urllib_request.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.host, "www.example.com")
+ self.assertIsNone(req._tunnel_host)
+ 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.assertNotIn(("Proxy-Authorization","FooBar"),
+ https_handler.httpconn.req_headers)
+ self.assertIn(("User-Agent","Grail"),
+ https_handler.httpconn.req_headers)
+ self.assertIsNotNone(req._tunnel_host)
+ self.assertEqual(req.host, "proxy.example.com:3128")
+ self.assertEqual(req.get_header("Proxy-authorization"),"FooBar")
+
+ # TODO: This should be only for OSX
+ @unittest.skipUnless(sys.platform == 'darwin', "only relevant for OSX")
+ def test_osx_proxy_bypass(self):
+ bypass = {
+ 'exclude_simple': False,
+ 'exceptions': ['foo.bar', '*.bar.com', '127.0.0.1', '10.10',
+ '10.0/16']
+ }
+ # Check hosts that should trigger the proxy bypass
+ for host in ('foo.bar', 'www.bar.com', '127.0.0.1', '10.10.0.1',
+ '10.0.0.1'):
+ self.assertTrue(_proxy_bypass_macosx_sysconf(host, bypass),
+ 'expected bypass of %s to be True' % host)
+ # Check hosts that should not trigger the proxy bypass
+ for host in ('abc.foo.bar', 'bar.com', '127.0.0.2', '10.11.0.1', 'test'):
+ self.assertFalse(_proxy_bypass_macosx_sysconf(host, bypass),
+ 'expected bypass of %s to be False' % host)
+
+ # Check the exclude_simple flag
+ bypass = {'exclude_simple': True, 'exceptions': []}
+ self.assertTrue(_proxy_bypass_macosx_sysconf('test', bypass))
+
+ def test_basic_auth(self, quote_char='"'):
+ opener = OpenerDirector()
+ password_manager = MockPasswordManager()
+ auth_handler = urllib_request.HTTPBasicAuthHandler(password_manager)
+ realm = "ACME Widget Store"
+ http_handler = MockHTTPHandler(
+ 401, 'WWW-Authenticate: Basic realm=%s%s%s\r\n\r\n' %
+ (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="'")
+
+ def test_basic_auth_with_unquoted_realm(self):
+ opener = OpenerDirector()
+ password_manager = MockPasswordManager()
+ auth_handler = urllib_request.HTTPBasicAuthHandler(password_manager)
+ realm = "ACME Widget Store"
+ http_handler = MockHTTPHandler(
+ 401, 'WWW-Authenticate: Basic realm=%s\r\n\r\n' % realm)
+ opener.add_handler(auth_handler)
+ opener.add_handler(http_handler)
+ with self.assertWarns(UserWarning):
+ 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_proxy_basic_auth(self):
+ opener = OpenerDirector()
+ ph = urllib_request.ProxyHandler(dict(http="proxy.example.com:3128"))
+ opener.add_handler(ph)
+ password_manager = MockPasswordManager()
+ auth_handler = urllib_request.ProxyBasicAuthHandler(password_manager)
+ realm = "ACME Networks"
+ http_handler = MockHTTPHandler(
+ 407, 'Proxy-Authenticate: Basic realm="%s"\r\n\r\n' % realm)
+ opener.add_handler(auth_handler)
+ opener.add_handler(http_handler)
+ self._test_basic_auth(opener, auth_handler, "Proxy-authorization",
+ realm, http_handler, password_manager,
+ "http://acme.example.com:3128/protected",
+ "proxy.example.com:3128",
+ )
+
+ def test_basic_and_digest_auth_handlers(self):
+ # HTTPDigestAuthHandler raised an exception if it couldn't handle a 40*
+ # response (http://python.org/sf/1479302), where it should instead
+ # return None to allow another handler (especially
+ # HTTPBasicAuthHandler) to handle the response.
+
+ # Also (http://python.org/sf/14797027, RFC 2617 section 1.2), we must
+ # try digest first (since it's the strongest auth scheme), so we record
+ # order of calls here to check digest comes first:
+ class RecordingOpenerDirector(OpenerDirector):
+ def __init__(self):
+ OpenerDirector.__init__(self)
+ self.recorded = []
+ def record(self, info):
+ self.recorded.append(info)
+ class TestDigestAuthHandler(urllib_request.HTTPDigestAuthHandler):
+ def http_error_401(self, *args, **kwds):
+ self.parent.record("digest")
+ urllib_request.HTTPDigestAuthHandler.http_error_401(self,
+ *args, **kwds)
+ class TestBasicAuthHandler(urllib_request.HTTPBasicAuthHandler):
+ def http_error_401(self, *args, **kwds):
+ self.parent.record("basic")
+ urllib_request.HTTPBasicAuthHandler.http_error_401(self,
+ *args, **kwds)
+
+ opener = RecordingOpenerDirector()
+ password_manager = MockPasswordManager()
+ digest_handler = TestDigestAuthHandler(password_manager)
+ basic_handler = TestBasicAuthHandler(password_manager)
+ realm = "ACME Networks"
+ http_handler = MockHTTPHandler(
+ 401, 'WWW-Authenticate: Basic realm="%s"\r\n\r\n' % realm)
+ opener.add_handler(basic_handler)
+ opener.add_handler(digest_handler)
+ opener.add_handler(http_handler)
+
+ # check basic auth isn't blocked by digest handler failing
+ self._test_basic_auth(opener, basic_handler, "Authorization",
+ 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)
+
+ def test_unsupported_auth_digest_handler(self):
+ opener = OpenerDirector()
+ # While using DigestAuthHandler
+ digest_auth_handler = urllib_request.HTTPDigestAuthHandler(None)
+ http_handler = MockHTTPHandler(
+ 401, 'WWW-Authenticate: Kerberos\r\n\r\n')
+ opener.add_handler(digest_auth_handler)
+ opener.add_handler(http_handler)
+ self.assertRaises(ValueError,opener.open,"http://www.example.com")
+
+ def test_unsupported_auth_basic_handler(self):
+ # While using BasicAuthHandler
+ opener = OpenerDirector()
+ basic_auth_handler = urllib_request.HTTPBasicAuthHandler(None)
+ http_handler = MockHTTPHandler(
+ 401, 'WWW-Authenticate: NTLM\r\n\r\n')
+ opener.add_handler(basic_auth_handler)
+ opener.add_handler(http_handler)
+ self.assertRaises(ValueError,opener.open,"http://www.example.com")
+
+ 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
+ auth_handler.add_password(realm, request_url, user, password)
+ self.assertEqual(realm, password_manager.realm)
+ self.assertEqual(request_url, password_manager.url)
+ self.assertEqual(user, password_manager.user)
+ self.assertEqual(password, password_manager.password)
+
+ opener.open(request_url)
+
+ # should have asked the password manager for the username/password
+ self.assertEqual(password_manager.target_realm, realm)
+ self.assertEqual(password_manager.target_url, protected_url)
+
+ # expect one request without authorization, then one with
+ self.assertEqual(len(http_handler.requests), 2)
+ self.assertFalse(http_handler.requests[0].has_header(auth_header))
+ userpass = bytes('%s:%s' % (user, password), "ascii")
+ auth_hdr_value = ('Basic ' +
+ base64.encodebytes(userpass).strip().decode())
+ self.assertEqual(http_handler.requests[1].get_header(auth_header),
+ auth_hdr_value)
+ self.assertEqual(http_handler.requests[1].unredirected_hdrs[auth_header],
+ auth_hdr_value)
+ # if the password manager can't find a password, the handler won't
+ # handle the HTTP auth error
+ password_manager.user = password_manager.password = None
+ http_handler.reset()
+ opener.open(request_url)
+ self.assertEqual(len(http_handler.requests), 1)
+ self.assertFalse(http_handler.requests[0].has_header(auth_header))
+
+
+class MiscTests(unittest.TestCase):
+
+ def opener_has_handler(self, opener, handler_class):
+ self.assertTrue(any(h.__class__ == handler_class
+ for h in opener.handlers))
+
+ def test_build_opener(self):
+ class MyHTTPHandler(urllib_request.HTTPHandler): pass
+ class FooHandler(urllib_request.BaseHandler):
+ def foo_open(self): pass
+ class BarHandler(urllib_request.BaseHandler):
+ def bar_open(self): pass
+
+ build_opener = urllib_request.build_opener
+
+ o = build_opener(FooHandler, BarHandler)
+ self.opener_has_handler(o, FooHandler)
+ self.opener_has_handler(o, BarHandler)
+
+ # can take a mix of classes and instances
+ o = build_opener(FooHandler, BarHandler())
+ self.opener_has_handler(o, FooHandler)
+ self.opener_has_handler(o, BarHandler)
+
+ # subclasses of default handlers override default handlers
+ o = build_opener(MyHTTPHandler)
+ self.opener_has_handler(o, MyHTTPHandler)
+
+ # a particular case of overriding: default handlers can be passed
+ # in explicitly
+ o = build_opener()
+ self.opener_has_handler(o, urllib_request.HTTPHandler)
+ o = build_opener(urllib_request.HTTPHandler)
+ self.opener_has_handler(o, urllib_request.HTTPHandler)
+ o = build_opener(urllib_request.HTTPHandler())
+ self.opener_has_handler(o, urllib_request.HTTPHandler)
+
+ # Issue2670: multiple handlers sharing the same base class
+ class MyOtherHTTPHandler(urllib_request.HTTPHandler): pass
+ o = build_opener(MyHTTPHandler, MyOtherHTTPHandler)
+ self.opener_has_handler(o, MyHTTPHandler)
+ self.opener_has_handler(o, MyOtherHTTPHandler)
+
+ def test_HTTPError_interface(self):
+ """
+ Issue 13211 reveals that HTTPError didn't implement the URLError
+ interface even though HTTPError is a subclass of URLError.
+ """
+ msg = 'something bad happened'
+ url = code = fp = None
+ hdrs = 'Content-Length: 42'
+ err = urllib_error.HTTPError(url, code, msg, hdrs, fp)
+ self.assertTrue(hasattr(err, 'reason'))
+ self.assertEqual(err.reason, 'something bad happened')
+ self.assertTrue(hasattr(err, 'hdrs'))
+ self.assertEqual(err.hdrs, 'Content-Length: 42')
+ expected_errmsg = 'HTTP Error %s: %s' % (err.code, err.msg)
+ self.assertEqual(str(err), expected_errmsg)
+
+
+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"})
+
+ def test_method(self):
+ self.assertEqual("POST", self.post.get_method())
+ self.assertEqual("GET", self.get.get_method())
+
+ def test_data(self):
+ self.assertFalse(self.get.data)
+ self.assertEqual("GET", self.get.get_method())
+ self.get.data = "spam"
+ self.assertTrue(self.get.data)
+ self.assertEqual("POST", self.get.get_method())
+
+ def test_get_full_url(self):
+ self.assertEqual("http://www.python.org/~jeremy/",
+ self.get.get_full_url())
+
+ def test_selector(self):
+ self.assertEqual("/~jeremy/", self.get.selector)
+ req = Request("http://www.python.org/")
+ self.assertEqual("/", req.selector)
+
+ def test_get_type(self):
+ self.assertEqual("http", self.get.type)
+
+ def test_get_host(self):
+ self.assertEqual("www.python.org", self.get.host)
+
+ def test_get_host_unquote(self):
+ req = Request("http://www.%70ython.org/")
+ self.assertEqual("www.python.org", req.host)
+
+ def test_proxy(self):
+ self.assertFalse(self.get.has_proxy())
+ self.get.set_proxy("www.perl.org", "http")
+ self.assertTrue(self.get.has_proxy())
+ self.assertEqual("www.python.org", self.get.origin_req_host)
+ self.assertEqual("www.perl.org", self.get.host)
+
+ def test_wrapped_url(self):
+ req = Request("<URL:http://www.python.org>")
+ self.assertEqual("www.python.org", req.host)
+
+ def test_url_fragment(self):
+ req = Request("http://www.python.org/?qs=query#fragment=true")
+ self.assertEqual("/?qs=query", req.selector)
+ req = Request("http://www.python.org/#fun=true")
+ self.assertEqual("/", req.selector)
+
+ # Issue 11703: geturl() omits fragment in the original URL.
+ url = 'http://docs.python.org/library/urllib2.html#OK'
+ req = Request(url)
+ self.assertEqual(req.get_full_url(), url)
+
+ def test_HTTPError_interface_call(self):
+ """
+ Issue 15701 - HTTPError interface has info method available from URLError
+ """
+ err = urllib_request.HTTPError(msg="something bad happened", url=None,
+ code=None, hdrs='Content-Length:42', fp=None)
+ self.assertTrue(hasattr(err, 'reason'))
+ assert hasattr(err, 'reason')
+ assert hasattr(err, 'info')
+ assert callable(err.info)
+ try:
+ err.info()
+ except AttributeError:
+ self.fail('err.info call failed.')
+ self.assertEqual(err.info(), "Content-Length:42")
+
+def test_main(verbose=None):
+ # support.run_doctest(test_urllib2, verbose)
+ # support.run_doctest(urllib_request, verbose)
+ tests = (TrivialTests,
+ OpenerDirectorTests,
+ HandlerTests,
+ MiscTests,
+ RequestTests,
+ RequestHdrsTests)
+ support.run_unittest(*tests)
+
+if __name__ == "__main__":
+ unittest.main()