bpo-40275: Move transient_internet from test.support to socket_helper (GH-19711)
diff --git a/Lib/test/support/__init__.py b/Lib/test/support/__init__.py
index ee5882f..bd21574 100644
--- a/Lib/test/support/__init__.py
+++ b/Lib/test/support/__init__.py
@@ -78,7 +78,7 @@
"requires_linux_version", "requires_mac_ver",
"check_syntax_error", "check_syntax_warning",
"TransientResource", "time_out", "socket_peer_reset", "ioerror_peer_reset",
- "transient_internet", "BasicTestRunner", "run_unittest", "run_doctest",
+ "BasicTestRunner", "run_unittest", "run_doctest",
"skip_unless_symlink", "requires_gzip", "requires_bz2", "requires_lzma",
"bigmemtest", "bigaddrspacetest", "cpython_only", "get_attribute",
"requires_IEEE_754", "skip_unless_xattr", "requires_zlib",
@@ -144,8 +144,6 @@
# option.
LONG_TIMEOUT = 5 * 60.0
-_NOT_SET = object()
-
class Error(Exception):
"""Base class for regression test exceptions."""
@@ -1387,90 +1385,6 @@
@contextlib.contextmanager
-def transient_internet(resource_name, *, timeout=_NOT_SET, errnos=()):
- """Return a context manager that raises ResourceDenied when various issues
- with the Internet connection manifest themselves as exceptions."""
- import socket
- import nntplib
- import urllib.error
- if timeout is _NOT_SET:
- timeout = INTERNET_TIMEOUT
-
- default_errnos = [
- ('ECONNREFUSED', 111),
- ('ECONNRESET', 104),
- ('EHOSTUNREACH', 113),
- ('ENETUNREACH', 101),
- ('ETIMEDOUT', 110),
- # socket.create_connection() fails randomly with
- # EADDRNOTAVAIL on Travis CI.
- ('EADDRNOTAVAIL', 99),
- ]
- default_gai_errnos = [
- ('EAI_AGAIN', -3),
- ('EAI_FAIL', -4),
- ('EAI_NONAME', -2),
- ('EAI_NODATA', -5),
- # Encountered when trying to resolve IPv6-only hostnames
- ('WSANO_DATA', 11004),
- ]
-
- denied = ResourceDenied("Resource %r is not available" % resource_name)
- captured_errnos = errnos
- gai_errnos = []
- if not captured_errnos:
- captured_errnos = [getattr(errno, name, num)
- for (name, num) in default_errnos]
- gai_errnos = [getattr(socket, name, num)
- for (name, num) in default_gai_errnos]
-
- def filter_error(err):
- n = getattr(err, 'errno', None)
- if (isinstance(err, socket.timeout) or
- (isinstance(err, socket.gaierror) and n in gai_errnos) or
- (isinstance(err, urllib.error.HTTPError) and
- 500 <= err.code <= 599) or
- (isinstance(err, urllib.error.URLError) and
- (("ConnectionRefusedError" in err.reason) or
- ("TimeoutError" in err.reason) or
- ("EOFError" in err.reason))) or
- n 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 nntplib.NNTPTemporaryError as err:
- if verbose:
- sys.stderr.write(denied.args[0] + "\n")
- raise denied from err
- except OSError as err:
- # urllib can wrap original socket errors multiple times (!), we must
- # unwrap to get at the original error.
- while True:
- a = err.args
- if len(a) >= 1 and isinstance(a[0], OSError):
- err = a[0]
- # The error can also be wrapped as args[1]:
- # except socket.error as msg:
- # raise OSError('socket error', msg).with_traceback(sys.exc_info()[2])
- elif len(a) >= 2 and isinstance(a[1], OSError):
- err = a[1]
- else:
- break
- filter_error(err)
- raise
- # XXX should we catch generic exceptions and look for their
- # __cause__ or __context__?
- finally:
- socket.setdefaulttimeout(old_timeout)
-
-
-@contextlib.contextmanager
def captured_output(stream_name):
"""Return a context manager used by captured_stdout/stdin/stderr
that temporarily replaces the sys stream *stream_name* with a StringIO."""
diff --git a/Lib/test/support/socket_helper.py b/Lib/test/support/socket_helper.py
index 5f4a7f1..b09c248 100644
--- a/Lib/test/support/socket_helper.py
+++ b/Lib/test/support/socket_helper.py
@@ -1,7 +1,11 @@
+import contextlib
import errno
import socket
import unittest
+from .. import support
+
+
HOST = "localhost"
HOSTv4 = "127.0.0.1"
HOSTv6 = "::1"
@@ -175,3 +179,88 @@
if not IPV6_ENABLED:
errors.append(errno.EAFNOSUPPORT)
return errors
+
+
+_NOT_SET = object()
+
+@contextlib.contextmanager
+def transient_internet(resource_name, *, timeout=_NOT_SET, errnos=()):
+ """Return a context manager that raises ResourceDenied when various issues
+ with the Internet connection manifest themselves as exceptions."""
+ import nntplib
+ import urllib.error
+ if timeout is _NOT_SET:
+ timeout = support.INTERNET_TIMEOUT
+
+ default_errnos = [
+ ('ECONNREFUSED', 111),
+ ('ECONNRESET', 104),
+ ('EHOSTUNREACH', 113),
+ ('ENETUNREACH', 101),
+ ('ETIMEDOUT', 110),
+ # socket.create_connection() fails randomly with
+ # EADDRNOTAVAIL on Travis CI.
+ ('EADDRNOTAVAIL', 99),
+ ]
+ default_gai_errnos = [
+ ('EAI_AGAIN', -3),
+ ('EAI_FAIL', -4),
+ ('EAI_NONAME', -2),
+ ('EAI_NODATA', -5),
+ # Encountered when trying to resolve IPv6-only hostnames
+ ('WSANO_DATA', 11004),
+ ]
+
+ denied = support.ResourceDenied("Resource %r is not available" % resource_name)
+ captured_errnos = errnos
+ gai_errnos = []
+ if not captured_errnos:
+ captured_errnos = [getattr(errno, name, num)
+ for (name, num) in default_errnos]
+ gai_errnos = [getattr(socket, name, num)
+ for (name, num) in default_gai_errnos]
+
+ def filter_error(err):
+ n = getattr(err, 'errno', None)
+ if (isinstance(err, socket.timeout) or
+ (isinstance(err, socket.gaierror) and n in gai_errnos) or
+ (isinstance(err, urllib.error.HTTPError) and
+ 500 <= err.code <= 599) or
+ (isinstance(err, urllib.error.URLError) and
+ (("ConnectionRefusedError" in err.reason) or
+ ("TimeoutError" in err.reason) or
+ ("EOFError" in err.reason))) or
+ n in captured_errnos):
+ if not support.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 nntplib.NNTPTemporaryError as err:
+ if support.verbose:
+ sys.stderr.write(denied.args[0] + "\n")
+ raise denied from err
+ except OSError as err:
+ # urllib can wrap original socket errors multiple times (!), we must
+ # unwrap to get at the original error.
+ while True:
+ a = err.args
+ if len(a) >= 1 and isinstance(a[0], OSError):
+ err = a[0]
+ # The error can also be wrapped as args[1]:
+ # except socket.error as msg:
+ # raise OSError('socket error', msg).with_traceback(sys.exc_info()[2])
+ elif len(a) >= 2 and isinstance(a[1], OSError):
+ err = a[1]
+ else:
+ break
+ filter_error(err)
+ raise
+ # XXX should we catch generic exceptions and look for their
+ # __cause__ or __context__?
+ finally:
+ socket.setdefaulttimeout(old_timeout)
diff --git a/Lib/test/test_httplib.py b/Lib/test/test_httplib.py
index 6b7a9de..e95487b 100644
--- a/Lib/test/test_httplib.py
+++ b/Lib/test/test_httplib.py
@@ -1629,7 +1629,7 @@
# Default settings: requires a valid cert from a trusted CA
import ssl
support.requires('network')
- with support.transient_internet('self-signed.pythontest.net'):
+ with socket_helper.transient_internet('self-signed.pythontest.net'):
h = client.HTTPSConnection('self-signed.pythontest.net', 443)
with self.assertRaises(ssl.SSLError) as exc_info:
h.request('GET', '/')
@@ -1639,7 +1639,7 @@
# Switch off cert verification
import ssl
support.requires('network')
- with support.transient_internet('self-signed.pythontest.net'):
+ with socket_helper.transient_internet('self-signed.pythontest.net'):
context = ssl._create_unverified_context()
h = client.HTTPSConnection('self-signed.pythontest.net', 443,
context=context)
@@ -1653,7 +1653,7 @@
def test_networked_trusted_by_default_cert(self):
# Default settings: requires a valid cert from a trusted CA
support.requires('network')
- with support.transient_internet('www.python.org'):
+ with socket_helper.transient_internet('www.python.org'):
h = client.HTTPSConnection('www.python.org', 443)
h.request('GET', '/')
resp = h.getresponse()
@@ -1667,7 +1667,7 @@
import ssl
support.requires('network')
selfsigned_pythontestdotnet = 'self-signed.pythontest.net'
- with support.transient_internet(selfsigned_pythontestdotnet):
+ with socket_helper.transient_internet(selfsigned_pythontestdotnet):
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
self.assertEqual(context.verify_mode, ssl.CERT_REQUIRED)
self.assertEqual(context.check_hostname, True)
@@ -1699,7 +1699,7 @@
# We feed a "CA" cert that is unrelated to the server's cert
import ssl
support.requires('network')
- with support.transient_internet('self-signed.pythontest.net'):
+ with socket_helper.transient_internet('self-signed.pythontest.net'):
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.load_verify_locations(CERT_localhost)
h = client.HTTPSConnection('self-signed.pythontest.net', 443, context=context)
diff --git a/Lib/test/test_imaplib.py b/Lib/test/test_imaplib.py
index ce60156..d1e3550 100644
--- a/Lib/test/test_imaplib.py
+++ b/Lib/test/test_imaplib.py
@@ -10,7 +10,7 @@
import threading
import socket
-from test.support import (reap_threads, verbose, transient_internet,
+from test.support import (reap_threads, verbose,
run_with_tz, run_with_locale, cpython_only)
from test.support import hashlib_helper
import unittest
@@ -968,16 +968,16 @@
imap_class = imaplib.IMAP4
def setUp(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
self.server = self.imap_class(self.host, self.port)
def tearDown(self):
if self.server is not None:
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
self.server.logout()
def test_logincapa(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
for cap in self.server.capabilities:
self.assertIsInstance(cap, str)
self.assertIn('LOGINDISABLED', self.server.capabilities)
@@ -986,7 +986,7 @@
self.assertEqual(rs[0], 'OK')
def test_logout(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
rs = self.server.logout()
self.server = None
self.assertEqual(rs[0], 'BYE', rs)
@@ -999,7 +999,7 @@
def setUp(self):
super().setUp()
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
rs = self.server.starttls()
self.assertEqual(rs[0], 'OK')
@@ -1039,24 +1039,24 @@
server.logout()
def test_logincapa(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
_server = self.imap_class(self.host, self.port)
self.check_logincapa(_server)
def test_logout(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
_server = self.imap_class(self.host, self.port)
rs = _server.logout()
self.assertEqual(rs[0], 'BYE', rs)
def test_ssl_context_certfile_exclusive(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
self.assertRaises(
ValueError, self.imap_class, self.host, self.port,
certfile=CERTFILE, ssl_context=self.create_ssl_context())
def test_ssl_context_keyfile_exclusive(self):
- with transient_internet(self.host):
+ with socket_helper.transient_internet(self.host):
self.assertRaises(
ValueError, self.imap_class, self.host, self.port,
keyfile=CERTFILE, ssl_context=self.create_ssl_context())
diff --git a/Lib/test/test_nntplib.py b/Lib/test/test_nntplib.py
index 2a5a0b9..8d29681 100644
--- a/Lib/test/test_nntplib.py
+++ b/Lib/test/test_nntplib.py
@@ -246,7 +246,7 @@
def wrap_meth(meth):
@functools.wraps(meth)
def wrapped(self):
- with support.transient_internet(self.NNTP_HOST):
+ with socket_helper.transient_internet(self.NNTP_HOST):
meth(self)
return wrapped
for name in dir(cls):
@@ -315,7 +315,7 @@
@classmethod
def setUpClass(cls):
support.requires("network")
- with support.transient_internet(cls.NNTP_HOST):
+ with socket_helper.transient_internet(cls.NNTP_HOST):
try:
cls.server = cls.NNTP_CLASS(cls.NNTP_HOST,
timeout=support.INTERNET_TIMEOUT,
diff --git a/Lib/test/test_robotparser.py b/Lib/test/test_robotparser.py
index 9d4764e..a3112b8 100644
--- a/Lib/test/test_robotparser.py
+++ b/Lib/test/test_robotparser.py
@@ -349,7 +349,7 @@
@classmethod
def setUpClass(cls):
support.requires('network')
- with support.transient_internet(cls.base_url):
+ with socket_helper.transient_internet(cls.base_url):
cls.parser = urllib.robotparser.RobotFileParser(cls.robots_txt)
cls.parser.read()
diff --git a/Lib/test/test_site.py b/Lib/test/test_site.py
index 12e357c..957e7a4 100644
--- a/Lib/test/test_site.py
+++ b/Lib/test/test_site.py
@@ -7,6 +7,7 @@
import unittest
import test.support
from test import support
+from test.support import socket_helper
from test.support import (captured_stderr, TESTFN, EnvironmentVarGuard,
change_cwd)
import builtins
@@ -509,7 +510,7 @@
url = license._Printer__data.split()[1]
req = urllib.request.Request(url, method='HEAD')
try:
- with test.support.transient_internet(url):
+ with socket_helper.transient_internet(url):
with urllib.request.urlopen(req) as data:
code = data.getcode()
except urllib.error.HTTPError as e:
diff --git a/Lib/test/test_smtpnet.py b/Lib/test/test_smtpnet.py
index b69cd9d..74a00a9 100644
--- a/Lib/test/test_smtpnet.py
+++ b/Lib/test/test_smtpnet.py
@@ -1,5 +1,6 @@
import unittest
from test import support
+from test.support import socket_helper
import smtplib
import socket
@@ -28,7 +29,7 @@
context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
- with support.transient_internet(self.testServer):
+ with socket_helper.transient_internet(self.testServer):
server = smtplib.SMTP(self.testServer, self.remotePort)
try:
server.starttls(context=context)
@@ -47,14 +48,14 @@
def test_connect(self):
support.get_attribute(smtplib, 'SMTP_SSL')
- with support.transient_internet(self.testServer):
+ with socket_helper.transient_internet(self.testServer):
server = smtplib.SMTP_SSL(self.testServer, self.remotePort)
server.ehlo()
server.quit()
def test_connect_default_port(self):
support.get_attribute(smtplib, 'SMTP_SSL')
- with support.transient_internet(self.testServer):
+ with socket_helper.transient_internet(self.testServer):
server = smtplib.SMTP_SSL(self.testServer)
server.ehlo()
server.quit()
@@ -64,20 +65,20 @@
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE
support.get_attribute(smtplib, 'SMTP_SSL')
- with support.transient_internet(self.testServer):
+ with socket_helper.transient_internet(self.testServer):
server = smtplib.SMTP_SSL(self.testServer, self.remotePort, context=context)
server.ehlo()
server.quit()
def test_connect_using_sslcontext_verified(self):
- with support.transient_internet(self.testServer):
+ with socket_helper.transient_internet(self.testServer):
can_verify = check_ssl_verifiy(self.testServer, self.remotePort)
if not can_verify:
self.skipTest("SSL certificate can't be verified")
support.get_attribute(smtplib, 'SMTP_SSL')
context = ssl.create_default_context()
- with support.transient_internet(self.testServer):
+ with socket_helper.transient_internet(self.testServer):
server = smtplib.SMTP_SSL(self.testServer, self.remotePort, context=context)
server.ehlo()
server.quit()
diff --git a/Lib/test/test_socket.py b/Lib/test/test_socket.py
index a70e282..87ae2e1 100755
--- a/Lib/test/test_socket.py
+++ b/Lib/test/test_socket.py
@@ -1533,7 +1533,7 @@
def test_idna(self):
# Check for internet access before running test
# (issue #12804, issue #25138).
- with support.transient_internet('python.org'):
+ with socket_helper.transient_internet('python.org'):
socket.gethostbyname('python.org')
# these should all be successful
diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
index dafdb6c..5d496c6 100644
--- a/Lib/test/test_ssl.py
+++ b/Lib/test/test_ssl.py
@@ -2246,7 +2246,7 @@
def test_timeout_connect_ex(self):
# Issue #12065: on a timeout, connect_ex() should return the original
# errno (mimicking the behaviour of non-SSL sockets).
- with support.transient_internet(REMOTE_HOST):
+ with socket_helper.transient_internet(REMOTE_HOST):
s = test_wrap_socket(socket.socket(socket.AF_INET),
cert_reqs=ssl.CERT_REQUIRED,
do_handshake_on_connect=False)
@@ -2259,7 +2259,7 @@
@unittest.skipUnless(socket_helper.IPV6_ENABLED, 'Needs IPv6')
def test_get_server_certificate_ipv6(self):
- with support.transient_internet('ipv6.google.com'):
+ with socket_helper.transient_internet('ipv6.google.com'):
_test_get_server_certificate(self, 'ipv6.google.com', 443)
_test_get_server_certificate_fail(self, 'ipv6.google.com', 443)
diff --git a/Lib/test/test_timeout.py b/Lib/test/test_timeout.py
index c0952c7..ac803f5 100644
--- a/Lib/test/test_timeout.py
+++ b/Lib/test/test_timeout.py
@@ -20,7 +20,7 @@
We must perform name resolution before timeout tests, otherwise it will be
performed by connect().
"""
- with support.transient_internet(host):
+ with socket_helper.transient_internet(host):
return socket.getaddrinfo(host, port, socket.AF_INET,
socket.SOCK_STREAM)[0][4]
@@ -230,12 +230,12 @@
# All that hard work just to test if connect times out in 0.001s ;-)
self.addr_remote = blackhole
- with support.transient_internet(self.addr_remote[0]):
+ with socket_helper.transient_internet(self.addr_remote[0]):
self._sock_operation(1, 0.001, 'connect', self.addr_remote)
def testRecvTimeout(self):
# Test recv() timeout
- with support.transient_internet(self.addr_remote[0]):
+ with socket_helper.transient_internet(self.addr_remote[0]):
self.sock.connect(self.addr_remote)
self._sock_operation(1, 1.5, 'recv', 1024)
diff --git a/Lib/test/test_urllib2.py b/Lib/test/test_urllib2.py
index e69ac3e..cbfa9ba 100644
--- a/Lib/test/test_urllib2.py
+++ b/Lib/test/test_urllib2.py
@@ -1,5 +1,6 @@
import unittest
from test import support
+from test.support import socket_helper
from test import test_urllib
import os
@@ -1776,7 +1777,7 @@
@unittest.skipUnless(support.is_resource_enabled('network'),
'test requires network access')
def test_issue16464(self):
- with support.transient_internet("http://www.example.com/"):
+ with socket_helper.transient_internet("http://www.example.com/"):
opener = urllib.request.build_opener()
request = urllib.request.Request("http://www.example.com/")
self.assertEqual(None, request.data)
diff --git a/Lib/test/test_urllib2net.py b/Lib/test/test_urllib2net.py
index b3a5e89..ba4c500 100644
--- a/Lib/test/test_urllib2net.py
+++ b/Lib/test/test_urllib2net.py
@@ -1,5 +1,6 @@
import unittest
from test import support
+from test.support import socket_helper
from test.test_urllib2 import sanepathname2url
import os
@@ -86,7 +87,7 @@
# calling .close() on urllib2's response objects should close the
# underlying socket
url = support.TEST_HTTP_URL
- with support.transient_internet(url):
+ with socket_helper.transient_internet(url):
response = _urlopen_with_retry(url)
sock = response.fp
self.assertFalse(sock.closed)
@@ -159,7 +160,7 @@
def test_urlwithfrag(self):
urlwith_frag = "http://www.pythontest.net/index.html#frag"
- with support.transient_internet(urlwith_frag):
+ with socket_helper.transient_internet(urlwith_frag):
req = urllib.request.Request(urlwith_frag)
res = urllib.request.urlopen(req)
self.assertEqual(res.geturl(),
@@ -167,7 +168,7 @@
def test_redirect_url_withfrag(self):
redirect_url_with_frag = "http://www.pythontest.net/redir/with_frag/"
- with support.transient_internet(redirect_url_with_frag):
+ with socket_helper.transient_internet(redirect_url_with_frag):
req = urllib.request.Request(redirect_url_with_frag)
res = urllib.request.urlopen(req)
self.assertEqual(res.geturl(),
@@ -175,7 +176,7 @@
def test_custom_headers(self):
url = support.TEST_HTTP_URL
- with support.transient_internet(url):
+ with socket_helper.transient_internet(url):
opener = urllib.request.build_opener()
request = urllib.request.Request(url)
self.assertFalse(request.header_items())
@@ -193,7 +194,7 @@
URL = 'http://www.imdb.com' # mangles Connection:close
- with support.transient_internet(URL):
+ with socket_helper.transient_internet(URL):
try:
with urllib.request.urlopen(URL) as res:
pass
@@ -223,7 +224,7 @@
else:
req = expected_err = None
- with support.transient_internet(url):
+ with socket_helper.transient_internet(url):
try:
f = urlopen(url, req, support.INTERNET_TIMEOUT)
# urllib.error.URLError is a subclass of OSError
@@ -265,7 +266,7 @@
def test_http_basic(self):
self.assertIsNone(socket.getdefaulttimeout())
url = support.TEST_HTTP_URL
- with support.transient_internet(url, timeout=None):
+ with socket_helper.transient_internet(url, timeout=None):
u = _urlopen_with_retry(url)
self.addCleanup(u.close)
self.assertIsNone(u.fp.raw._sock.gettimeout())
@@ -273,7 +274,7 @@
def test_http_default_timeout(self):
self.assertIsNone(socket.getdefaulttimeout())
url = support.TEST_HTTP_URL
- with support.transient_internet(url):
+ with socket_helper.transient_internet(url):
socket.setdefaulttimeout(60)
try:
u = _urlopen_with_retry(url)
@@ -285,7 +286,7 @@
def test_http_no_timeout(self):
self.assertIsNone(socket.getdefaulttimeout())
url = support.TEST_HTTP_URL
- with support.transient_internet(url):
+ with socket_helper.transient_internet(url):
socket.setdefaulttimeout(60)
try:
u = _urlopen_with_retry(url, timeout=None)
@@ -296,7 +297,7 @@
def test_http_timeout(self):
url = support.TEST_HTTP_URL
- with support.transient_internet(url):
+ with socket_helper.transient_internet(url):
u = _urlopen_with_retry(url, timeout=120)
self.addCleanup(u.close)
self.assertEqual(u.fp.raw._sock.gettimeout(), 120)
@@ -306,7 +307,7 @@
@skip_ftp_test_on_travis
def test_ftp_basic(self):
self.assertIsNone(socket.getdefaulttimeout())
- with support.transient_internet(self.FTP_HOST, timeout=None):
+ with socket_helper.transient_internet(self.FTP_HOST, timeout=None):
u = _urlopen_with_retry(self.FTP_HOST)
self.addCleanup(u.close)
self.assertIsNone(u.fp.fp.raw._sock.gettimeout())
@@ -314,7 +315,7 @@
@skip_ftp_test_on_travis
def test_ftp_default_timeout(self):
self.assertIsNone(socket.getdefaulttimeout())
- with support.transient_internet(self.FTP_HOST):
+ with socket_helper.transient_internet(self.FTP_HOST):
socket.setdefaulttimeout(60)
try:
u = _urlopen_with_retry(self.FTP_HOST)
@@ -326,7 +327,7 @@
@skip_ftp_test_on_travis
def test_ftp_no_timeout(self):
self.assertIsNone(socket.getdefaulttimeout())
- with support.transient_internet(self.FTP_HOST):
+ with socket_helper.transient_internet(self.FTP_HOST):
socket.setdefaulttimeout(60)
try:
u = _urlopen_with_retry(self.FTP_HOST, timeout=None)
@@ -337,7 +338,7 @@
@skip_ftp_test_on_travis
def test_ftp_timeout(self):
- with support.transient_internet(self.FTP_HOST):
+ with socket_helper.transient_internet(self.FTP_HOST):
u = _urlopen_with_retry(self.FTP_HOST, timeout=60)
self.addCleanup(u.close)
self.assertEqual(u.fp.fp.raw._sock.gettimeout(), 60)
diff --git a/Lib/test/test_urllibnet.py b/Lib/test/test_urllibnet.py
index 422d529..28680aa 100644
--- a/Lib/test/test_urllibnet.py
+++ b/Lib/test/test_urllibnet.py
@@ -1,5 +1,6 @@
import unittest
from test import support
+from test.support import socket_helper
import contextlib
import socket
@@ -27,7 +28,7 @@
self.addCleanup(urllib.request.urlcleanup)
domain = urllib.parse.urlparse(support.TEST_HTTP_URL).netloc
- with support.transient_internet(domain):
+ with socket_helper.transient_internet(domain):
f = urllib.request.urlopen(support.TEST_HTTP_URL)
f.read()
@@ -56,7 +57,7 @@
@contextlib.contextmanager
def urlopen(self, *args, **kwargs):
resource = args[0]
- with support.transient_internet(resource):
+ with socket_helper.transient_internet(resource):
r = urllib.request.urlopen(*args, **kwargs)
try:
yield r
@@ -98,7 +99,7 @@
def test_getcode(self):
# test getcode() with the fancy opener to get 404 error codes
URL = self.url + "XXXinvalidXXX"
- with support.transient_internet(URL):
+ with socket_helper.transient_internet(URL):
with self.assertWarns(DeprecationWarning):
open_url = urllib.request.FancyURLopener().open(URL)
try:
@@ -156,7 +157,7 @@
@contextlib.contextmanager
def urlretrieve(self, *args, **kwargs):
resource = args[0]
- with support.transient_internet(resource):
+ with socket_helper.transient_internet(resource):
file_location, info = urllib.request.urlretrieve(*args, **kwargs)
try:
yield file_location, info