Mirror
/
cpython
mirror of https://github.com/python/cpython
4
1
Fork 0
Code Issues Releases Wiki Activity

Merged revisions 85893 via svnmerge from 

svn+ssh://pythondev@svn.python.org/python/branches/py3k

........
  r85893 | georg.brandl | 2010-10-28 16:55:02 +0200 (jeu., 28 oct. 2010) | 1 line

  #10116: wrap transient_internet() around net access in test_urllib2net.
........
This commit is contained in:
Antoine Pitrou 2010-10-31 13:52:53 +00:00
parent a4f1afa640
commit 27683c954b
1 changed files with 84 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

153
Lib/test/test_urllib2net.py
View File

@ -154,22 +154,24 @@ class OtherNetworkTests(unittest.TestCase):
def test_urlwithfrag(self): def test_urlwithfrag(self):
urlwith_frag = "http://docs.python.org/glossary.html#glossary" urlwith_frag = "http://docs.python.org/glossary.html#glossary"
req = urllib.request.Request(urlwith_frag) with support.transient_internet(urlwith_frag):
res = urllib.request.urlopen(req) req = urllib.request.Request(urlwith_frag)
self.assertEqual(res.geturl(), res = urllib.request.urlopen(req)
"http://docs.python.org/glossary.html") self.assertEqual(res.geturl(),
"http://docs.python.org/glossary.html")
def test_custom_headers(self): def test_custom_headers(self):
url = "http://www.example.com" url = "http://www.example.com"
opener = urllib.request.build_opener() with support.transient_internet(url):
request = urllib.request.Request(url) opener = urllib.request.build_opener()
self.assertFalse(request.header_items()) request = urllib.request.Request(url)
opener.open(request) self.assertFalse(request.header_items())
self.assertTrue(request.header_items()) opener.open(request)
self.assertTrue(request.has_header('User-agent')) self.assertTrue(request.header_items())
request.add_header('User-Agent','Test-Agent') self.assertTrue(request.has_header('User-agent'))
opener.open(request) request.add_header('User-Agent','Test-Agent')
self.assertEqual(request.get_header('User-agent'),'Test-Agent') opener.open(request)
self.assertEqual(request.get_header('User-agent'),'Test-Agent')
def _test_urls(self, urls, handlers, retry=True): def _test_urls(self, urls, handlers, retry=True):
import socket import socket
@ -186,31 +188,32 @@ class OtherNetworkTests(unittest.TestCase):
url, req, expected_err = url url, req, expected_err = url
else: else:
req = expected_err = None req = expected_err = None
debug(url) with support.transient_internet(url):
try: debug(url)
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.assertTrue(isinstance(err, expected_err), msg)
except urllib.error.URLError as err:
if isinstance(err[0], socket.timeout):
print("<timeout: %s>" % url, file=sys.stderr)
continue
else:
raise
else:
try: try:
with support.time_out, \ f = urlopen(url, req, TIMEOUT)
support.socket_peer_reset, \ except EnvironmentError as err:
support.ioerror_peer_reset: debug(err)
buf = f.read() if expected_err:
debug("read %d bytes" % len(buf)) msg = ("Didn't get expected error(s) %s for %s %s, got %s: %s" %
except socket.timeout: (expected_err, url, req, type(err), err))
print("<timeout: %s>" % url, file=sys.stderr) self.assertTrue(isinstance(err, expected_err), msg)
f.close() except urllib.error.URLError as err:
if isinstance(err[0], socket.timeout):
print("<timeout: %s>" % url, file=sys.stderr)
continue
else:
raise
else:
try:
with support.time_out, \
support.socket_peer_reset, \
support.ioerror_peer_reset:
buf = f.read()
debug("read %d bytes" % len(buf))
except socket.timeout:
print("<timeout: %s>" % url, file=sys.stderr)
f.close()
debug("******** next url coming up...") debug("******** next url coming up...")
time.sleep(0.1) time.sleep(0.1)
@ -227,59 +230,71 @@ class OtherNetworkTests(unittest.TestCase):
class TimeoutTest(unittest.TestCase): class TimeoutTest(unittest.TestCase):
def test_http_basic(self): def test_http_basic(self):
self.assertTrue(socket.getdefaulttimeout() is None) self.assertTrue(socket.getdefaulttimeout() is None)
u = _urlopen_with_retry("http://www.python.org") url = "http://www.python.org"
self.assertTrue(u.fp.raw._sock.gettimeout() is None) with support.transient_internet(url, timeout=None):
u = _urlopen_with_retry(url)
self.assertTrue(u.fp.raw._sock.gettimeout() is None)
def test_http_default_timeout(self): def test_http_default_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None) self.assertTrue(socket.getdefaulttimeout() is None)
socket.setdefaulttimeout(60) url = "http://www.python.org"
try: with support.transient_internet(url):
u = _urlopen_with_retry("http://www.python.org") socket.setdefaulttimeout(60)
finally: try:
socket.setdefaulttimeout(None) u = _urlopen_with_retry(url)
self.assertEqual(u.fp.raw._sock.gettimeout(), 60) finally:
socket.setdefaulttimeout(None)
self.assertEqual(u.fp.raw._sock.gettimeout(), 60)
def test_http_no_timeout(self): def test_http_no_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None) self.assertTrue(socket.getdefaulttimeout() is None)
socket.setdefaulttimeout(60) url = "http://www.python.org"
try: with support.transient_internet(url):
u = _urlopen_with_retry("http://www.python.org", timeout=None) socket.setdefaulttimeout(60)
finally: try:
socket.setdefaulttimeout(None) u = _urlopen_with_retry(url, timeout=None)
self.assertTrue(u.fp.raw._sock.gettimeout() is None) finally:
socket.setdefaulttimeout(None)
self.assertTrue(u.fp.raw._sock.gettimeout() is None)
def test_http_timeout(self): def test_http_timeout(self):
u = _urlopen_with_retry("http://www.python.org", timeout=120) url = "http://www.python.org"
self.assertEqual(u.fp.raw._sock.gettimeout(), 120) with support.transient_internet(url):
u = _urlopen_with_retry(url, timeout=120)
self.assertEqual(u.fp.raw._sock.gettimeout(), 120)
FTP_HOST = "ftp://ftp.mirror.nl/pub/mirror/gnu/" FTP_HOST = "ftp://ftp.mirror.nl/pub/mirror/gnu/"
def test_ftp_basic(self): def test_ftp_basic(self):
self.assertTrue(socket.getdefaulttimeout() is None) self.assertTrue(socket.getdefaulttimeout() is None)
u = _urlopen_with_retry(self.FTP_HOST) with support.transient_internet(self.FTP_HOST, timeout=None):
self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None) u = _urlopen_with_retry(self.FTP_HOST)
self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None)
def test_ftp_default_timeout(self): def test_ftp_default_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None) self.assertTrue(socket.getdefaulttimeout() is None)
socket.setdefaulttimeout(60) with support.transient_internet(self.FTP_HOST):
try: socket.setdefaulttimeout(60)
u = _urlopen_with_retry(self.FTP_HOST) try:
finally: u = _urlopen_with_retry(self.FTP_HOST)
socket.setdefaulttimeout(None) finally:
self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60) socket.setdefaulttimeout(None)
self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
def test_ftp_no_timeout(self): def test_ftp_no_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None) self.assertTrue(socket.getdefaulttimeout() is None)
socket.setdefaulttimeout(60) with support.transient_internet(self.FTP_HOST):
try: socket.setdefaulttimeout(60)
u = _urlopen_with_retry(self.FTP_HOST, timeout=None) try:
finally: u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
socket.setdefaulttimeout(None) finally:
self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None) socket.setdefaulttimeout(None)
self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None)
def test_ftp_timeout(self): def test_ftp_timeout(self):
u = _urlopen_with_retry(self.FTP_HOST, timeout=60) with support.transient_internet(self.FTP_HOST):
self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60) u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
def test_main(): def test_main():