diff options
Diffstat (limited to 'Lib/test')
| -rw-r--r-- | Lib/test/support/__init__.py | 88 | ||||
| -rw-r--r-- | Lib/test/support/socket_helper.py | 89 | ||||
| -rw-r--r-- | Lib/test/test_httplib.py | 10 | ||||
| -rw-r--r-- | Lib/test/test_imaplib.py | 20 | ||||
| -rw-r--r-- | Lib/test/test_nntplib.py | 4 | ||||
| -rw-r--r-- | Lib/test/test_robotparser.py | 2 | ||||
| -rw-r--r-- | Lib/test/test_site.py | 3 | ||||
| -rw-r--r-- | Lib/test/test_smtpnet.py | 13 | ||||
| -rwxr-xr-x | Lib/test/test_socket.py | 2 | ||||
| -rw-r--r-- | Lib/test/test_ssl.py | 4 | ||||
| -rw-r--r-- | Lib/test/test_timeout.py | 6 | ||||
| -rw-r--r-- | Lib/test/test_urllib2.py | 3 | ||||
| -rw-r--r-- | Lib/test/test_urllib2net.py | 29 | ||||
| -rw-r--r-- | Lib/test/test_urllibnet.py | 9 |
14 files changed, 145 insertions, 137 deletions
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 @@ __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.""" @@ -1387,90 +1385,6 @@ 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 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 @@ 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) 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 @@ 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) 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 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()) 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 @@ 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, 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 @@ 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() 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 @@ 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: 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 @@ 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() 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 @@ 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 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 @@ 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) 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 @@ 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) 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 @@ 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) 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 @@ 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) 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 @@ 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 |