Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_run_foreground_socket_error():
pidfile = os.path.join(os.getcwd(), "blackhole-test.pid")
cfile = create_config(
("listen=127.0.0.1:9000", "pidfile={}".format(pidfile))
)
c = Config(cfile).load()
with mock.patch("sys.argv", ["-c {}".format(cfile)]), mock.patch(
"blackhole.config.Config.test", return_value=c
), mock.patch("blackhole.config.warn_options"), mock.patch(
"atexit.register"
), mock.patch(
"blackhole.supervisor.Supervisor.close_socks"
), mock.patch(
"blackhole.supervisor.Supervisor.generate_servers",
side_effect=BlackholeRuntimeException,
), pytest.raises(
SystemExit
) as exc:
run()
assert exc.value.code == 77
def test_spawn_ipv4_fail():
cfile = create_config(("listen=:9999",))
Config(cfile).load()
loop = asyncio.new_event_loop()
with mock.patch("socket.socket.bind", side_effect=OSError), pytest.raises(
BlackholeRuntimeException
):
Supervisor(loop=loop)
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
def test_create_ipv6_socket_fails():
with mock.patch("socket.socket.bind", side_effect=OSError), pytest.raises(
BlackholeRuntimeException
):
_socket("::", 25, socket.AF_INET)
def test_create_server_ipv4_tls_bind_fails():
cfile = create_config(("tls_listen=127.0.0.1:9000",))
Config(cfile).load()
with mock.patch(
"socket.socket.bind", side_effect=OSError
) as mock_sock, pytest.raises(BlackholeRuntimeException):
server("127.0.0.1", 9000, socket.AF_INET)
assert mock_sock.called is True
assert mock_sock.call_count == 1
def test_spawn_ipv4_and_ipv6_fail():
cfile = create_config(("listen=:9999, :::9999",))
Config(cfile).load()
loop = asyncio.new_event_loop()
with mock.patch("socket.socket.bind", side_effect=OSError), pytest.raises(
BlackholeRuntimeException
):
Supervisor(loop=loop)
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
logger = logging.getLogger("blackhole")
if args.test:
config_test(args)
try:
config = Config(args.config_file).load().test()
config.args = args
warn_options(config)
daemon = Daemon(config.pidfile)
supervisor = Supervisor()
pid_permissions()
setgid()
setuid()
except (ConfigException, DaemonException) as err:
logger.critical(err)
raise SystemExit(os.EX_USAGE)
except BlackholeRuntimeException as err:
logger.critical(err)
raise SystemExit(os.EX_NOPERM)
if args.background:
try:
daemon.daemonize()
except DaemonException as err:
supervisor.close_socks()
logger.critical(err)
raise SystemExit(os.EX_NOPERM)
try:
supervisor.run()
except KeyboardInterrupt:
pass
raise SystemExit(os.EX_OK)
:type loop: :py:class:`syncio.unix_events._UnixSelectorEventLoop` or
:py:obj:`None` to get the current event loop using
:py:func:`asyncio.get_event_loop`.
:raises BlackholeRuntimeException: When an error occurs generating
servers.
"""
logger.debug("Initiating the supervisor")
self.config = Config()
self.loop = loop if loop is not None else asyncio.get_event_loop()
self.socks = []
self.workers = []
if setproctitle:
setproctitle.setproctitle("blackhole: master")
try:
self.generate_servers()
except BlackholeRuntimeException:
self.close_socks()
raise BlackholeRuntimeException()
"""
sock = socket.socket(family, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
except (AttributeError, OSError):
pass
if family == socket.AF_INET6:
sock.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 1)
try:
sock.bind((addr, port))
except OSError:
msg = f"Cannot bind to {addr}:{port}."
logger.critical(msg)
sock.close()
raise BlackholeRuntimeException(msg)
os.set_inheritable(sock.fileno(), True)
sock.listen(1024)
sock.setblocking(False)
return sock
:py:func:`asyncio.get_event_loop`.
:raises BlackholeRuntimeException: When an error occurs generating
servers.
"""
logger.debug("Initiating the supervisor")
self.config = Config()
self.loop = loop if loop is not None else asyncio.get_event_loop()
self.socks = []
self.workers = []
if setproctitle:
setproctitle.setproctitle("blackhole: master")
try:
self.generate_servers()
except BlackholeRuntimeException:
self.close_socks()
raise BlackholeRuntimeException()