[3.12] gh-109015: Add test.support.socket_helper.tcp_blackhole() (GH-109016) (#109041)

gh-109015: Add test.support.socket_helper.tcp_blackhole() (GH-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)

Co-authored-by: Victor Stinner <vstinner@python.org>

Lib/test/support/socket_helper.py

@@ -3,6 +3,7 @@ import errno
 import os.path
 import socket
 import sys
+import subprocess
 import tempfile
 import unittest
 
@@ -283,3 +284,62 @@ def create_unix_domain_name():
     """
     return tempfile.mktemp(prefix="test_python_", suffix='.sock',
                            dir=os.path.curdir)
+
+
+# consider that sysctl values should not change while tests are running
+_sysctl_cache = {}
+
+def _get_sysctl(name):
+    """Get a sysctl value as an integer."""
+    try:
+        return _sysctl_cache[name]
+    except KeyError:
+        pass
+
+    # At least Linux and FreeBSD support the "-n" option
+    cmd = ['sysctl', '-n', name]
+    proc = subprocess.run(cmd,
+                          stdout=subprocess.PIPE,
+                          stderr=subprocess.STDOUT,
+                          text=True)
+    if proc.returncode:
+        support.print_warning(f'{' '.join(cmd)!r} command failed with '
+                              f'exit code {proc.returncode}')
+        # cache the error to only log the warning once
+        _sysctl_cache[name] = None
+        return None
+    output = proc.stdout
+
+    # Parse '0\n' to get '0'
+    try:
+        value = int(output.strip())
+    except Exception as exc:
+        support.print_warning(f'Failed to parse {' '.join(cmd)!r} '
+                              f'command output {output!r}: {exc!r}')
+        # cache the error to only log the warning once
+        _sysctl_cache[name] = None
+        return None
+
+    _sysctl_cache[name] = value
+    return value
+
+
+def tcp_blackhole():
+    if not sys.platform.startswith('freebsd'):
+        return False
+
+    # gh-109015: test if FreeBSD TCP blackhole is enabled
+    value = _get_sysctl('net.inet.tcp.blackhole')
+    if value is None:
+        # don't skip if we fail to get the sysctl value
+        return False
+    return (value != 0)
+
+
+def skip_if_tcp_blackhole(test):
+    """Decorator skipping test if TCP blackhole is enabled."""
+    skip_if = unittest.skipIf(
+        tcp_blackhole(),
+        "TCP blackhole is enabled (sysctl net.inet.tcp.blackhole)"
+    )
+    return skip_if(test)

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()
@@ -1271,6 +1273,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'

Lib/test/test_imaplib.py

@@ -74,6 +74,7 @@ class TestImaplib(unittest.TestCase):
         for t in self.timevalues():
             imaplib.Time2Internaldate(t)
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_imap4_host_default_value(self):
         # Check whether the IMAP4_PORT is truly unavailable.
         with socket.socket() as s:

Lib/test/test_socket.py

@@ -5288,6 +5288,7 @@ class NetworkConnectionNoServer(unittest.TestCase):
         finally:
             socket.socket = old_socket
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_connect(self):
         port = socket_helper.find_unused_port()
         cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -5296,6 +5297,7 @@ class NetworkConnectionNoServer(unittest.TestCase):
             cli.connect((HOST, port))
         self.assertEqual(cm.exception.errno, errno.ECONNREFUSED)
 
+    @socket_helper.skip_if_tcp_blackhole
     def test_create_connection(self):
         # Issue #9792: errors raised by create_connection() should have
         # a proper errno attribute.

Misc/NEWS.d/next/Tests/2023-09-06-18-27-53.gh-issue-109015.1dS1AQ.rst

@@ -0,0 +1,6 @@
+Fix test_asyncio, test_imaplib and test_socket tests on FreeBSD if the TCP
+blackhole is enabled (``sysctl net.inet.tcp.blackhole``). Skip the few tests
+which failed with ``ETIMEDOUT`` which such non standard configuration.
+Currently, the `FreeBSD GCP image enables TCP and UDP blackhole
+<https://reviews.freebsd.org/D41751>`_ (``sysctl net.inet.tcp.blackhole=2``
+and ``sysctl net.inet.udp.blackhole=1``).  Patch by Victor Stinner.