traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

bpo-40275: Move transient_internet from test.support to socket_helper (GH-19711)

This commit is contained in:
Serhiy Storchaka 2020-04-29 10:36:20 +03:00 committed by GitHub
parent bb4a585d90
commit bfb1cf4465
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 153 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
Doc/library/test.rst
View File

@ -314,7 +314,7 @@ The :mod:`test.support` module defines the following constants:
Usually, a timeout using :data:`INTERNET_TIMEOUT` should not mark a test as
failed, but skip the test instead: see
:func:`~test.support.transient_internet`.
:func:`~test.support.socket_helper.transient_internet`.
Its default value is 1 minute.
@ -759,12 +759,6 @@ The :mod:`test.support` module defines the following functions:
A context manager that temporarily sets the process umask.
.. function:: transient_internet(resource_name, *, timeout=30.0, errnos=())
A context manager that raises :exc:`ResourceDenied` when various issues
with the internet connection manifest themselves as exceptions.
.. function:: disable_faulthandler()
A context manager that replaces ``sys.stderr`` with ``sys.__stderr__``.
@ -1488,6 +1482,13 @@ The :mod:`test.support.socket_helper` module provides support for socket tests.
sockets.
.. function:: transient_internet(resource_name, *, timeout=30.0, errnos=())
A context manager that raises :exc:`~test.support.ResourceDenied` when
various issues with the internet connection manifest themselves as
exceptions.
:mod:`test.support.script_helper` --- Utilities for the Python execution tests
==============================================================================

88
Lib/test/support/__init__.py
View File

@ -78,7 +78,7 @@ __all__ = [
"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 @@ SHORT_TIMEOUT = 30.0
# option.
LONG_TIMEOUT = 5 * 60.0
_NOT_SET = object()
class Error(Exception):
"""Base class for regression test exceptions."""
@ -1386,90 +1384,6 @@ socket_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
ioerror_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
@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

89
Lib/test/support/socket_helper.py
View File

@ -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 @@ def get_socket_conn_refused_errs():
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)

10
Lib/test/test_httplib.py
View File

@ -1629,7 +1629,7 @@ class HTTPSTest(TestCase):
# 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 @@ class HTTPSTest(TestCase):
# 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 @@ class HTTPSTest(TestCase):
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 @@ class HTTPSTest(TestCase):
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 @@ class HTTPSTest(TestCase):
# 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)

20
Lib/test/test_imaplib.py
View File

@ -10,7 +10,7 @@ import calendar
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 @@ class RemoteIMAPTest(unittest.TestCase):
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 @@ class RemoteIMAPTest(unittest.TestCase):
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 @@ class RemoteIMAP_STARTTLSTest(RemoteIMAPTest):
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 @@ class RemoteIMAP_SSLTest(RemoteIMAPTest):
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())

4
Lib/test/test_nntplib.py
View File

@ -246,7 +246,7 @@ class NetworkedNNTPTestsMixin:
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 @@ class NetworkedNNTPTests(NetworkedNNTPTestsMixin, unittest.TestCase):
@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,

2
Lib/test/test_robotparser.py
View File

@ -349,7 +349,7 @@ class NetworkTestCase(unittest.TestCase):
@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()

3
Lib/test/test_site.py
View File

@ -7,6 +7,7 @@ executing have not been removed.
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 @@ class ImportSideEffectTests(unittest.TestCase):
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:

13
Lib/test/test_smtpnet.py
View File

@ -1,5 +1,6 @@
import unittest
from test import support
from test.support import socket_helper
import smtplib
import socket
@ -28,7 +29,7 @@ class SmtpTest(unittest.TestCase):
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 @@ class SmtpSSLTest(unittest.TestCase):
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 @@ class SmtpSSLTest(unittest.TestCase):
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()

2
Lib/test/test_socket.py
View File

@ -1533,7 +1533,7 @@ class GeneralModuleTests(unittest.TestCase):
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

4
Lib/test/test_ssl.py
View File

@ -2246,7 +2246,7 @@ class NetworkedTests(unittest.TestCase):
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 @@ class NetworkedTests(unittest.TestCase):
@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)

6
Lib/test/test_timeout.py
View File

@ -20,7 +20,7 @@ def resolve_address(host, port):
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 @@ class TCPTimeoutTestCase(TimeoutTestCase):
# 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)

3
Lib/test/test_urllib2.py
View File

@ -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 @@ class MiscTests(unittest.TestCase):
@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)

29
Lib/test/test_urllib2net.py
View File

@ -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 @@ class CloseSocketTest(unittest.TestCase):
# 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 @@ class OtherNetworkTests(unittest.TestCase):
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 @@ class OtherNetworkTests(unittest.TestCase):
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 @@ class OtherNetworkTests(unittest.TestCase):
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 @@ class OtherNetworkTests(unittest.TestCase):
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 @@ class OtherNetworkTests(unittest.TestCase):
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 @@ class TimeoutTest(unittest.TestCase):
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 @@ class TimeoutTest(unittest.TestCase):
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 @@ class TimeoutTest(unittest.TestCase):
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 @@ class TimeoutTest(unittest.TestCase):
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 @@ class TimeoutTest(unittest.TestCase):
@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 @@ class TimeoutTest(unittest.TestCase):
@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 @@ class TimeoutTest(unittest.TestCase):
@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 @@ class TimeoutTest(unittest.TestCase):
@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)

9
Lib/test/test_urllibnet.py
View File

@ -1,5 +1,6 @@
import unittest
from test import support
from test.support import socket_helper
import contextlib
import socket
@ -27,7 +28,7 @@ class URLTimeoutTest(unittest.TestCase):
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 @@ class urlopenNetworkTests(unittest.TestCase):
@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 @@ class urlopenNetworkTests(unittest.TestCase):
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 @@ class urlretrieveNetworkTests(unittest.TestCase):
@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