summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_urllib2net.py
diff options
context:
space:
mode:
authorGeorg Brandl <georg@python.org>2010-10-28 14:55:02 (GMT)
committerGeorg Brandl <georg@python.org>2010-10-28 14:55:02 (GMT)
commit5be365f55d37dbceef6c12492007514c558437f6 (patch)
tree6b43a86b654c0814019e6481169a5291f0586057 /Lib/test/test_urllib2net.py
parentae2d8320a5aa46eaae78db5abd1e3fec38e68f08 (diff)
downloadcpython-5be365f55d37dbceef6c12492007514c558437f6.zip
cpython-5be365f55d37dbceef6c12492007514c558437f6.tar.gz
cpython-5be365f55d37dbceef6c12492007514c558437f6.tar.bz2
#10116: wrap transient_internet() around net access in test_urllib2net.
Diffstat (limited to 'Lib/test/test_urllib2net.py')
-rw-r--r--Lib/test/test_urllib2net.py154
1 files changed, 85 insertions, 69 deletions
diff --git a/Lib/test/test_urllib2net.py b/Lib/test/test_urllib2net.py
index 0a777c4..dedb9d8 100644
--- a/Lib/test/test_urllib2net.py
+++ b/Lib/test/test_urllib2net.py
@@ -155,22 +155,24 @@ class OtherNetworkTests(unittest.TestCase):
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 time
@@ -186,31 +188,33 @@ class OtherNetworkTests(unittest.TestCase):
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.assertIsInstance(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.assertIsInstance(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 +231,71 @@ class OtherNetworkTests(unittest.TestCase):
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/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)
@unittest.skipUnless(ssl, "requires SSL support")