Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@contextlib.contextmanager
def _safe_wait(host, port):
"""
On systems where a loopback interface is not available and the
server is bound to all interfaces, it's difficult to determine
whether the server is in fact occupying the port. In this case,
just issue a warning and move on. See issue #1100.
"""
try:
yield
except portend.Timeout:
if host == portend.client_host(host):
raise
msg = 'Unable to verify that the server is bound on %r' % port
warnings.warn(msg)
def check_port_availability(host, port):
"""
Make sure the port is available for the server to start.
"""
try:
portend.free(host, port, timeout=PORT_AVAILABILITY_CHECK_TIMEOUT)
except portend.Timeout:
# Bypass check when socket activation is used
# https://manpages.debian.org/testing/libsystemd-dev/sd_listen_fds.3.en.html#ENVIRONMENT
if not os.environ.get("LISTEN_PID", None):
# Port is occupied
logger.error(
"Port {} is occupied.\n"
"Please check that you do not have other processes "
"running on this port and try again.\n".format(port)
)
sys.exit(1)
def _safe_wait(host, port):
"""
On systems where a loopback interface is not available and the
server is bound to all interfaces, it's difficult to determine
whether the server is in fact occupying the port. In this case,
just issue a warning and move on. See issue #1100.
"""
try:
yield
except portend.Timeout:
if host == portend.client_host(host):
raise
msg = 'Unable to verify that the server is bound on %r' % port
warnings.warn(msg)