[Python-checkins] r84597 - in python/branches/py3k/Lib/test: support.py test_ssl.py

[antoine.pitrou]

Author: antoine.pitrou
Date: Tue Sep  7 22:42:19 2010
New Revision: 84597

Log:
Issue #8574: better implementation of test.support.transient_internet().
Original patch by Victor.




Modified:
   python/branches/py3k/Lib/test/support.py
   python/branches/py3k/Lib/test/test_ssl.py

Modified: python/branches/py3k/Lib/test/support.py
==============================================================================
--- python/branches/py3k/Lib/test/support.py	(original)
+++ python/branches/py3k/Lib/test/support.py	Tue Sep  7 22:42:19 2010
@@ -35,7 +35,7 @@
     "check_warnings", "CleanImport", "EnvironmentVarGuard",
     "TransientResource", "captured_output", "captured_stdout",
     "time_out", "socket_peer_reset", "ioerror_peer_reset",
-    "run_with_locale", 'temp_umask',
+    "run_with_locale", 'temp_umask', "transient_internet",
     "set_memlimit", "bigmemtest", "bigaddrspacetest", "BasicTestRunner",
     "run_unittest", "run_doctest", "threading_setup", "threading_cleanup",
     "reap_children", "cpython_only", "check_impl_detail", "get_attribute",
@@ -775,23 +775,47 @@
             else:
                 raise ResourceDenied("an optional resource is not available")
 
-
 # Context managers that raise ResourceDenied when various issues
 # with the Internet connection manifest themselves as exceptions.
+# XXX deprecate these and use transient_internet() instead
 time_out = TransientResource(IOError, errno=errno.ETIMEDOUT)
 socket_peer_reset = TransientResource(socket.error, errno=errno.ECONNRESET)
 ioerror_peer_reset = TransientResource(IOError, errno=errno.ECONNRESET)
 
 
 @contextlib.contextmanager
-def transient_internet():
+def transient_internet(resource_name, *, timeout=30.0, errnos=()):
     """Return a context manager that raises ResourceDenied when various issues
     with the Internet connection manifest themselves as exceptions."""
-    time_out = TransientResource(IOError, errno=errno.ETIMEDOUT)
-    socket_peer_reset = TransientResource(socket.error, errno=errno.ECONNRESET)
-    ioerror_peer_reset = TransientResource(IOError, errno=errno.ECONNRESET)
-    with time_out, socket_peer_reset, ioerror_peer_reset:
+    denied = ResourceDenied("Resource '%s' is not available" % resource_name)
+    captured_errnos = errnos or (errno.ETIMEDOUT, errno.ECONNRESET)
+
+    def filter_error(err):
+        if (isinstance(err, socket.timeout) or
+            getattr(err, 'errno', None) in captured_errnos):
+            if not verbose:
+                sys.stderr.write(denied.args[0] + "\n")
+            raise denied from err
+
+    old_timeout = socket.getdefaulttimeout()
+    try:
+        if timeout is not None:
+            socket.setdefaulttimeout(timeout)
         yield
+    except IOError as err:
+        # socket.error inherits IOError
+        filter_error(err)
+        # urllib.request wraps the original socket.error with IOerror:
+        #
+        #    except socket.error as msg:
+        #        raise IOError('socket error', msg).with_traceback(sys.exc_info()[2])
+        if len(err.args) >= 2 and isinstance(err.args[1], socket.error):
+            filter_error(err.args[1])
+        raise
+    # XXX should we catch generic exceptions and look for their
+    # __cause__ or __context__?
+    finally:
+        socket.setdefaulttimeout(old_timeout)
 
 
 @contextlib.contextmanager

Modified: python/branches/py3k/Lib/test/test_ssl.py
==============================================================================
--- python/branches/py3k/Lib/test/test_ssl.py	(original)
+++ python/branches/py3k/Lib/test/test_ssl.py	Tue Sep  7 22:42:19 2010
@@ -477,10 +477,10 @@
         # NOTE: https://sha256.tbs-internet.com is another possible test host
         remote = ("sha2.hboeck.de", 443)
         sha256_cert = os.path.join(os.path.dirname(__file__), "sha256.pem")
-        s = ssl.wrap_socket(socket.socket(socket.AF_INET),
-                            cert_reqs=ssl.CERT_REQUIRED,
-                            ca_certs=sha256_cert,)
-        with support.transient_internet():
+        with support.transient_internet("sha2.hboeck.de"):
+            s = ssl.wrap_socket(socket.socket(socket.AF_INET),
+                                cert_reqs=ssl.CERT_REQUIRED,
+                                ca_certs=sha256_cert,)
             try:
                 s.connect(remote)
                 if support.verbose: