diff options
Diffstat (limited to 'Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py')
| -rw-r--r-- | Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py | 1569 |
1 files changed, 0 insertions, 1569 deletions
diff --git a/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py b/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py deleted file mode 100644 index 2d69dad..0000000 --- a/Python/Dependencies/future-0.18.2/tests/test_future/test_urllib2.py +++ /dev/null @@ -1,1569 +0,0 @@ -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() |