mirror of
https://github.com/python/cpython.git
synced 2025-07-24 11:44:31 +00:00
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).
This commit is contained in:
parent
60a9eea3f5
commit
a52a350977
7 changed files with 79 additions and 0 deletions
|
@ -3,6 +3,7 @@ import errno
|
|||
import os.path
|
||||
import socket
|
||||
import sys
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
|
@ -277,3 +278,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)
|
||||
|
|
Loading…
Add table
Add a link
Reference in a new issue