From 9f3f9c512544be77f72a36c5077ab6d7d2a1a83e Mon Sep 17 00:00:00 2001 From: Antoine Pitrou Date: Sun, 31 Oct 2010 13:58:00 +0000 Subject: [PATCH] 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. ........ --- Lib/test/test_urllib2net.py | 149 ++++++++++++++++++++---------------- 1 file changed, 82 insertions(+), 67 deletions(-) diff --git a/Lib/test/test_urllib2net.py b/Lib/test/test_urllib2net.py index 5ace162e7b6..1a7477e780e 100644 --- a/Lib/test/test_urllib2net.py +++ b/Lib/test/test_urllib2net.py @@ -156,10 +156,11 @@ class OtherNetworkTests(unittest.TestCase): def test_urlwithfrag(self): urlwith_frag = "http://docs.python.org/glossary.html#glossary" - req = urllib2.Request(urlwith_frag) - res = urllib2.urlopen(req) - self.assertEqual(res.geturl(), - "http://docs.python.org/glossary.html") + with test_support.transient_internet(urlwith_frag): + req = urllib2.Request(urlwith_frag) + res = urllib2.urlopen(req) + self.assertEqual(res.geturl(), + "http://docs.python.org/glossary.html") def test_fileno(self): req = urllib2.Request("http://www.python.org") @@ -174,15 +175,16 @@ class OtherNetworkTests(unittest.TestCase): def test_custom_headers(self): url = "http://www.example.com" - 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') + 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_urls(self, urls, handlers, retry=True): import time @@ -198,29 +200,30 @@ class OtherNetworkTests(unittest.TestCase): url, req, expected_err = url else: req = expected_err = None - debug(url) - try: - f = urlopen(url, req, TIMEOUT) - except EnvironmentError, 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, "" % url - continue - else: - raise - else: + with test_support.transient_internet(url): + debug(url) try: - with test_support.transient_internet(url): - buf = f.read() - debug("read %d bytes" % len(buf)) - except socket.timeout: - print >>sys.stderr, "" % url - 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 urllib2.URLError as err: + if isinstance(err[0], socket.timeout): + print >>sys.stderr, "" % 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, "" % url + f.close() debug("******** next url coming up...") time.sleep(0.1) @@ -237,59 +240,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._sock.fp._sock.gettimeout() 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) - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry("http://www.python.org") - finally: - socket.setdefaulttimeout(None) - self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 60) + 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) - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry("http://www.python.org", timeout=None) - finally: - socket.setdefaulttimeout(None) - self.assertTrue(u.fp._sock.fp._sock.gettimeout() 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): - u = _urlopen_with_retry("http://www.python.org", timeout=120) - self.assertEqual(u.fp._sock.fp._sock.gettimeout(), 120) + 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) - u = _urlopen_with_retry(self.FTP_HOST) - self.assertTrue(u.fp.fp._sock.gettimeout() 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) - socket.setdefaulttimeout(60) - try: - u = _urlopen_with_retry(self.FTP_HOST) - finally: - socket.setdefaulttimeout(None) - self.assertEqual(u.fp.fp._sock.gettimeout(), 60) + 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) - 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) + 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): - u = _urlopen_with_retry(self.FTP_HOST, timeout=60) - self.assertEqual(u.fp.fp._sock.gettimeout(), 60) + 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():