diff options
Diffstat (limited to 'lib/python2.7/test/test_urllib2net.py')
| -rw-r--r-- | lib/python2.7/test/test_urllib2net.py | 332 |
1 files changed, 0 insertions, 332 deletions
diff --git a/lib/python2.7/test/test_urllib2net.py b/lib/python2.7/test/test_urllib2net.py deleted file mode 100644 index 4d73ac0..0000000 --- a/lib/python2.7/test/test_urllib2net.py +++ /dev/null @@ -1,332 +0,0 @@ -#!/usr/bin/env python - -import unittest -from test import test_support -from test.test_urllib2 import sanepathname2url - -import socket -import urllib2 -import os -import sys - -TIMEOUT = 60 # seconds - - -def _retry_thrice(func, exc, *args, **kwargs): - for i in range(3): - try: - return func(*args, **kwargs) - except exc, last_exc: - continue - except: - raise - raise last_exc - -def _wrap_with_retry_thrice(func, exc): - def wrapped(*args, **kwargs): - return _retry_thrice(func, exc, *args, **kwargs) - return wrapped - -# Connecting to remote hosts is flaky. Make it more robust by retrying -# the connection several times. -_urlopen_with_retry = _wrap_with_retry_thrice(urllib2.urlopen, urllib2.URLError) - - -class AuthTests(unittest.TestCase): - """Tests urllib2 authentication features.""" - -## Disabled at the moment since there is no page under python.org which -## could be used to HTTP authentication. -# -# def test_basic_auth(self): -# import httplib -# -# test_url = "http://www.python.org/test/test_urllib2/basic_auth" -# test_hostport = "www.python.org" -# test_realm = 'Test Realm' -# test_user = 'test.test_urllib2net' -# test_password = 'blah' -# -# # failure -# try: -# _urlopen_with_retry(test_url) -# except urllib2.HTTPError, exc: -# self.assertEqual(exc.code, 401) -# else: -# self.fail("urlopen() should have failed with 401") -# -# # success -# auth_handler = urllib2.HTTPBasicAuthHandler() -# auth_handler.add_password(test_realm, test_hostport, -# test_user, test_password) -# opener = urllib2.build_opener(auth_handler) -# f = opener.open('http://localhost/') -# response = _urlopen_with_retry("http://www.python.org/") -# -# # The 'userinfo' URL component is deprecated by RFC 3986 for security -# # reasons, let's not implement it! (it's already implemented for proxy -# # specification strings (that is, URLs or authorities specifying a -# # proxy), so we must keep that) -# self.assertRaises(httplib.InvalidURL, -# urllib2.urlopen, "http://evil:thing@example.com") - - -class CloseSocketTest(unittest.TestCase): - - def test_close(self): - import httplib - - # calling .close() on urllib2's response objects should close the - # underlying socket - - # delve deep into response to fetch socket._socketobject - response = _urlopen_with_retry("http://www.python.org/") - abused_fileobject = response.fp - self.assertTrue(abused_fileobject.__class__ is socket._fileobject) - httpresponse = abused_fileobject._sock - self.assertTrue(httpresponse.__class__ is httplib.HTTPResponse) - fileobject = httpresponse.fp - self.assertTrue(fileobject.__class__ is socket._fileobject) - - self.assertTrue(not fileobject.closed) - response.close() - self.assertTrue(fileobject.closed) - -class OtherNetworkTests(unittest.TestCase): - def setUp(self): - if 0: # for debugging - import logging - logger = logging.getLogger("test_urllib2net") - logger.addHandler(logging.StreamHandler()) - - # XXX The rest of these tests aren't very good -- they don't check much. - # They do sometimes catch some major disasters, though. - - def test_ftp(self): - urls = [ - 'ftp://ftp.kernel.org/pub/linux/kernel/README', - 'ftp://ftp.kernel.org/pub/linux/kernel/non-existent-file', - #'ftp://ftp.kernel.org/pub/leenox/kernel/test', - 'ftp://gatekeeper.research.compaq.com/pub/DEC/SRC' - '/research-reports/00README-Legal-Rules-Regs', - ] - self._test_urls(urls, self._extra_handlers()) - - def test_file(self): - TESTFN = test_support.TESTFN - f = open(TESTFN, 'w') - try: - f.write('hi there\n') - f.close() - urls = [ - 'file:'+sanepathname2url(os.path.abspath(TESTFN)), - ('file:///nonsensename/etc/passwd', None, urllib2.URLError), - ] - self._test_urls(urls, self._extra_handlers(), retry=True) - finally: - os.remove(TESTFN) - - self.assertRaises(ValueError, urllib2.urlopen,'./relative_path/to/file') - - # XXX Following test depends on machine configurations that are internal - # to CNRI. Need to set up a public server with the right authentication - # configuration for test purposes. - -## def test_cnri(self): -## if socket.gethostname() == 'bitdiddle': -## localhost = 'bitdiddle.cnri.reston.va.us' -## elif socket.gethostname() == 'bitdiddle.concentric.net': -## localhost = 'localhost' -## else: -## localhost = None -## if localhost is not None: -## urls = [ -## 'file://%s/etc/passwd' % localhost, -## 'http://%s/simple/' % localhost, -## 'http://%s/digest/' % localhost, -## 'http://%s/not/found.h' % localhost, -## ] - -## bauth = HTTPBasicAuthHandler() -## bauth.add_password('basic_test_realm', localhost, 'jhylton', -## 'password') -## dauth = HTTPDigestAuthHandler() -## dauth.add_password('digest_test_realm', localhost, 'jhylton', -## 'password') - -## self._test_urls(urls, self._extra_handlers()+[bauth, dauth]) - - def test_urlwithfrag(self): - urlwith_frag = "http://docs.python.org/2/glossary.html#glossary" - with test_support.transient_internet(urlwith_frag): - req = urllib2.Request(urlwith_frag) - res = urllib2.urlopen(req) - self.assertEqual(res.geturl(), - "http://docs.python.org/2/glossary.html#glossary") - - def test_fileno(self): - req = urllib2.Request("http://www.python.org") - opener = urllib2.build_opener() - res = opener.open(req) - try: - res.fileno() - except AttributeError: - self.fail("HTTPResponse object should return a valid fileno") - finally: - res.close() - - def test_custom_headers(self): - url = "http://www.example.com" - with test_support.transient_internet(url): - opener = urllib2.build_opener() - request = urllib2.Request(url) - self.assertFalse(request.header_items()) - opener.open(request) - self.assertTrue(request.header_items()) - self.assertTrue(request.has_header('User-agent')) - request.add_header('User-Agent','Test-Agent') - opener.open(request) - self.assertEqual(request.get_header('User-agent'),'Test-Agent') - - def test_sites_no_connection_close(self): - # Some sites do not send Connection: close header. - # Verify that those work properly. (#issue12576) - - URL = 'http://www.imdb.com' # No Connection:close - with test_support.transient_internet(URL): - req = urllib2.urlopen(URL) - res = req.read() - self.assertTrue(res) - - def _test_urls(self, urls, handlers, retry=True): - import time - import logging - debug = logging.getLogger("test_urllib2").debug - - urlopen = urllib2.build_opener(*handlers).open - if retry: - urlopen = _wrap_with_retry_thrice(urlopen, urllib2.URLError) - - for url in urls: - if isinstance(url, tuple): - url, req, expected_err = url - else: - req = expected_err = None - with test_support.transient_internet(url): - debug(url) - try: - f = urlopen(url, req, TIMEOUT) - except EnvironmentError as err: - debug(err) - if expected_err: - msg = ("Didn't get expected error(s) %s for %s %s, got %s: %s" % - (expected_err, url, req, type(err), err)) - self.assertIsInstance(err, expected_err, msg) - except urllib2.URLError as err: - if isinstance(err[0], socket.timeout): - print >>sys.stderr, "<timeout: %s>" % url - continue - else: - raise - else: - try: - with test_support.transient_internet(url): - buf = f.read() - debug("read %d bytes" % len(buf)) - except socket.timeout: - print >>sys.stderr, "<timeout: %s>" % url - f.close() - debug("******** next url coming up...") - time.sleep(0.1) - - def _extra_handlers(self): - handlers = [] - - cfh = urllib2.CacheFTPHandler() - self.addCleanup(cfh.clear_cache) - cfh.setTimeout(1) - handlers.append(cfh) - - return handlers - - -class TimeoutTest(unittest.TestCase): - def test_http_basic(self): - self.assertTrue(socket.getdefaulttimeout() is None) - url = "http://www.python.org" - with test_support.transient_internet(url, timeout=None): - u = _urlopen_with_retry(url) - self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None) - - def test_http_default_timeout(self): - self.assertTrue(socket.getdefaulttimeout() is None) - url = "http://www.python.org" - with test_support.transient_internet(url): - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry(url) - finally: - socket.setdefaulttimeout(None) - self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60) - - def test_http_no_timeout(self): - self.assertTrue(socket.getdefaulttimeout() is None) - url = "http://www.python.org" - with test_support.transient_internet(url): - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry(url, timeout=None) - finally: - socket.setdefaulttimeout(None) - self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None) - - def test_http_timeout(self): - url = "http://www.python.org" - with test_support.transient_internet(url): - u = _urlopen_with_retry(url, timeout=120) - self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120) - - FTP_HOST = "ftp://ftp.mirror.nl/pub/gnu/" - - def test_ftp_basic(self): - self.assertTrue(socket.getdefaulttimeout() is None) - with test_support.transient_internet(self.FTP_HOST, timeout=None): - u = _urlopen_with_retry(self.FTP_HOST) - self.assertTrue(u.fp.fp._sock.gettimeout() is None) - - def test_ftp_default_timeout(self): - self.assertTrue(socket.getdefaulttimeout() is None) - with test_support.transient_internet(self.FTP_HOST): - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry(self.FTP_HOST) - finally: - socket.setdefaulttimeout(None) - self.assertEqual(u.fp.fp._sock.gettimeout(), 60) - - def test_ftp_no_timeout(self): - self.assertTrue(socket.getdefaulttimeout() is None) - with test_support.transient_internet(self.FTP_HOST): - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry(self.FTP_HOST, timeout=None) - finally: - socket.setdefaulttimeout(None) - self.assertTrue(u.fp.fp._sock.gettimeout() is None) - - def test_ftp_timeout(self): - with test_support.transient_internet(self.FTP_HOST): - u = _urlopen_with_retry(self.FTP_HOST, timeout=60) - self.assertEqual(u.fp.fp._sock.gettimeout(), 60) - - -def test_main(): - test_support.requires("network") - test_support.run_unittest(AuthTests, - OtherNetworkTests, - CloseSocketTest, - TimeoutTest, - ) - -if __name__ == "__main__": - test_main() |