#10116: wrap transient_internet() around net access in test_urllib2net.

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")