diff options
author | Serhiy Storchaka <storchaka@gmail.com> | 2020-04-25 07:06:29 (GMT) |
---|---|---|
committer | GitHub <noreply@github.com> | 2020-04-25 07:06:29 (GMT) |
commit | 16994912c93e8e5db7365d48b75d67d3f70dd7b2 (patch) | |
tree | 248f177a93676406af6d6ae977bed868aa2d1a34 /Lib/test/test_asyncio | |
parent | 3c8a5b459d68b4337776ada1e04f5b33f90a2275 (diff) | |
download | cpython-16994912c93e8e5db7365d48b75d67d3f70dd7b2.zip cpython-16994912c93e8e5db7365d48b75d67d3f70dd7b2.tar.gz cpython-16994912c93e8e5db7365d48b75d67d3f70dd7b2.tar.bz2 |
bpo-40275: Avoid importing socket in test.support (GH-19603)
* Move socket related functions from test.support to socket_helper.
* Import socket, nntplib and urllib.error lazily in transient_internet().
* Remove importing multiprocess.
Diffstat (limited to 'Lib/test/test_asyncio')
-rw-r--r-- | Lib/test/test_asyncio/test_base_events.py | 15 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_events.py | 19 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_proactor_events.py | 3 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_sendfile.py | 13 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_server.py | 5 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_sock_lowlevel.py | 3 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_streams.py | 10 | ||||
-rw-r--r-- | Lib/test/test_asyncio/test_unix_events.py | 13 |
8 files changed, 44 insertions, 37 deletions
diff --git a/Lib/test/test_asyncio/test_base_events.py b/Lib/test/test_asyncio/test_base_events.py index fc832d3..4adcbf4 100644 --- a/Lib/test/test_asyncio/test_base_events.py +++ b/Lib/test/test_asyncio/test_base_events.py @@ -17,6 +17,7 @@ from asyncio import constants from test.test_asyncio import utils as test_utils from test import support from test.support.script_helper import assert_python_ok +from test.support import socket_helper MOCK_ANY = mock.ANY @@ -91,7 +92,7 @@ class BaseEventTests(test_utils.TestCase): self.assertIsNone( base_events._ipaddr_info('1.2.3.4', 1, UNSPEC, 0, 0)) - if support.IPV6_ENABLED: + if socket_helper.IPV6_ENABLED: # IPv4 address with family IPv6. self.assertIsNone( base_events._ipaddr_info('1.2.3.4', 1, INET6, STREAM, TCP)) @@ -1156,7 +1157,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): srv.close() self.loop.run_until_complete(srv.wait_closed()) - @unittest.skipUnless(support.IPV6_ENABLED, 'no IPv6 support') + @unittest.skipUnless(socket_helper.IPV6_ENABLED, 'no IPv6 support') def test_create_server_ipv6(self): async def main(): with self.assertWarns(DeprecationWarning): @@ -1288,7 +1289,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): t.close() test_utils.run_briefly(self.loop) # allow transport to close - if support.IPV6_ENABLED: + if socket_helper.IPV6_ENABLED: sock.family = socket.AF_INET6 coro = self.loop.create_connection(asyncio.Protocol, '::1', 80) t, p = self.loop.run_until_complete(coro) @@ -1307,7 +1308,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): t.close() test_utils.run_briefly(self.loop) # allow transport to close - @unittest.skipUnless(support.IPV6_ENABLED, 'no IPv6 support') + @unittest.skipUnless(socket_helper.IPV6_ENABLED, 'no IPv6 support') @unittest.skipIf(sys.platform.startswith('aix'), "bpo-25545: IPv6 scope id and getaddrinfo() behave differently on AIX") @patch_socket @@ -1639,7 +1640,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): self.assertRaises( OSError, self.loop.run_until_complete, coro) - @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled') + @unittest.skipUnless(socket_helper.IPV6_ENABLED, 'IPv6 not supported or enabled') def test_create_datagram_endpoint_no_matching_family(self): coro = self.loop.create_datagram_endpoint( asyncio.DatagramProtocol, @@ -1700,7 +1701,7 @@ class BaseEventLoopWithSelectorTests(test_utils.TestCase): self.loop.run_until_complete(protocol.done) self.assertEqual('CLOSED', protocol.state) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_create_datagram_endpoint_existing_sock_unix(self): with test_utils.unix_socket_path() as path: sock = socket.socket(socket.AF_UNIX, type=socket.SOCK_DGRAM) @@ -2015,7 +2016,7 @@ class BaseLoopSockSendfileTests(test_utils.TestCase): sock = self.make_socket() proto = self.MyProto(self.loop) server = self.run_loop(self.loop.create_server( - lambda: proto, support.HOST, 0, family=socket.AF_INET)) + lambda: proto, socket_helper.HOST, 0, family=socket.AF_INET)) addr = server.sockets[0].getsockname() for _ in range(10): diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py index 1f71c1f..aa4daca 100644 --- a/Lib/test/test_asyncio/test_events.py +++ b/Lib/test/test_asyncio/test_events.py @@ -32,6 +32,7 @@ from asyncio import proactor_events from asyncio import selector_events from test.test_asyncio import utils as test_utils from test import support +from test.support import socket_helper from test.support import ALWAYS_EQ, LARGEST, SMALLEST @@ -517,7 +518,7 @@ class EventLoopTestsMixin: lambda: MyProto(loop=self.loop), *httpd.address) self._basetest_create_connection(conn_fut) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_create_unix_connection(self): # Issue #20682: On Mac OS X Tiger, getsockname() returns a # zero-length address for UNIX socket. @@ -619,7 +620,7 @@ class EventLoopTestsMixin: self._test_create_ssl_connection(httpd, create_connection, peername=httpd.address) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket @unittest.skipIf(ssl is None, 'No ssl module') def test_create_ssl_unix_connection(self): # Issue #20682: On Mac OS X Tiger, getsockname() returns a @@ -638,7 +639,7 @@ class EventLoopTestsMixin: def test_create_connection_local_addr(self): with test_utils.run_test_server() as httpd: - port = support.find_unused_port() + port = socket_helper.find_unused_port() f = self.loop.create_connection( lambda: MyProto(loop=self.loop), *httpd.address, local_addr=(httpd.address[0], port)) @@ -843,7 +844,7 @@ class EventLoopTestsMixin: return server, path - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_create_unix_server(self): proto = MyProto(loop=self.loop) server, path = self._make_unix_server(lambda: proto) @@ -935,7 +936,7 @@ class EventLoopTestsMixin: # stop serving server.close() - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket @unittest.skipIf(ssl is None, 'No ssl module') def test_create_unix_server_ssl(self): proto = MyProto(loop=self.loop) @@ -995,7 +996,7 @@ class EventLoopTestsMixin: self.assertIsNone(proto.transport) server.close() - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket @unittest.skipIf(ssl is None, 'No ssl module') def test_create_unix_server_ssl_verify_failed(self): proto = MyProto(loop=self.loop) @@ -1055,7 +1056,7 @@ class EventLoopTestsMixin: self.assertIsNone(proto.transport) server.close() - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket @unittest.skipIf(ssl is None, 'No ssl module') def test_create_unix_server_ssl_verified(self): proto = MyProto(loop=self.loop) @@ -1148,7 +1149,7 @@ class EventLoopTestsMixin: server.close() - @unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 not supported or enabled') + @unittest.skipUnless(socket_helper.IPV6_ENABLED, 'IPv6 not supported or enabled') def test_create_server_dual_stack(self): f_proto = self.loop.create_future() @@ -1160,7 +1161,7 @@ class EventLoopTestsMixin: try_count = 0 while True: try: - port = support.find_unused_port() + port = socket_helper.find_unused_port() f = self.loop.create_server(TestMyProto, host=None, port=port) server = self.loop.run_until_complete(f) except OSError as ex: diff --git a/Lib/test/test_asyncio/test_proactor_events.py b/Lib/test/test_asyncio/test_proactor_events.py index 007039a..b5d1df9 100644 --- a/Lib/test/test_asyncio/test_proactor_events.py +++ b/Lib/test/test_asyncio/test_proactor_events.py @@ -13,6 +13,7 @@ from asyncio.proactor_events import _ProactorWritePipeTransport from asyncio.proactor_events import _ProactorDuplexPipeTransport from asyncio.proactor_events import _ProactorDatagramTransport from test import support +from test.support import socket_helper from test.test_asyncio import utils as test_utils @@ -950,7 +951,7 @@ class ProactorEventLoopUnixSockSendfileTests(test_utils.TestCase): def prepare(self): sock = self.make_socket() proto = self.MyProto(self.loop) - port = support.find_unused_port() + port = socket_helper.find_unused_port() srv_sock = self.make_socket(cleanup=False) srv_sock.bind(('127.0.0.1', port)) server = self.run_loop(self.loop.create_server( diff --git a/Lib/test/test_asyncio/test_sendfile.py b/Lib/test/test_asyncio/test_sendfile.py index 3b7f784..dbce199 100644 --- a/Lib/test/test_asyncio/test_sendfile.py +++ b/Lib/test/test_asyncio/test_sendfile.py @@ -10,6 +10,7 @@ from asyncio import base_events from asyncio import constants from unittest import mock from test import support +from test.support import socket_helper from test.test_asyncio import utils as test_utils try: @@ -163,9 +164,9 @@ class SockSendfileMixin(SendfileBase): def prepare_socksendfile(self): proto = MyProto(self.loop) - port = support.find_unused_port() + port = socket_helper.find_unused_port() srv_sock = self.make_socket(cleanup=False) - srv_sock.bind((support.HOST, port)) + srv_sock.bind((socket_helper.HOST, port)) server = self.run_loop(self.loop.create_server( lambda: proto, sock=srv_sock)) self.reduce_receive_buffer_size(srv_sock) @@ -240,7 +241,7 @@ class SendfileMixin(SendfileBase): # Note: sendfile via SSL transport is equal to sendfile fallback def prepare_sendfile(self, *, is_ssl=False, close_after=0): - port = support.find_unused_port() + port = socket_helper.find_unused_port() srv_proto = MySendfileProto(loop=self.loop, close_after=close_after) if is_ssl: @@ -252,17 +253,17 @@ class SendfileMixin(SendfileBase): srv_ctx = None cli_ctx = None srv_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - srv_sock.bind((support.HOST, port)) + srv_sock.bind((socket_helper.HOST, port)) server = self.run_loop(self.loop.create_server( lambda: srv_proto, sock=srv_sock, ssl=srv_ctx)) self.reduce_receive_buffer_size(srv_sock) if is_ssl: - server_hostname = support.HOST + server_hostname = socket_helper.HOST else: server_hostname = None cli_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - cli_sock.connect((support.HOST, port)) + cli_sock.connect((socket_helper.HOST, port)) cli_proto = MySendfileProto(loop=self.loop) tr, pr = self.run_loop(self.loop.create_connection( diff --git a/Lib/test/test_asyncio/test_server.py b/Lib/test/test_asyncio/test_server.py index d47ccc0..006ead2 100644 --- a/Lib/test/test_asyncio/test_server.py +++ b/Lib/test/test_asyncio/test_server.py @@ -4,6 +4,7 @@ import threading import unittest from test import support +from test.support import socket_helper from test.test_asyncio import utils as test_utils from test.test_asyncio import functional as func_tests @@ -47,7 +48,7 @@ class BaseStartServer(func_tests.FunctionalTestCaseMixin): with self.assertWarns(DeprecationWarning): srv = self.loop.run_until_complete(asyncio.start_server( - serve, support.HOSTv4, 0, loop=self.loop, start_serving=False)) + serve, socket_helper.HOSTv4, 0, loop=self.loop, start_serving=False)) self.assertFalse(srv.is_serving()) @@ -73,7 +74,7 @@ class SelectorStartServerTests(BaseStartServer, unittest.TestCase): def new_loop(self): return asyncio.SelectorEventLoop() - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_start_unix_server_1(self): HELLO_MSG = b'1' * 1024 * 5 + b'\n' started = threading.Event() diff --git a/Lib/test/test_asyncio/test_sock_lowlevel.py b/Lib/test/test_asyncio/test_sock_lowlevel.py index 89c2af9..2f2d5a4 100644 --- a/Lib/test/test_asyncio/test_sock_lowlevel.py +++ b/Lib/test/test_asyncio/test_sock_lowlevel.py @@ -5,6 +5,7 @@ from asyncio import proactor_events from itertools import cycle, islice from test.test_asyncio import utils as test_utils from test import support +from test.support import socket_helper class MyProto(asyncio.Protocol): @@ -225,7 +226,7 @@ class BaseSockTestsMixin: self.loop.run_until_complete( self._basetest_huge_content_recvinto(httpd.address)) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_unix_sock_client_ops(self): with test_utils.run_test_unix_server() as httpd: sock = socket.socket(socket.AF_UNIX) diff --git a/Lib/test/test_asyncio/test_streams.py b/Lib/test/test_asyncio/test_streams.py index 12bd536..1e9d115 100644 --- a/Lib/test/test_asyncio/test_streams.py +++ b/Lib/test/test_asyncio/test_streams.py @@ -9,7 +9,7 @@ import sys import threading import unittest from unittest import mock -from test import support +from test.support import socket_helper try: import ssl except ImportError: @@ -66,7 +66,7 @@ class StreamTests(test_utils.TestCase): loop=self.loop) self._basetest_open_connection(conn_fut) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_open_unix_connection(self): with test_utils.run_test_unix_server() as httpd: conn_fut = asyncio.open_unix_connection(httpd.address, @@ -99,7 +99,7 @@ class StreamTests(test_utils.TestCase): self._basetest_open_connection_no_loop_ssl(conn_fut) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket @unittest.skipIf(ssl is None, 'No ssl module') def test_open_unix_connection_no_loop_ssl(self): with test_utils.run_test_unix_server(use_ssl=True) as httpd: @@ -130,7 +130,7 @@ class StreamTests(test_utils.TestCase): loop=self.loop) self._basetest_open_connection_error(conn_fut) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_open_unix_connection_error(self): with test_utils.run_test_unix_server() as httpd: conn_fut = asyncio.open_unix_connection(httpd.address, @@ -653,7 +653,7 @@ class StreamTests(test_utils.TestCase): self.assertEqual(messages, []) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_start_unix_server(self): class MyServer: diff --git a/Lib/test/test_asyncio/test_unix_events.py b/Lib/test/test_asyncio/test_unix_events.py index 5487b7a..10bd46d 100644 --- a/Lib/test/test_asyncio/test_unix_events.py +++ b/Lib/test/test_asyncio/test_unix_events.py @@ -15,6 +15,7 @@ import threading import unittest from unittest import mock from test import support +from test.support import socket_helper if sys.platform == 'win32': raise unittest.SkipTest('UNIX only') @@ -273,7 +274,7 @@ class SelectorEventLoopUnixSocketTests(test_utils.TestCase): self.loop = asyncio.SelectorEventLoop() self.set_event_loop(self.loop) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_create_unix_server_existing_path_sock(self): with test_utils.unix_socket_path() as path: sock = socket.socket(socket.AF_UNIX) @@ -286,7 +287,7 @@ class SelectorEventLoopUnixSocketTests(test_utils.TestCase): srv.close() self.loop.run_until_complete(srv.wait_closed()) - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_create_unix_server_pathlib(self): with test_utils.unix_socket_path() as path: path = pathlib.Path(path) @@ -344,7 +345,7 @@ class SelectorEventLoopUnixSocketTests(test_utils.TestCase): @unittest.skipUnless(hasattr(socket, 'SOCK_NONBLOCK'), 'no socket.SOCK_NONBLOCK (linux only)') - @support.skip_unless_bind_unix_socket + @socket_helper.skip_unless_bind_unix_socket def test_create_unix_server_path_stream_bittype(self): sock = socket.socket( socket.AF_UNIX, socket.SOCK_STREAM | socket.SOCK_NONBLOCK) @@ -497,12 +498,12 @@ class SelectorEventLoopUnixSockSendfileTests(test_utils.TestCase): def prepare(self): sock = self.make_socket() proto = self.MyProto(self.loop) - port = support.find_unused_port() + port = socket_helper.find_unused_port() srv_sock = self.make_socket(cleanup=False) - srv_sock.bind((support.HOST, port)) + srv_sock.bind((socket_helper.HOST, port)) server = self.run_loop(self.loop.create_server( lambda: proto, sock=srv_sock)) - self.run_loop(self.loop.sock_connect(sock, (support.HOST, port))) + self.run_loop(self.loop.sock_connect(sock, (socket_helper.HOST, port))) self.run_loop(proto._ready) def cleanup(): |