[3.11] gh-109015: Add test.support.socket_helper.tcp_blackhole() (#109016) (#109042)

gh-109015: Add test.support.socket_helper.tcp_blackhole() (#109016)

Skip test_asyncio, test_imaplib and test_socket tests if FreeBSD TCP
blackhole is enabled (net.inet.tcp.blackhole=2).

(cherry picked from commit a52a350977)

Lib/test/test_asyncio/test_events.py

@@ -671,6 +671,7 @@ class EventLoopTestsMixin:
             self.assertEqual(port, expected)
             tr.close()
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_create_connection_local_addr_skip_different_family(self):
         # See https://github.com/python/cpython/issues/86508
         port1 = socket_helper.find_unused_port()
@@ -692,6 +693,7 @@ class EventLoopTestsMixin:
         with self.assertRaises(OSError):
             self.loop.run_until_complete(f)
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_create_connection_local_addr_nomatch_family(self):
         # See https://github.com/python/cpython/issues/86508
         port1 = socket_helper.find_unused_port()
@@ -1248,6 +1250,7 @@ class EventLoopTestsMixin:
 
         server.close()
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_server_close(self):
         f = self.loop.create_server(MyProto, '0.0.0.0', 0)
         server = self.loop.run_until_complete(f)

Lib/test/test_asyncio/test_sock_lowlevel.py

@@ -10,6 +10,10 @@ from test.test_asyncio import utils as test_utils
 from test import support
 from test.support import socket_helper
 
+if socket_helper.tcp_blackhole():
+    raise unittest.SkipTest('Not relevant to ProactorEventLoop')
+
+
 def tearDownModule():
     asyncio.set_event_loop_policy(None)
 

Lib/test/test_asyncio/test_sslproto.py

@@ -5,6 +5,7 @@ import socket
 import unittest
 import weakref
 from test import support
+from test.support import socket_helper
 from unittest import mock
 try:
     import ssl
@@ -350,6 +351,7 @@ class BaseStartTLS(func_tests.FunctionalTestCaseMixin):
         support.gc_collect()
         self.assertIsNone(client_context())
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_start_tls_client_buf_proto_1(self):
         HELLO_MSG = b'1' * self.PAYLOAD_SIZE
 
@@ -502,6 +504,7 @@ class BaseStartTLS(func_tests.FunctionalTestCaseMixin):
                 asyncio.wait_for(client(srv.addr),
                                  timeout=support.SHORT_TIMEOUT))
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_start_tls_server_1(self):
         HELLO_MSG = b'1' * self.PAYLOAD_SIZE
         ANSWER = b'answer'