[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 a52a3509770f29f940cda9307704908949912276)

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

1 files changed, 60 insertions, 0 deletions

diff --git a/Lib/test/support/socket_helper.py b/Lib/test/support/socket_helper.py
index 7840923..e85d912 100644
--- a/Lib/test/support/socket_helper.py
+++ b/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)