bpo-40275: Move TransientResource to test_urllib2net (GH-20812)

Move TransientResource, time_out, socket_peer_reset and
ioerror_peer_reset from test.support to test_urllib2net.

Remove "import errno" from test.support.
This commit is contained in:
Victor Stinner 2020-06-11 18:26:23 +02:00 committed by GitHub
parent bcd7deed91
commit 311110abcd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
4 changed files with 37 additions and 48 deletions

View file

@ -1,3 +1,4 @@
import errno
import unittest
from test import support
from test.support import socket_helper
@ -39,6 +40,39 @@ _urlopen_with_retry = _wrap_with_retry_thrice(urllib.request.urlopen,
urllib.error.URLError)
class TransientResource(object):
"""Raise ResourceDenied if an exception is raised while the context manager
is in effect that matches the specified exception and attributes."""
def __init__(self, exc, **kwargs):
self.exc = exc
self.attrs = kwargs
def __enter__(self):
return self
def __exit__(self, type_=None, value=None, traceback=None):
"""If type_ is a subclass of self.exc and value has attributes matching
self.attrs, raise ResourceDenied. Otherwise let the exception
propagate (if any)."""
if type_ is not None and issubclass(self.exc, type_):
for attr, attr_value in self.attrs.items():
if not hasattr(value, attr):
break
if getattr(value, attr) != attr_value:
break
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(OSError, errno=errno.ETIMEDOUT)
socket_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
ioerror_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
class AuthTests(unittest.TestCase):
"""Tests urllib2 authentication features."""
@ -237,9 +271,9 @@ class OtherNetworkTests(unittest.TestCase):
raise
else:
try:
with support.time_out, \
support.socket_peer_reset, \
support.ioerror_peer_reset:
with time_out, \
socket_peer_reset, \
ioerror_peer_reset:
buf = f.read()
debug("read %d bytes" % len(buf))
except socket.timeout: