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_delete_pid_exit():
pfile = create_file("test.pid", 123)
with mock.patch("os.getpid", return_value=123), mock.patch(
"atexit.register"
):
daemon = Daemon(pfile)
assert daemon.pid == 123
daemon._exit()
assert daemon.pid is None
def test_set_pid_valid_path():
pid = os.path.join(os.getcwd(), "fake.pid")
with mock.patch("os.getpid", return_value=666), mock.patch(
"atexit.register"
):
daemon = Daemon(pid)
assert daemon.pidfile == pid
assert daemon.pid == 666
def test_delete_pid():
pfile = create_file("test.pid", 123)
with mock.patch("os.getpid", return_value=123), mock.patch(
"atexit.register"
):
daemon = Daemon(pfile)
assert daemon.pid == 123
del daemon.pid
assert daemon.pid is None
def test_fork_error():
pfile = create_file("test.pid", 123)
with mock.patch("atexit.register"):
daemon = Daemon(pfile)
with mock.patch(
"os.fork", side_effect=OSError
) as mock_fork, pytest.raises(DaemonException):
daemon.fork()
assert mock_fork.called is True
def test_fork():
pfile = create_file("test.pid", 123)
with mock.patch("atexit.register"):
daemon = Daemon(pfile)
with mock.patch("os.fork", return_value=9999) as mock_fork, mock.patch(
"os._exit"
) as mock_exit:
daemon.fork()
assert mock_fork.called is True
assert mock_exit.called is True
def test_delete_pid_no_exists():
pfile = create_file("test.pid", 123)
daemon = Daemon(pfile)
with mock.patch("os.remove") as mock_rm, mock.patch(
"atexit.register"
), mock.patch("os.path.exists", return_value=False):
del daemon.pid
assert mock_rm.called is False
with mock.patch(
"builtins.open", side_effect=FileNotFoundError
), mock.patch("atexit.register"), pytest.raises(DaemonException):
Daemon("/fake/path.pid")
with mock.patch("builtins.open", side_effect=IOError), mock.patch(
"atexit.register"
), pytest.raises(DaemonException):
Daemon("/fake/path.pid")
with mock.patch(
"builtins.open", side_effect=PermissionError
), mock.patch("atexit.register"), pytest.raises(DaemonException):
Daemon("/fake/path.pid")
with mock.patch("builtins.open", side_effect=OSError), mock.patch(
"atexit.register"
), pytest.raises(DaemonException):
Daemon("/fake/path.pid")
def test_instantiated_but_not_daemonised():
pid = os.path.join(os.getcwd(), "fake.pid")
with mock.patch("os.getpid", return_value=666):
daemon = Daemon(pid)
assert daemon.pidfile == pid
assert daemon.pid == 666
def test_daemonise():
pfile = create_file("test.pid", 123)
with mock.patch("blackhole.daemon.Daemon.fork") as mock_fork, mock.patch(
"os.chdir"
) as mock_chdir, mock.patch("os.setsid") as mock_setsid, mock.patch(
"os.umask"
) as mock_umask, mock.patch(
"atexit.register"
) as mock_atexit, mock.patch(
"os.getpid", return_value=123
):
daemon = Daemon(pfile)
daemon.daemonize()
assert mock_fork.called is True
assert mock_chdir.called is True
assert mock_setsid.called is True
assert mock_umask.called is True
assert mock_atexit.called is True
:raises SystemExit: Exit code :py:obj:`os.EX_USAGE` when a configuration
error occurs, :py:obj:`os.EX_NOPERM` when a permission
error occurs or :py:obj:`os.EX_OK` when the program
exits cleanly.
"""
args = parse_cmd_args(sys.argv[1:])
configure_logs(args)
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)