0001 import unittest 0002 from test import test_support 0003 0004 import os, socket 0005 import StringIO 0006 0007 import urllib2 0008 from urllib2 import Request, OpenerDirector 0009 0010 # XXX 0011 # Request 0012 # CacheFTPHandler (hard to write) 0013 # parse_keqv_list, parse_http_list (I'm leaving this for Anthony Baxter 0014 # and Greg Stein, since they're doing Digest Authentication) 0015 # Authentication stuff (ditto) 0016 # ProxyHandler, CustomProxy, CustomProxyHandler (I don't use a proxy) 0017 # GopherHandler (haven't used gopher for a decade or so...) 0018 0019 class TrivialTests(unittest.TestCase): 0020 def test_trivial(self): 0021 # A couple trivial tests 0022 0023 self.assertRaises(ValueError, urllib2.urlopen, 'bogus url') 0024 0025 # XXX Name hacking to get this to work on Windows. 0026 fname = os.path.abspath(urllib2.__file__).replace('\\', '/') 0027 if fname[1:2] == ":": 0028 fname = fname[2:] 0029 # And more hacking to get it to work on MacOS. This assumes 0030 # urllib.pathname2url works, unfortunately... 0031 if os.name == 'mac': 0032 fname = '/' + fname.replace(':', '/') 0033 elif os.name == 'riscos': 0034 import string 0035 fname = os.expand(fname) 0036 fname = fname.translate(string.maketrans("/.", "./")) 0037 0038 file_url = "file://%s" % fname 0039 f = urllib2.urlopen(file_url) 0040 0041 buf = f.read() 0042 f.close() 0043 0044 0045 class MockOpener: 0046 addheaders = [] 0047 def open(self, req, data=None): 0048 self.req, self.data = req, data 0049 def error(self, proto, *args): 0050 self.proto, self.args = proto, args 0051 0052 class MockFile: 0053 def read(self, count=None): pass 0054 def readline(self, count=None): pass 0055 def close(self): pass 0056 0057 class MockHeaders(dict): 0058 def getheaders(self, name): 0059 return self.values() 0060 0061 class MockResponse(StringIO.StringIO): 0062 def __init__(self, code, msg, headers, data, url=None): 0063 StringIO.StringIO.__init__(self, data) 0064 self.code, self.msg, self.headers, self.url = code, msg, headers, url 0065 def info(self): 0066 return self.headers 0067 def geturl(self): 0068 return self.url 0069 0070 class MockCookieJar: 0071 def add_cookie_header(self, request): 0072 self.ach_req = request 0073 def extract_cookies(self, response, request): 0074 self.ec_req, self.ec_r = request, response 0075 0076 class FakeMethod: 0077 def __init__(self, meth_name, action, handle): 0078 self.meth_name = meth_name 0079 self.handle = handle 0080 self.action = action 0081 def __call__(self, *args): 0082 return self.handle(self.meth_name, self.action, *args) 0083 0084 class MockHandler: 0085 def __init__(self, methods): 0086 self._define_methods(methods) 0087 def _define_methods(self, methods): 0088 for spec in methods: 0089 if len(spec) == 2: name, action = spec 0090 else: name, action = spec, None 0091 meth = FakeMethod(name, action, self.handle) 0092 setattr(self.__class__, name, meth) 0093 def handle(self, fn_name, action, *args, **kwds): 0094 self.parent.calls.append((self, fn_name, args, kwds)) 0095 if action is None: 0096 return None 0097 elif action == "return self": 0098 return self 0099 elif action == "return response": 0100 res = MockResponse(200, "OK", {}, "") 0101 return res 0102 elif action == "return request": 0103 return Request("http://blah/") 0104 elif action.startswith("error"): 0105 code = action[action.rfind(" ")+1:] 0106 try: 0107 code = int(code) 0108 except ValueError: 0109 pass 0110 res = MockResponse(200, "OK", {}, "") 0111 return self.parent.error("http", args[0], res, code, "", {}) 0112 elif action == "raise": 0113 raise urllib2.URLError("blah") 0114 assert False 0115 def close(self): pass 0116 def add_parent(self, parent): 0117 self.parent = parent 0118 self.parent.calls = [] 0119 def __lt__(self, other): 0120 if not hasattr(other, "handler_order"): 0121 # No handler_order, leave in original order. Yuck. 0122 return True 0123 return self.handler_order < other.handler_order 0124 0125 def add_ordered_mock_handlers(opener, meth_spec): 0126 """Create MockHandlers and add them to an OpenerDirector. 0127 0128 meth_spec: list of lists of tuples and strings defining methods to define 0129 on handlers. eg: 0130 0131 [["http_error", "ftp_open"], ["http_open"]] 0132 0133 defines methods .http_error() and .ftp_open() on one handler, and 0134 .http_open() on another. These methods just record their arguments and 0135 return None. Using a tuple instead of a string causes the method to 0136 perform some action (see MockHandler.handle()), eg: 0137 0138 [["http_error"], [("http_open", "return request")]] 0139 0140 defines .http_error() on one handler (which simply returns None), and 0141 .http_open() on another handler, which returns a Request object. 0142 0143 """ 0144 handlers = [] 0145 count = 0 0146 for meths in meth_spec: 0147 class MockHandlerSubclass(MockHandler): pass 0148 h = MockHandlerSubclass(meths) 0149 h.handler_order = count 0150 h.add_parent(opener) 0151 count = count + 1 0152 handlers.append(h) 0153 opener.add_handler(h) 0154 return handlers 0155 0156 class OpenerDirectorTests(unittest.TestCase): 0157 0158 def test_handled(self): 0159 # handler returning non-None means no more handlers will be called 0160 o = OpenerDirector() 0161 meth_spec = [ 0162 ["http_open", "ftp_open", "http_error_302"], 0163 ["ftp_open"], 0164 [("http_open", "return self")], 0165 [("http_open", "return self")], 0166 ] 0167 handlers = add_ordered_mock_handlers(o, meth_spec) 0168 0169 req = Request("http://example.com/") 0170 r = o.open(req) 0171 # Second .http_open() gets called, third doesn't, since second returned 0172 # non-None. Handlers without .http_open() never get any methods called 0173 # on them. 0174 # In fact, second mock handler defining .http_open() returns self 0175 # (instead of response), which becomes the OpenerDirector's return 0176 # value. 0177 self.assertEqual(r, handlers[2]) 0178 calls = [(handlers[0], "http_open"), (handlers[2], "http_open")] 0179 for expected, got in zip(calls, o.calls): 0180 handler, name, args, kwds = got 0181 self.assertEqual((handler, name), expected) 0182 self.assertEqual(args, (req,)) 0183 0184 def test_handler_order(self): 0185 o = OpenerDirector() 0186 handlers = [] 0187 for meths, handler_order in [ 0188 ([("http_open", "return self")], 500), 0189 (["http_open"], 0), 0190 ]: 0191 class MockHandlerSubclass(MockHandler): pass 0192 h = MockHandlerSubclass(meths) 0193 h.handler_order = handler_order 0194 handlers.append(h) 0195 o.add_handler(h) 0196 0197 r = o.open("http://example.com/") 0198 # handlers called in reverse order, thanks to their sort order 0199 self.assertEqual(o.calls[0][0], handlers[1]) 0200 self.assertEqual(o.calls[1][0], handlers[0]) 0201 0202 def test_raise(self): 0203 # raising URLError stops processing of request 0204 o = OpenerDirector() 0205 meth_spec = [ 0206 [("http_open", "raise")], 0207 [("http_open", "return self")], 0208 ] 0209 handlers = add_ordered_mock_handlers(o, meth_spec) 0210 0211 req = Request("http://example.com/") 0212 self.assertRaises(urllib2.URLError, o.open, req) 0213 self.assertEqual(o.calls, [(handlers[0], "http_open", (req,), {})]) 0214 0215 ## def test_error(self): 0216 ## # XXX this doesn't actually seem to be used in standard library, 0217 ## # but should really be tested anyway... 0218 0219 def test_http_error(self): 0220 # XXX http_error_default 0221 # http errors are a special case 0222 o = OpenerDirector() 0223 meth_spec = [ 0224 [("http_open", "error 302")], 0225 [("http_error_400", "raise"), "http_open"], 0226 [("http_error_302", "return response"), "http_error_303", 0227 "http_error"], 0228 [("http_error_302")], 0229 ] 0230 handlers = add_ordered_mock_handlers(o, meth_spec) 0231 0232 class Unknown: 0233 def __eq__(self, other): return True 0234 0235 req = Request("http://example.com/") 0236 r = o.open(req) 0237 assert len(o.calls) == 2 0238 calls = [(handlers[0], "http_open", (req,)), 0239 (handlers[2], "http_error_302", 0240 (req, Unknown(), 302, "", {}))] 0241 for expected, got in zip(calls, o.calls): 0242 handler, method_name, args = expected 0243 self.assertEqual((handler, method_name), got[:2]) 0244 self.assertEqual(args, got[2]) 0245 0246 def test_processors(self): 0247 # *_request / *_response methods get called appropriately 0248 o = OpenerDirector() 0249 meth_spec = [ 0250 [("http_request", "return request"), 0251 ("http_response", "return response")], 0252 [("http_request", "return request"), 0253 ("http_response", "return response")], 0254 ] 0255 handlers = add_ordered_mock_handlers(o, meth_spec) 0256 0257 req = Request("http://example.com/") 0258 r = o.open(req) 0259 # processor methods are called on *all* handlers that define them, 0260 # not just the first handler that handles the request 0261 calls = [ 0262 (handlers[0], "http_request"), (handlers[1], "http_request"), 0263 (handlers[0], "http_response"), (handlers[1], "http_response")] 0264 0265 for i, (handler, name, args, kwds) in enumerate(o.calls): 0266 if i < 2: 0267 # *_request 0268 self.assertEqual((handler, name), calls[i]) 0269 self.assertEqual(len(args), 1) 0270 self.assert_(isinstance(args[0], Request)) 0271 else: 0272 # *_response 0273 self.assertEqual((handler, name), calls[i]) 0274 self.assertEqual(len(args), 2) 0275 self.assert_(isinstance(args[0], Request)) 0276 # response from opener.open is None, because there's no 0277 # handler that defines http_open to handle it 0278 self.assert_(args[1] is None or 0279 isinstance(args[1], MockResponse)) 0280 0281 0282 def sanepathname2url(path): 0283 import urllib 0284 urlpath = urllib.pathname2url(path) 0285 if os.name == "nt" and urlpath.startswith("///"): 0286 urlpath = urlpath[2:] 0287 # XXX don't ask me about the mac... 0288 return urlpath 0289 0290 class HandlerTests(unittest.TestCase): 0291 0292 def test_ftp(self): 0293 class MockFTPWrapper: 0294 def __init__(self, data): self.data = data 0295 def retrfile(self, filename, filetype): 0296 self.filename, self.filetype = filename, filetype 0297 return StringIO.StringIO(self.data), len(self.data) 0298 0299 class NullFTPHandler(urllib2.FTPHandler): 0300 def __init__(self, data): self.data = data 0301 def connect_ftp(self, user, passwd, host, port, dirs): 0302 self.user, self.passwd = user, passwd 0303 self.host, self.port = host, port 0304 self.dirs = dirs 0305 self.ftpwrapper = MockFTPWrapper(self.data) 0306 return self.ftpwrapper 0307 0308 import ftplib, socket 0309 data = "rheum rhaponicum" 0310 h = NullFTPHandler(data) 0311 o = h.parent = MockOpener() 0312 0313 for url, host, port, type_, dirs, filename, mimetype in [ 0314 ("ftp://localhost/foo/bar/baz.html", 0315 "localhost", ftplib.FTP_PORT, "I", 0316 ["foo", "bar"], "baz.html", "text/html"), 0317 ("ftp://localhost:80/foo/bar/", 0318 "localhost", 80, "D", 0319 ["foo", "bar"], "", None), 0320 ("ftp://localhost/baz.gif;type=a", 0321 "localhost", ftplib.FTP_PORT, "A", 0322 [], "baz.gif", None), # XXX really this should guess image/gif 0323 ]: 0324 r = h.ftp_open(Request(url)) 0325 # ftp authentication not yet implemented by FTPHandler 0326 self.assert_(h.user == h.passwd == "") 0327 self.assertEqual(h.host, socket.gethostbyname(host)) 0328 self.assertEqual(h.port, port) 0329 self.assertEqual(h.dirs, dirs) 0330 self.assertEqual(h.ftpwrapper.filename, filename) 0331 self.assertEqual(h.ftpwrapper.filetype, type_) 0332 headers = r.info() 0333 self.assertEqual(headers.get("Content-type"), mimetype) 0334 self.assertEqual(int(headers["Content-length"]), len(data)) 0335 0336 def test_file(self): 0337 import time, rfc822, socket 0338 h = urllib2.FileHandler() 0339 o = h.parent = MockOpener() 0340 0341 TESTFN = test_support.TESTFN 0342 urlpath = sanepathname2url(os.path.abspath(TESTFN)) 0343 towrite = "hello, world\n" 0344 for url in [ 0345 "file://localhost%s" % urlpath, 0346 "file://%s" % urlpath, 0347 "file://%s%s" % (socket.gethostbyname('localhost'), urlpath), 0348 "file://%s%s" % (socket.gethostbyname(socket.gethostname()), 0349 urlpath), 0350 ]: 0351 f = open(TESTFN, "wb") 0352 try: 0353 try: 0354 f.write(towrite) 0355 finally: 0356 f.close() 0357 0358 r = h.file_open(Request(url)) 0359 try: 0360 data = r.read() 0361 headers = r.info() 0362 newurl = r.geturl() 0363 finally: 0364 r.close() 0365 stats = os.stat(TESTFN) 0366 modified = rfc822.formatdate(stats.st_mtime) 0367 finally: 0368 os.remove(TESTFN) 0369 self.assertEqual(data, towrite) 0370 self.assertEqual(headers["Content-type"], "text/plain") 0371 self.assertEqual(headers["Content-length"], "13") 0372 self.assertEqual(headers["Last-modified"], modified) 0373 0374 for url in [ 0375 "file://localhost:80%s" % urlpath, 0376 # XXXX bug: these fail with socket.gaierror, should be URLError 0377 ## "file://%s:80%s/%s" % (socket.gethostbyname('localhost'), 0378 ## os.getcwd(), TESTFN), 0379 ## "file://somerandomhost.ontheinternet.com%s/%s" % 0380 ## (os.getcwd(), TESTFN), 0381 ]: 0382 try: 0383 f = open(TESTFN, "wb") 0384 try: 0385 f.write(towrite) 0386 finally: 0387 f.close() 0388 0389 self.assertRaises(urllib2.URLError, 0390 h.file_open, Request(url)) 0391 finally: 0392 os.remove(TESTFN) 0393 0394 h = urllib2.FileHandler() 0395 o = h.parent = MockOpener() 0396 # XXXX why does // mean ftp (and /// mean not ftp!), and where 0397 # is file: scheme specified? I think this is really a bug, and 0398 # what was intended was to distinguish between URLs like: 0399 # file:/blah.txt (a file) 0400 # file://localhost/blah.txt (a file) 0401 # file:///blah.txt (a file) 0402 # file://ftp.example.com/blah.txt (an ftp URL) 0403 for url, ftp in [ 0404 ("file://ftp.example.com//foo.txt", True), 0405 ("file://ftp.example.com///foo.txt", False), 0406 # XXXX bug: fails with OSError, should be URLError 0407 ("file://ftp.example.com/foo.txt", False), 0408 ]: 0409 req = Request(url) 0410 try: 0411 h.file_open(req) 0412 # XXXX remove OSError when bug fixed 0413 except (urllib2.URLError, OSError): 0414 self.assert_(not ftp) 0415 else: 0416 self.assert_(o.req is req) 0417 self.assertEqual(req.type, "ftp") 0418 0419 def test_http(self): 0420 class MockHTTPResponse: 0421 def __init__(self, fp, msg, status, reason): 0422 self.fp = fp 0423 self.msg = msg 0424 self.status = status 0425 self.reason = reason 0426 def read(self): 0427 return '' 0428 class MockHTTPClass: 0429 def __init__(self): 0430 self.req_headers = [] 0431 self.data = None 0432 self.raise_on_endheaders = False 0433 def __call__(self, host): 0434 self.host = host 0435 return self 0436 def set_debuglevel(self, level): 0437 self.level = level 0438 def request(self, method, url, body=None, headers={}): 0439 self.method = method 0440 self.selector = url 0441 self.req_headers += headers.items() 0442 if body: 0443 self.data = body 0444 if self.raise_on_endheaders: 0445 import socket 0446 raise socket.error() 0447 def getresponse(self): 0448 return MockHTTPResponse(MockFile(), {}, 200, "OK") 0449 0450 h = urllib2.AbstractHTTPHandler() 0451 o = h.parent = MockOpener() 0452 0453 url = "http://example.com/" 0454 for method, data in [("GET", None), ("POST", "blah")]: 0455 req = Request(url, data, {"Foo": "bar"}) 0456 req.add_unredirected_header("Spam", "eggs") 0457 http = MockHTTPClass() 0458 r = h.do_open(http, req) 0459 0460 # result attributes 0461 r.read; r.readline # wrapped MockFile methods 0462 r.info; r.geturl # addinfourl methods 0463 r.code, r.msg == 200, "OK" # added from MockHTTPClass.getreply() 0464 hdrs = r.info() 0465 hdrs.get; hdrs.has_key # r.info() gives dict from .getreply() 0466 self.assertEqual(r.geturl(), url) 0467 0468 self.assertEqual(http.host, "example.com") 0469 self.assertEqual(http.level, 0) 0470 self.assertEqual(http.method, method) 0471 self.assertEqual(http.selector, "/") 0472 self.assertEqual(http.req_headers, 0473 [("Connection", "close"), 0474 ("Foo", "bar"), ("Spam", "eggs")]) 0475 self.assertEqual(http.data, data) 0476 0477 # check socket.error converted to URLError 0478 http.raise_on_endheaders = True 0479 self.assertRaises(urllib2.URLError, h.do_open, http, req) 0480 0481 # check adding of standard headers 0482 o.addheaders = [("Spam", "eggs")] 0483 for data in "", None: # POST, GET 0484 req = Request("http://example.com/", data) 0485 r = MockResponse(200, "OK", {}, "") 0486 newreq = h.do_request_(req) 0487 if data is None: # GET 0488 self.assert_("Content-length" not in req.unredirected_hdrs) 0489 self.assert_("Content-type" not in req.unredirected_hdrs) 0490 else: # POST 0491 self.assertEqual(req.unredirected_hdrs["Content-length"], "0") 0492 self.assertEqual(req.unredirected_hdrs["Content-type"], 0493 "application/x-www-form-urlencoded") 0494 # XXX the details of Host could be better tested 0495 self.assertEqual(req.unredirected_hdrs["Host"], "example.com") 0496 self.assertEqual(req.unredirected_hdrs["Spam"], "eggs") 0497 0498 # don't clobber existing headers 0499 req.add_unredirected_header("Content-length", "foo") 0500 req.add_unredirected_header("Content-type", "bar") 0501 req.add_unredirected_header("Host", "baz") 0502 req.add_unredirected_header("Spam", "foo") 0503 newreq = h.do_request_(req) 0504 self.assertEqual(req.unredirected_hdrs["Content-length"], "foo") 0505 self.assertEqual(req.unredirected_hdrs["Content-type"], "bar") 0506 self.assertEqual(req.unredirected_hdrs["Host"], "baz") 0507 self.assertEqual(req.unredirected_hdrs["Spam"], "foo") 0508 0509 def test_errors(self): 0510 h = urllib2.HTTPErrorProcessor() 0511 o = h.parent = MockOpener() 0512 0513 url = "http://example.com/" 0514 req = Request(url) 0515 # 200 OK is passed through 0516 r = MockResponse(200, "OK", {}, "", url) 0517 newr = h.http_response(req, r) 0518 self.assert_(r is newr) 0519 self.assert_(not hasattr(o, "proto")) # o.error not called 0520 # anything else calls o.error (and MockOpener returns None, here) 0521 r = MockResponse(201, "Created", {}, "", url) 0522 self.assert_(h.http_response(req, r) is None) 0523 self.assertEqual(o.proto, "http") # o.error called 0524 self.assertEqual(o.args, (req, r, 201, "Created", {})) 0525 0526 def test_cookies(self): 0527 cj = MockCookieJar() 0528 h = urllib2.HTTPCookieProcessor(cj) 0529 o = h.parent = MockOpener() 0530 0531 req = Request("http://example.com/") 0532 r = MockResponse(200, "OK", {}, "") 0533 newreq = h.http_request(req) 0534 self.assert_(cj.ach_req is req is newreq) 0535 self.assertEquals(req.get_origin_req_host(), "example.com") 0536 self.assert_(not req.is_unverifiable()) 0537 newr = h.http_response(req, r) 0538 self.assert_(cj.ec_req is req) 0539 self.assert_(cj.ec_r is r is newr) 0540 0541 def test_redirect(self): 0542 from_url = "http://example.com/a.html" 0543 to_url = "http://example.com/b.html" 0544 h = urllib2.HTTPRedirectHandler() 0545 o = h.parent = MockOpener() 0546 0547 # ordinary redirect behaviour 0548 for code in 301, 302, 303, 307: 0549 for data in None, "blah\nblah\n": 0550 method = getattr(h, "http_error_%s" % code) 0551 req = Request(from_url, data) 0552 req.add_header("Nonsense", "viking=withhold") 0553 req.add_unredirected_header("Spam", "spam") 0554 try: 0555 method(req, MockFile(), code, "Blah", 0556 MockHeaders({"location": to_url})) 0557 except urllib2.HTTPError: 0558 # 307 in response to POST requires user OK 0559 self.assert_(code == 307 and data is not None) 0560 self.assertEqual(o.req.get_full_url(), to_url) 0561 try: 0562 self.assertEqual(o.req.get_method(), "GET") 0563 except AttributeError: 0564 self.assert_(not o.req.has_data()) 0565 self.assertEqual(o.req.headers["Nonsense"], 0566 "viking=withhold") 0567 self.assert_("Spam" not in o.req.headers) 0568 self.assert_("Spam" not in o.req.unredirected_hdrs) 0569 0570 # loop detection 0571 req = Request(from_url) 0572 def redirect(h, req, url=to_url): 0573 h.http_error_302(req, MockFile(), 302, "Blah", 0574 MockHeaders({"location": url})) 0575 # Note that the *original* request shares the same record of 0576 # redirections with the sub-requests caused by the redirections. 0577 0578 # detect infinite loop redirect of a URL to itself 0579 req = Request(from_url, origin_req_host="example.com") 0580 count = 0 0581 try: 0582 while 1: 0583 redirect(h, req, "http://example.com/") 0584 count = count + 1 0585 except urllib2.HTTPError: 0586 # don't stop until max_repeats, because cookies may introduce state 0587 self.assertEqual(count, urllib2.HTTPRedirectHandler.max_repeats) 0588 0589 # detect endless non-repeating chain of redirects 0590 req = Request(from_url, origin_req_host="example.com") 0591 count = 0 0592 try: 0593 while 1: 0594 redirect(h, req, "http://example.com/%d" % count) 0595 count = count + 1 0596 except urllib2.HTTPError: 0597 self.assertEqual(count, 0598 urllib2.HTTPRedirectHandler.max_redirections) 0599 0600 def test_cookie_redirect(self): 0601 class MockHTTPHandler(urllib2.HTTPHandler): 0602 def __init__(self): self._count = 0 0603 def http_open(self, req): 0604 import mimetools 0605 from StringIO import StringIO 0606 if self._count == 0: 0607 self._count = self._count + 1 0608 msg = mimetools.Message( 0609 StringIO("Location: http://www.cracker.com/\r\n\r\n")) 0610 return self.parent.error( 0611 "http", req, MockFile(), 302, "Found", msg) 0612 else: 0613 self.req = req 0614 msg = mimetools.Message(StringIO("\r\n\r\n")) 0615 return MockResponse(200, "OK", msg, "", req.get_full_url()) 0616 # cookies shouldn't leak into redirected requests 0617 from cookielib import CookieJar 0618 from urllib2 import build_opener, HTTPHandler, HTTPError, \ 0619 HTTPCookieProcessor 0620 0621 from test_cookielib import interact_netscape 0622 0623 cj = CookieJar() 0624 interact_netscape(cj, "http://www.example.com/", "spam=eggs") 0625 hh = MockHTTPHandler() 0626 cp = HTTPCookieProcessor(cj) 0627 o = build_opener(hh, cp) 0628 o.open("http://www.example.com/") 0629 self.assert_(not hh.req.has_header("Cookie")) 0630 0631 0632 class MiscTests(unittest.TestCase): 0633 0634 def test_build_opener(self): 0635 class MyHTTPHandler(urllib2.HTTPHandler): pass 0636 class FooHandler(urllib2.BaseHandler): 0637 def foo_open(self): pass 0638 class BarHandler(urllib2.BaseHandler): 0639 def bar_open(self): pass 0640 0641 build_opener = urllib2.build_opener 0642 0643 o = build_opener(FooHandler, BarHandler) 0644 self.opener_has_handler(o, FooHandler) 0645 self.opener_has_handler(o, BarHandler) 0646 0647 # can take a mix of classes and instances 0648 o = build_opener(FooHandler, BarHandler()) 0649 self.opener_has_handler(o, FooHandler) 0650 self.opener_has_handler(o, BarHandler) 0651 0652 # subclasses of default handlers override default handlers 0653 o = build_opener(MyHTTPHandler) 0654 self.opener_has_handler(o, MyHTTPHandler) 0655 0656 # a particular case of overriding: default handlers can be passed 0657 # in explicitly 0658 o = build_opener() 0659 self.opener_has_handler(o, urllib2.HTTPHandler) 0660 o = build_opener(urllib2.HTTPHandler) 0661 self.opener_has_handler(o, urllib2.HTTPHandler) 0662 o = build_opener(urllib2.HTTPHandler()) 0663 self.opener_has_handler(o, urllib2.HTTPHandler) 0664 0665 def opener_has_handler(self, opener, handler_class): 0666 for h in opener.handlers: 0667 if h.__class__ == handler_class: 0668 break 0669 else: 0670 self.assert_(False) 0671 0672 class NetworkTests(unittest.TestCase): 0673 def setUp(self): 0674 if 0: # for debugging 0675 import logging 0676 logger = logging.getLogger("test_urllib2") 0677 logger.addHandler(logging.StreamHandler()) 0678 0679 def test_range (self): 0680 req = urllib2.Request("http://www.python.org", 0681 headers={'Range': 'bytes=20-39'}) 0682 result = urllib2.urlopen(req) 0683 data = result.read() 0684 self.assertEqual(len(data), 20) 0685 0686 # XXX The rest of these tests aren't very good -- they don't check much. 0687 # They do sometimes catch some major disasters, though. 0688 0689 def test_ftp(self): 0690 urls = [ 0691 'ftp://www.python.org/pub/python/misc/sousa.au', 0692 'ftp://www.python.org/pub/tmp/blat', 0693 'ftp://gatekeeper.research.compaq.com/pub/DEC/SRC' 0694 '/research-reports/00README-Legal-Rules-Regs', 0695 ] 0696 self._test_urls(urls, self._extra_handlers()) 0697 0698 def test_gopher(self): 0699 urls = [ 0700 # Thanks to Fred for finding these! 0701 'gopher://gopher.lib.ncsu.edu/11/library/stacks/Alex', 0702 'gopher://gopher.vt.edu:10010/10/33', 0703 ] 0704 self._test_urls(urls, self._extra_handlers()) 0705 0706 def test_file(self): 0707 TESTFN = test_support.TESTFN 0708 f = open(TESTFN, 'w') 0709 try: 0710 f.write('hi there\n') 0711 f.close() 0712 urls = [ 0713 'file:'+sanepathname2url(os.path.abspath(TESTFN)), 0714 0715 # XXX bug, should raise URLError 0716 #('file://nonsensename/etc/passwd', None, urllib2.URLError) 0717 ('file://nonsensename/etc/passwd', None, (OSError, socket.error)) 0718 ] 0719 self._test_urls(urls, self._extra_handlers()) 0720 finally: 0721 os.remove(TESTFN) 0722 0723 def test_http(self): 0724 urls = [ 0725 'http://www.espn.com/', # redirect 0726 'http://www.python.org/Spanish/Inquistion/', 0727 ('http://www.python.org/cgi-bin/faqw.py', 0728 'query=pythonistas&querytype=simple&casefold=yes&req=search', None), 0729 'http://www.python.org/', 0730 ] 0731 self._test_urls(urls, self._extra_handlers()) 0732 0733 # XXX Following test depends on machine configurations that are internal 0734 # to CNRI. Need to set up a public server with the right authentication 0735 # configuration for test purposes. 0736 0737 ## def test_cnri(self): 0738 ## if socket.gethostname() == 'bitdiddle': 0739 ## localhost = 'bitdiddle.cnri.reston.va.us' 0740 ## elif socket.gethostname() == 'bitdiddle.concentric.net': 0741 ## localhost = 'localhost' 0742 ## else: 0743 ## localhost = None 0744 ## if localhost is not None: 0745 ## urls = [ 0746 ## 'file://%s/etc/passwd' % localhost, 0747 ## 'http://%s/simple/' % localhost, 0748 ## 'http://%s/digest/' % localhost, 0749 ## 'http://%s/not/found.h' % localhost, 0750 ## ] 0751 0752 ## bauth = HTTPBasicAuthHandler() 0753 ## bauth.add_password('basic_test_realm', localhost, 'jhylton', 0754 ## 'password') 0755 ## dauth = HTTPDigestAuthHandler() 0756 ## dauth.add_password('digest_test_realm', localhost, 'jhylton', 0757 ## 'password') 0758 0759 ## self._test_urls(urls, self._extra_handlers()+[bauth, dauth]) 0760 0761 def _test_urls(self, urls, handlers): 0762 import socket 0763 import time 0764 import logging 0765 debug = logging.getLogger("test_urllib2").debug 0766 0767 urllib2.install_opener(urllib2.build_opener(*handlers)) 0768 0769 for url in urls: 0770 if isinstance(url, tuple): 0771 url, req, expected_err = url 0772 else: 0773 req = expected_err = None 0774 debug(url) 0775 try: 0776 f = urllib2.urlopen(url, req) 0777 except (IOError, socket.error, OSError), err: 0778 debug(err) 0779 if expected_err: 0780 self.assert_(isinstance(err, expected_err)) 0781 else: 0782 buf = f.read() 0783 f.close() 0784 debug("read %d bytes" % len(buf)) 0785 debug("******** next url coming up...") 0786 time.sleep(0.1) 0787 0788 def _extra_handlers(self): 0789 handlers = [] 0790 0791 handlers.append(urllib2.GopherHandler) 0792 0793 cfh = urllib2.CacheFTPHandler() 0794 cfh.setTimeout(1) 0795 handlers.append(cfh) 0796 0797 ## # XXX try out some custom proxy objects too! 0798 ## def at_cnri(req): 0799 ## host = req.get_host() 0800 ## debug(host) 0801 ## if host[-18:] == '.cnri.reston.va.us': 0802 ## return True 0803 ## p = CustomProxy('http', at_cnri, 'proxy.cnri.reston.va.us') 0804 ## ph = CustomProxyHandler(p) 0805 ## handlers.append(ph) 0806 0807 return handlers 0808 0809 0810 def test_main(verbose=None): 0811 tests = (TrivialTests, 0812 OpenerDirectorTests, 0813 HandlerTests, 0814 MiscTests) 0815 if test_support.is_resource_enabled('network'): 0816 tests += (NetworkTests,) 0817 test_support.run_unittest(*tests) 0818 0819 if __name__ == "__main__": 0820 test_main(verbose=True) 0821
Generated by PyXR 0.9.4