| 1 | #!/usr/bin/env python
|
|---|
| 2 |
|
|---|
| 3 | import unittest
|
|---|
| 4 | from test import test_support
|
|---|
| 5 | from test.test_urllib2 import sanepathname2url
|
|---|
| 6 |
|
|---|
| 7 | import socket
|
|---|
| 8 | import urllib2
|
|---|
| 9 | import os
|
|---|
| 10 | import sys
|
|---|
| 11 |
|
|---|
| 12 | TIMEOUT = 60 # seconds
|
|---|
| 13 |
|
|---|
| 14 |
|
|---|
| 15 | def _retry_thrice(func, exc, *args, **kwargs):
|
|---|
| 16 | for i in range(3):
|
|---|
| 17 | try:
|
|---|
| 18 | return func(*args, **kwargs)
|
|---|
| 19 | except exc, last_exc:
|
|---|
| 20 | continue
|
|---|
| 21 | except:
|
|---|
| 22 | raise
|
|---|
| 23 | raise last_exc
|
|---|
| 24 |
|
|---|
| 25 | def _wrap_with_retry_thrice(func, exc):
|
|---|
| 26 | def wrapped(*args, **kwargs):
|
|---|
| 27 | return _retry_thrice(func, exc, *args, **kwargs)
|
|---|
| 28 | return wrapped
|
|---|
| 29 |
|
|---|
| 30 | # Connecting to remote hosts is flaky. Make it more robust by retrying
|
|---|
| 31 | # the connection several times.
|
|---|
| 32 | _urlopen_with_retry = _wrap_with_retry_thrice(urllib2.urlopen, urllib2.URLError)
|
|---|
| 33 |
|
|---|
| 34 |
|
|---|
| 35 | class AuthTests(unittest.TestCase):
|
|---|
| 36 | """Tests urllib2 authentication features."""
|
|---|
| 37 |
|
|---|
| 38 | ## Disabled at the moment since there is no page under python.org which
|
|---|
| 39 | ## could be used to HTTP authentication.
|
|---|
| 40 | #
|
|---|
| 41 | # def test_basic_auth(self):
|
|---|
| 42 | # import httplib
|
|---|
| 43 | #
|
|---|
| 44 | # test_url = "http://www.python.org/test/test_urllib2/basic_auth"
|
|---|
| 45 | # test_hostport = "www.python.org"
|
|---|
| 46 | # test_realm = 'Test Realm'
|
|---|
| 47 | # test_user = 'test.test_urllib2net'
|
|---|
| 48 | # test_password = 'blah'
|
|---|
| 49 | #
|
|---|
| 50 | # # failure
|
|---|
| 51 | # try:
|
|---|
| 52 | # _urlopen_with_retry(test_url)
|
|---|
| 53 | # except urllib2.HTTPError, exc:
|
|---|
| 54 | # self.assertEqual(exc.code, 401)
|
|---|
| 55 | # else:
|
|---|
| 56 | # self.fail("urlopen() should have failed with 401")
|
|---|
| 57 | #
|
|---|
| 58 | # # success
|
|---|
| 59 | # auth_handler = urllib2.HTTPBasicAuthHandler()
|
|---|
| 60 | # auth_handler.add_password(test_realm, test_hostport,
|
|---|
| 61 | # test_user, test_password)
|
|---|
| 62 | # opener = urllib2.build_opener(auth_handler)
|
|---|
| 63 | # f = opener.open('http://localhost/')
|
|---|
| 64 | # response = _urlopen_with_retry("http://www.python.org/")
|
|---|
| 65 | #
|
|---|
| 66 | # # The 'userinfo' URL component is deprecated by RFC 3986 for security
|
|---|
| 67 | # # reasons, let's not implement it! (it's already implemented for proxy
|
|---|
| 68 | # # specification strings (that is, URLs or authorities specifying a
|
|---|
| 69 | # # proxy), so we must keep that)
|
|---|
| 70 | # self.assertRaises(httplib.InvalidURL,
|
|---|
| 71 | # urllib2.urlopen, "http://evil:thing@example.com")
|
|---|
| 72 |
|
|---|
| 73 |
|
|---|
| 74 | class CloseSocketTest(unittest.TestCase):
|
|---|
| 75 |
|
|---|
| 76 | def test_close(self):
|
|---|
| 77 | import httplib
|
|---|
| 78 |
|
|---|
| 79 | # calling .close() on urllib2's response objects should close the
|
|---|
| 80 | # underlying socket
|
|---|
| 81 |
|
|---|
| 82 | # delve deep into response to fetch socket._socketobject
|
|---|
| 83 | response = _urlopen_with_retry("http://www.python.org/")
|
|---|
| 84 | abused_fileobject = response.fp
|
|---|
| 85 | self.assertTrue(abused_fileobject.__class__ is socket._fileobject)
|
|---|
| 86 | httpresponse = abused_fileobject._sock
|
|---|
| 87 | self.assertTrue(httpresponse.__class__ is httplib.HTTPResponse)
|
|---|
| 88 | fileobject = httpresponse.fp
|
|---|
| 89 | self.assertTrue(fileobject.__class__ is socket._fileobject)
|
|---|
| 90 |
|
|---|
| 91 | self.assertTrue(not fileobject.closed)
|
|---|
| 92 | response.close()
|
|---|
| 93 | self.assertTrue(fileobject.closed)
|
|---|
| 94 |
|
|---|
| 95 | class OtherNetworkTests(unittest.TestCase):
|
|---|
| 96 | def setUp(self):
|
|---|
| 97 | if 0: # for debugging
|
|---|
| 98 | import logging
|
|---|
| 99 | logger = logging.getLogger("test_urllib2net")
|
|---|
| 100 | logger.addHandler(logging.StreamHandler())
|
|---|
| 101 |
|
|---|
| 102 | # XXX The rest of these tests aren't very good -- they don't check much.
|
|---|
| 103 | # They do sometimes catch some major disasters, though.
|
|---|
| 104 |
|
|---|
| 105 | def test_ftp(self):
|
|---|
| 106 | urls = [
|
|---|
| 107 | 'ftp://ftp.kernel.org/pub/linux/kernel/README',
|
|---|
| 108 | 'ftp://ftp.kernel.org/pub/linux/kernel/non-existent-file',
|
|---|
| 109 | #'ftp://ftp.kernel.org/pub/leenox/kernel/test',
|
|---|
| 110 | 'ftp://gatekeeper.research.compaq.com/pub/DEC/SRC'
|
|---|
| 111 | '/research-reports/00README-Legal-Rules-Regs',
|
|---|
| 112 | ]
|
|---|
| 113 | self._test_urls(urls, self._extra_handlers())
|
|---|
| 114 |
|
|---|
| 115 | def test_file(self):
|
|---|
| 116 | TESTFN = test_support.TESTFN
|
|---|
| 117 | f = open(TESTFN, 'w')
|
|---|
| 118 | try:
|
|---|
| 119 | f.write('hi there\n')
|
|---|
| 120 | f.close()
|
|---|
| 121 | urls = [
|
|---|
| 122 | 'file:'+sanepathname2url(os.path.abspath(TESTFN)),
|
|---|
| 123 | ('file:///nonsensename/etc/passwd', None, urllib2.URLError),
|
|---|
| 124 | ]
|
|---|
| 125 | self._test_urls(urls, self._extra_handlers(), retry=True)
|
|---|
| 126 | finally:
|
|---|
| 127 | os.remove(TESTFN)
|
|---|
| 128 |
|
|---|
| 129 | self.assertRaises(ValueError, urllib2.urlopen,'./relative_path/to/file')
|
|---|
| 130 |
|
|---|
| 131 | # XXX Following test depends on machine configurations that are internal
|
|---|
| 132 | # to CNRI. Need to set up a public server with the right authentication
|
|---|
| 133 | # configuration for test purposes.
|
|---|
| 134 |
|
|---|
| 135 | ## def test_cnri(self):
|
|---|
| 136 | ## if socket.gethostname() == 'bitdiddle':
|
|---|
| 137 | ## localhost = 'bitdiddle.cnri.reston.va.us'
|
|---|
| 138 | ## elif socket.gethostname() == 'bitdiddle.concentric.net':
|
|---|
| 139 | ## localhost = 'localhost'
|
|---|
| 140 | ## else:
|
|---|
| 141 | ## localhost = None
|
|---|
| 142 | ## if localhost is not None:
|
|---|
| 143 | ## urls = [
|
|---|
| 144 | ## 'file://%s/etc/passwd' % localhost,
|
|---|
| 145 | ## 'http://%s/simple/' % localhost,
|
|---|
| 146 | ## 'http://%s/digest/' % localhost,
|
|---|
| 147 | ## 'http://%s/not/found.h' % localhost,
|
|---|
| 148 | ## ]
|
|---|
| 149 |
|
|---|
| 150 | ## bauth = HTTPBasicAuthHandler()
|
|---|
| 151 | ## bauth.add_password('basic_test_realm', localhost, 'jhylton',
|
|---|
| 152 | ## 'password')
|
|---|
| 153 | ## dauth = HTTPDigestAuthHandler()
|
|---|
| 154 | ## dauth.add_password('digest_test_realm', localhost, 'jhylton',
|
|---|
| 155 | ## 'password')
|
|---|
| 156 |
|
|---|
| 157 | ## self._test_urls(urls, self._extra_handlers()+[bauth, dauth])
|
|---|
| 158 |
|
|---|
| 159 | def test_urlwithfrag(self):
|
|---|
| 160 | urlwith_frag = "http://docs.python.org/2/glossary.html#glossary"
|
|---|
| 161 | with test_support.transient_internet(urlwith_frag):
|
|---|
| 162 | req = urllib2.Request(urlwith_frag)
|
|---|
| 163 | res = urllib2.urlopen(req)
|
|---|
| 164 | self.assertEqual(res.geturl(),
|
|---|
| 165 | "http://docs.python.org/2/glossary.html#glossary")
|
|---|
| 166 |
|
|---|
| 167 | def test_fileno(self):
|
|---|
| 168 | req = urllib2.Request("http://www.python.org")
|
|---|
| 169 | opener = urllib2.build_opener()
|
|---|
| 170 | res = opener.open(req)
|
|---|
| 171 | try:
|
|---|
| 172 | res.fileno()
|
|---|
| 173 | except AttributeError:
|
|---|
| 174 | self.fail("HTTPResponse object should return a valid fileno")
|
|---|
| 175 | finally:
|
|---|
| 176 | res.close()
|
|---|
| 177 |
|
|---|
| 178 | def test_custom_headers(self):
|
|---|
| 179 | url = "http://www.example.com"
|
|---|
| 180 | with test_support.transient_internet(url):
|
|---|
| 181 | opener = urllib2.build_opener()
|
|---|
| 182 | request = urllib2.Request(url)
|
|---|
| 183 | self.assertFalse(request.header_items())
|
|---|
| 184 | opener.open(request)
|
|---|
| 185 | self.assertTrue(request.header_items())
|
|---|
| 186 | self.assertTrue(request.has_header('User-agent'))
|
|---|
| 187 | request.add_header('User-Agent','Test-Agent')
|
|---|
| 188 | opener.open(request)
|
|---|
| 189 | self.assertEqual(request.get_header('User-agent'),'Test-Agent')
|
|---|
| 190 |
|
|---|
| 191 | def test_sites_no_connection_close(self):
|
|---|
| 192 | # Some sites do not send Connection: close header.
|
|---|
| 193 | # Verify that those work properly. (#issue12576)
|
|---|
| 194 |
|
|---|
| 195 | URL = 'http://www.imdb.com' # No Connection:close
|
|---|
| 196 | with test_support.transient_internet(URL):
|
|---|
| 197 | req = urllib2.urlopen(URL)
|
|---|
| 198 | res = req.read()
|
|---|
| 199 | self.assertTrue(res)
|
|---|
| 200 |
|
|---|
| 201 | def _test_urls(self, urls, handlers, retry=True):
|
|---|
| 202 | import time
|
|---|
| 203 | import logging
|
|---|
| 204 | debug = logging.getLogger("test_urllib2").debug
|
|---|
| 205 |
|
|---|
| 206 | urlopen = urllib2.build_opener(*handlers).open
|
|---|
| 207 | if retry:
|
|---|
| 208 | urlopen = _wrap_with_retry_thrice(urlopen, urllib2.URLError)
|
|---|
| 209 |
|
|---|
| 210 | for url in urls:
|
|---|
| 211 | if isinstance(url, tuple):
|
|---|
| 212 | url, req, expected_err = url
|
|---|
| 213 | else:
|
|---|
| 214 | req = expected_err = None
|
|---|
| 215 | with test_support.transient_internet(url):
|
|---|
| 216 | debug(url)
|
|---|
| 217 | try:
|
|---|
| 218 | f = urlopen(url, req, TIMEOUT)
|
|---|
| 219 | except EnvironmentError as err:
|
|---|
| 220 | debug(err)
|
|---|
| 221 | if expected_err:
|
|---|
| 222 | msg = ("Didn't get expected error(s) %s for %s %s, got %s: %s" %
|
|---|
| 223 | (expected_err, url, req, type(err), err))
|
|---|
| 224 | self.assertIsInstance(err, expected_err, msg)
|
|---|
| 225 | except urllib2.URLError as err:
|
|---|
| 226 | if isinstance(err[0], socket.timeout):
|
|---|
| 227 | print >>sys.stderr, "<timeout: %s>" % url
|
|---|
| 228 | continue
|
|---|
| 229 | else:
|
|---|
| 230 | raise
|
|---|
| 231 | else:
|
|---|
| 232 | try:
|
|---|
| 233 | with test_support.transient_internet(url):
|
|---|
| 234 | buf = f.read()
|
|---|
| 235 | debug("read %d bytes" % len(buf))
|
|---|
| 236 | except socket.timeout:
|
|---|
| 237 | print >>sys.stderr, "<timeout: %s>" % url
|
|---|
| 238 | f.close()
|
|---|
| 239 | debug("******** next url coming up...")
|
|---|
| 240 | time.sleep(0.1)
|
|---|
| 241 |
|
|---|
| 242 | def _extra_handlers(self):
|
|---|
| 243 | handlers = []
|
|---|
| 244 |
|
|---|
| 245 | cfh = urllib2.CacheFTPHandler()
|
|---|
| 246 | self.addCleanup(cfh.clear_cache)
|
|---|
| 247 | cfh.setTimeout(1)
|
|---|
| 248 | handlers.append(cfh)
|
|---|
| 249 |
|
|---|
| 250 | return handlers
|
|---|
| 251 |
|
|---|
| 252 |
|
|---|
| 253 | class TimeoutTest(unittest.TestCase):
|
|---|
| 254 | def test_http_basic(self):
|
|---|
| 255 | self.assertTrue(socket.getdefaulttimeout() is None)
|
|---|
| 256 | url = "http://www.python.org"
|
|---|
| 257 | with test_support.transient_internet(url, timeout=None):
|
|---|
| 258 | u = _urlopen_with_retry(url)
|
|---|
| 259 | self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
|
|---|
| 260 |
|
|---|
| 261 | def test_http_default_timeout(self):
|
|---|
| 262 | self.assertTrue(socket.getdefaulttimeout() is None)
|
|---|
| 263 | url = "http://www.python.org"
|
|---|
| 264 | with test_support.transient_internet(url):
|
|---|
| 265 | socket.setdefaulttimeout(60)
|
|---|
| 266 | try:
|
|---|
| 267 | u = _urlopen_with_retry(url)
|
|---|
| 268 | finally:
|
|---|
| 269 | socket.setdefaulttimeout(None)
|
|---|
| 270 | self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60)
|
|---|
| 271 |
|
|---|
| 272 | def test_http_no_timeout(self):
|
|---|
| 273 | self.assertTrue(socket.getdefaulttimeout() is None)
|
|---|
| 274 | url = "http://www.python.org"
|
|---|
| 275 | with test_support.transient_internet(url):
|
|---|
| 276 | socket.setdefaulttimeout(60)
|
|---|
| 277 | try:
|
|---|
| 278 | u = _urlopen_with_retry(url, timeout=None)
|
|---|
| 279 | finally:
|
|---|
| 280 | socket.setdefaulttimeout(None)
|
|---|
| 281 | self.assertTrue(u.fp._sock.fp._sock.gettimeout() is None)
|
|---|
| 282 |
|
|---|
| 283 | def test_http_timeout(self):
|
|---|
| 284 | url = "http://www.python.org"
|
|---|
| 285 | with test_support.transient_internet(url):
|
|---|
| 286 | u = _urlopen_with_retry(url, timeout=120)
|
|---|
| 287 | self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120)
|
|---|
| 288 |
|
|---|
| 289 | FTP_HOST = "ftp://ftp.mirror.nl/pub/gnu/"
|
|---|
| 290 |
|
|---|
| 291 | def test_ftp_basic(self):
|
|---|
| 292 | self.assertTrue(socket.getdefaulttimeout() is None)
|
|---|
| 293 | with test_support.transient_internet(self.FTP_HOST, timeout=None):
|
|---|
| 294 | u = _urlopen_with_retry(self.FTP_HOST)
|
|---|
| 295 | self.assertTrue(u.fp.fp._sock.gettimeout() is None)
|
|---|
| 296 |
|
|---|
| 297 | def test_ftp_default_timeout(self):
|
|---|
| 298 | self.assertTrue(socket.getdefaulttimeout() is None)
|
|---|
| 299 | with test_support.transient_internet(self.FTP_HOST):
|
|---|
| 300 | socket.setdefaulttimeout(60)
|
|---|
| 301 | try:
|
|---|
| 302 | u = _urlopen_with_retry(self.FTP_HOST)
|
|---|
| 303 | finally:
|
|---|
| 304 | socket.setdefaulttimeout(None)
|
|---|
| 305 | self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
|
|---|
| 306 |
|
|---|
| 307 | def test_ftp_no_timeout(self):
|
|---|
| 308 | self.assertTrue(socket.getdefaulttimeout() is None)
|
|---|
| 309 | with test_support.transient_internet(self.FTP_HOST):
|
|---|
| 310 | socket.setdefaulttimeout(60)
|
|---|
| 311 | try:
|
|---|
| 312 | u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
|
|---|
| 313 | finally:
|
|---|
| 314 | socket.setdefaulttimeout(None)
|
|---|
| 315 | self.assertTrue(u.fp.fp._sock.gettimeout() is None)
|
|---|
| 316 |
|
|---|
| 317 | def test_ftp_timeout(self):
|
|---|
| 318 | with test_support.transient_internet(self.FTP_HOST):
|
|---|
| 319 | u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
|
|---|
| 320 | self.assertEqual(u.fp.fp._sock.gettimeout(), 60)
|
|---|
| 321 |
|
|---|
| 322 |
|
|---|
| 323 | def test_main():
|
|---|
| 324 | test_support.requires("network")
|
|---|
| 325 | test_support.run_unittest(AuthTests,
|
|---|
| 326 | OtherNetworkTests,
|
|---|
| 327 | CloseSocketTest,
|
|---|
| 328 | TimeoutTest,
|
|---|
| 329 | )
|
|---|
| 330 |
|
|---|
| 331 | if __name__ == "__main__":
|
|---|
| 332 | test_main()
|
|---|
