[Python-checkins] r86032 - in python/branches/release31-maint: Lib/test/test_urllib2net.py
antoine.pitrou
python-checkins at python.org
Sun Oct 31 14:52:54 CET 2010
Author: antoine.pitrou
Date: Sun Oct 31 14:52:53 2010
New Revision: 86032
Log:
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.
........
Modified:
python/branches/release31-maint/ (props changed)
python/branches/release31-maint/Lib/test/test_urllib2net.py
Modified: python/branches/release31-maint/Lib/test/test_urllib2net.py
==============================================================================
--- python/branches/release31-maint/Lib/test/test_urllib2net.py (original)
+++ python/branches/release31-maint/Lib/test/test_urllib2net.py Sun Oct 31 14:52:53 2010
@@ -154,22 +154,24 @@
def test_urlwithfrag(self):
urlwith_frag = "http://docs.python.org/glossary.html#glossary"
- req = urllib.request.Request(urlwith_frag)
- res = urllib.request.urlopen(req)
- self.assertEqual(res.geturl(),
- "http://docs.python.org/glossary.html")
+ with support.transient_internet(urlwith_frag):
+ req = urllib.request.Request(urlwith_frag)
+ res = urllib.request.urlopen(req)
+ self.assertEqual(res.geturl(),
+ "http://docs.python.org/glossary.html")
def test_custom_headers(self):
url = "http://www.example.com"
- opener = urllib.request.build_opener()
- request = urllib.request.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')
+ with support.transient_internet(url):
+ opener = urllib.request.build_opener()
+ request = urllib.request.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_urls(self, urls, handlers, retry=True):
import socket
@@ -186,31 +188,32 @@
url, req, expected_err = url
else:
req = expected_err = None
- 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.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:
+ with support.transient_internet(url):
+ debug(url)
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()
+ 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:
+ 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...")
time.sleep(0.1)
@@ -227,59 +230,71 @@
class TimeoutTest(unittest.TestCase):
def test_http_basic(self):
self.assertTrue(socket.getdefaulttimeout() is None)
- u = _urlopen_with_retry("http://www.python.org")
- self.assertTrue(u.fp.raw._sock.gettimeout() is None)
+ url = "http://www.python.org"
+ 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):
self.assertTrue(socket.getdefaulttimeout() is None)
- socket.setdefaulttimeout(60)
- try:
- u = _urlopen_with_retry("http://www.python.org")
- finally:
- socket.setdefaulttimeout(None)
- self.assertEqual(u.fp.raw._sock.gettimeout(), 60)
+ url = "http://www.python.org"
+ with support.transient_internet(url):
+ socket.setdefaulttimeout(60)
+ try:
+ u = _urlopen_with_retry(url)
+ finally:
+ socket.setdefaulttimeout(None)
+ self.assertEqual(u.fp.raw._sock.gettimeout(), 60)
def test_http_no_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None)
- socket.setdefaulttimeout(60)
- try:
- u = _urlopen_with_retry("http://www.python.org", timeout=None)
- finally:
- socket.setdefaulttimeout(None)
- self.assertTrue(u.fp.raw._sock.gettimeout() is None)
+ url = "http://www.python.org"
+ with support.transient_internet(url):
+ socket.setdefaulttimeout(60)
+ try:
+ u = _urlopen_with_retry(url, timeout=None)
+ finally:
+ socket.setdefaulttimeout(None)
+ self.assertTrue(u.fp.raw._sock.gettimeout() is None)
def test_http_timeout(self):
- u = _urlopen_with_retry("http://www.python.org", timeout=120)
- self.assertEqual(u.fp.raw._sock.gettimeout(), 120)
+ url = "http://www.python.org"
+ 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/"
def test_ftp_basic(self):
self.assertTrue(socket.getdefaulttimeout() is None)
- u = _urlopen_with_retry(self.FTP_HOST)
- self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None)
+ with support.transient_internet(self.FTP_HOST, timeout=None):
+ u = _urlopen_with_retry(self.FTP_HOST)
+ self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None)
def test_ftp_default_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None)
- socket.setdefaulttimeout(60)
- try:
- u = _urlopen_with_retry(self.FTP_HOST)
- finally:
- socket.setdefaulttimeout(None)
- self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
+ with 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.raw._sock.gettimeout(), 60)
def test_ftp_no_timeout(self):
self.assertTrue(socket.getdefaulttimeout() is None)
- socket.setdefaulttimeout(60)
- try:
- u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
- finally:
- socket.setdefaulttimeout(None)
- self.assertTrue(u.fp.fp.raw._sock.gettimeout() is None)
+ with 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.raw._sock.gettimeout() is None)
def test_ftp_timeout(self):
- u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
- self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
+ with support.transient_internet(self.FTP_HOST):
+ u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
+ self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
def test_main():
More information about the Python-checkins
mailing list