Refactored Daemon helper, added status command and added unit tests for all of it
This commit is contained in:
Gina Häußge
parent
4476545a55
commit
56805ab13d
3 changed files with 609 additions and 71 deletions
|
|
@ -70,8 +70,8 @@ def serve_command(obj, host, port, logging, allow_root, debug):
|
|||
help="Allow OctoPrint to run as user root.")
|
||||
@click.option("--debug", is_flag=True,
|
||||
help="Enable debug mode")
|
||||
@click.argument("command", type=click.Choice(["start", "stop", "restart"]),
|
||||
metavar="start|stop|restart")
|
||||
@click.argument("command", type=click.Choice(["start", "stop", "restart", "status"]),
|
||||
metavar="start|stop|restart|status")
|
||||
@pass_octoprint_ctx
|
||||
def daemon_command(octoprint_ctx, pid, host, port, logging, allow_root, debug, command):
|
||||
"""
|
||||
|
|
@ -115,5 +115,7 @@ def daemon_command(octoprint_ctx, pid, host, port, logging, allow_root, debug, c
|
|||
octoprint_daemon.stop()
|
||||
elif command == "restart":
|
||||
octoprint_daemon.restart()
|
||||
elif command == "status":
|
||||
octoprint_daemon.status()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,46 +1,65 @@
|
|||
# coding=utf-8
|
||||
"""
|
||||
Generic linux daemon base class for python 3.x
|
||||
Generic linux daemon base class
|
||||
|
||||
Originally from http://www.jejik.com/articles/2007/02/a_simple_unix_linux_daemon_in_python/#c35
|
||||
"""
|
||||
|
||||
from __future__ import (print_function, absolute_import)
|
||||
import sys, os, time, signal
|
||||
|
||||
class Daemon:
|
||||
"""A generic daemon class.
|
||||
"""
|
||||
A generic daemon class.
|
||||
|
||||
Usage: subclass the daemon class and override the run() method."""
|
||||
Usage: subclass the daemon class and override the run() method.
|
||||
|
||||
def __init__(self, pidfile): self.pidfile = pidfile
|
||||
|
||||
def daemonize(self):
|
||||
"""Deamonize class. UNIX double fork mechanism."""
|
||||
If you want to log the output to someplace different that stdout and stderr,
|
||||
also override the echo() and error() methods.
|
||||
"""
|
||||
|
||||
try:
|
||||
pid = os.fork()
|
||||
def __init__(self, pidfile):
|
||||
self.pidfile = pidfile
|
||||
|
||||
def _daemonize(self):
|
||||
"""Daemonize class. UNIX double fork mechanism."""
|
||||
|
||||
self._double_fork()
|
||||
self._redirect_io()
|
||||
|
||||
# write pidfile
|
||||
pid = str(os.getpid())
|
||||
self.set_pid(pid)
|
||||
|
||||
# register listener for SIGTERM
|
||||
signal.signal(signal.SIGTERM, self._on_sigterm)
|
||||
|
||||
def _double_fork(self):
|
||||
try:
|
||||
pid = os.fork()
|
||||
if pid > 0:
|
||||
# exit first parent
|
||||
sys.exit(0)
|
||||
except OSError as err:
|
||||
sys.stderr.write('fork #1 failed: {0}\n'.format(err))
|
||||
sys.exit(0)
|
||||
except OSError as err:
|
||||
self.error("First fork failed: {}".format(str(err)))
|
||||
sys.exit(1)
|
||||
|
||||
# decouple from parent environment
|
||||
os.chdir('/')
|
||||
os.setsid()
|
||||
os.umask(002)
|
||||
|
||||
# do second fork
|
||||
try:
|
||||
pid = os.fork()
|
||||
if pid > 0:
|
||||
|
||||
# decouple from parent environment
|
||||
os.chdir('/')
|
||||
os.setsid()
|
||||
os.umask(002)
|
||||
|
||||
# do second fork
|
||||
try:
|
||||
pid = os.fork()
|
||||
if pid > 0:
|
||||
# exit from second parent
|
||||
sys.exit(0)
|
||||
except OSError as err:
|
||||
sys.stderr.write('fork #2 failed: {0}\n'.format(err))
|
||||
sys.exit(1)
|
||||
|
||||
sys.exit(0)
|
||||
except OSError as err:
|
||||
self.error("Second fork failed: {}".format(str(err)))
|
||||
sys.exit(1)
|
||||
|
||||
def _redirect_io(self):
|
||||
# redirect standard file descriptors
|
||||
sys.stdout.flush()
|
||||
sys.stderr.flush()
|
||||
|
|
@ -51,57 +70,39 @@ class Daemon:
|
|||
os.dup2(si.fileno(), sys.stdin.fileno())
|
||||
os.dup2(so.fileno(), sys.stdout.fileno())
|
||||
os.dup2(se.fileno(), sys.stderr.fileno())
|
||||
|
||||
# write pidfile
|
||||
pid = str(os.getpid())
|
||||
with open(self.pidfile,'w+') as f:
|
||||
f.write(pid + '\n')
|
||||
|
||||
# register listener for SIGTERM
|
||||
signal.signal(signal.SIGTERM, self.term)
|
||||
|
||||
def term(self, _signo, _stack_frame):
|
||||
os.remove(self.pidfile)
|
||||
def _on_sigterm(self, _signo, _stack_frame):
|
||||
"""Signal handler for SIGTERM, deletes the pidfile."""
|
||||
self.remove_pidfile()
|
||||
sys.exit(0)
|
||||
|
||||
def start(self):
|
||||
"""Start the daemon."""
|
||||
|
||||
# Check for a pidfile to see if the daemon already runs
|
||||
try:
|
||||
with open(self.pidfile,'r') as pf:
|
||||
|
||||
pid = int(pf.read().strip())
|
||||
except IOError:
|
||||
pid = None
|
||||
|
||||
pid = self.get_pid()
|
||||
if pid:
|
||||
message = "pidfile {0} already exist. " + \
|
||||
"Daemon already running?\n"
|
||||
sys.stderr.write(message.format(self.pidfile))
|
||||
self.error("pidfile {} already exist. Is the daemon already running?".format(self.pidfile))
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
self.echo("Starting daemon...")
|
||||
|
||||
# Start the daemon
|
||||
self.daemonize()
|
||||
self._daemonize()
|
||||
self.run()
|
||||
|
||||
def stop(self):
|
||||
def stop(self, check_running=True):
|
||||
"""Stop the daemon."""
|
||||
|
||||
# Get the pid from the pidfile
|
||||
try:
|
||||
with open(self.pidfile,'r') as pf:
|
||||
pid = int(pf.read().strip())
|
||||
except IOError:
|
||||
pid = None
|
||||
|
||||
pid = self.get_pid()
|
||||
if not pid:
|
||||
message = "pidfile {0} does not exist. " + \
|
||||
"Daemon not running?\n"
|
||||
sys.stderr.write(message.format(self.pidfile))
|
||||
return # not an error in a restart
|
||||
if not check_running:
|
||||
return
|
||||
self.error("pidfile {} does not exist. Is the daemon really running?".format(self.pidfile))
|
||||
sys.exit(1)
|
||||
|
||||
# Try killing the daemon process
|
||||
self.echo("Stopping daemon...")
|
||||
|
||||
# Try killing the daemon process
|
||||
try:
|
||||
while 1:
|
||||
os.kill(pid, signal.SIGTERM)
|
||||
|
|
@ -109,20 +110,71 @@ class Daemon:
|
|||
except OSError as err:
|
||||
e = str(err.args)
|
||||
if e.find("No such process") > 0:
|
||||
if os.path.exists(self.pidfile):
|
||||
os.remove(self.pidfile)
|
||||
self.remove_pidfile()
|
||||
else:
|
||||
print (str(err.args))
|
||||
self.error(e)
|
||||
sys.exit(1)
|
||||
|
||||
def restart(self):
|
||||
"""Restart the daemon."""
|
||||
self.stop()
|
||||
self.stop(check_running=False)
|
||||
self.start()
|
||||
|
||||
def status(self):
|
||||
"""Prints the daemon status."""
|
||||
if self.is_running():
|
||||
self.echo("Daemon is running")
|
||||
else:
|
||||
self.echo("Daemon is not running")
|
||||
|
||||
def is_running(self):
|
||||
"""Check if a process is running under the specified pid."""
|
||||
pid = self.get_pid()
|
||||
if pid is None:
|
||||
return False
|
||||
|
||||
try:
|
||||
os.kill(pid, 0)
|
||||
except OSError:
|
||||
try:
|
||||
self.remove_pidfile()
|
||||
except:
|
||||
self.error("Daemon found not running, but could not remove stale pidfile")
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def get_pid(self):
|
||||
"""Get the pid from the pidfile."""
|
||||
try:
|
||||
with open(self.pidfile,'r') as pf:
|
||||
pid = int(pf.read().strip())
|
||||
except (IOError, ValueError):
|
||||
pid = None
|
||||
return pid
|
||||
|
||||
def set_pid(self, pid):
|
||||
"""Write the pid to the pidfile."""
|
||||
with open(self.pidfile,'w+') as f:
|
||||
f.write(str(pid) + '\n')
|
||||
|
||||
def remove_pidfile(self):
|
||||
"""Removes the pidfile."""
|
||||
if os.path.isfile(self.pidfile):
|
||||
os.remove(self.pidfile)
|
||||
|
||||
def run(self):
|
||||
"""You should override this method when you subclass Daemon.
|
||||
|
||||
It will be called after the process has been daemonized by
|
||||
|
||||
It will be called after the process has been daemonized by
|
||||
start() or restart()."""
|
||||
|
||||
raise NotImplementedError()
|
||||
|
||||
@classmethod
|
||||
def echo(cls, line):
|
||||
print(line)
|
||||
|
||||
@classmethod
|
||||
def error(cls, line):
|
||||
print(line, stream=sys.stderr)
|
||||
|
|
|
|||
484
tests/test_daemon.py
Normal file
484
tests/test_daemon.py
Normal file
|
|
@ -0,0 +1,484 @@
|
|||
# coding=utf-8
|
||||
from __future__ import (print_function, absolute_import)
|
||||
|
||||
import unittest
|
||||
import mock
|
||||
|
||||
import octoprint.daemon
|
||||
|
||||
class ExpectedExit(BaseException):
|
||||
pass
|
||||
|
||||
class DaemonTest(unittest.TestCase):
|
||||
def setUp(self):
|
||||
|
||||
run_method = mock.MagicMock()
|
||||
echo_method = mock.MagicMock()
|
||||
error_method = mock.MagicMock()
|
||||
|
||||
class TestDaemon(octoprint.daemon.Daemon):
|
||||
|
||||
def run(self):
|
||||
run_method()
|
||||
|
||||
def echo(cls, line):
|
||||
echo_method(line)
|
||||
|
||||
def error(cls, line):
|
||||
error_method(line)
|
||||
|
||||
self.pidfile = "/my/pid/file"
|
||||
self.daemon = TestDaemon(self.pidfile)
|
||||
self.run_method = run_method
|
||||
self.echo_method = echo_method
|
||||
self.error_method = error_method
|
||||
|
||||
@mock.patch("os.fork", create=True)
|
||||
@mock.patch("os.chdir")
|
||||
@mock.patch("os.setsid", create=True)
|
||||
@mock.patch("os.umask")
|
||||
@mock.patch("sys.exit")
|
||||
def test_double_fork(self, mock_exit, mock_umask, mock_setsid, mock_chdir, mock_fork):
|
||||
# setup
|
||||
pid1 = 1234
|
||||
pid2 = 2345
|
||||
mock_fork.side_effect = [pid1, pid2]
|
||||
|
||||
# test
|
||||
self.daemon._double_fork()
|
||||
|
||||
# assert
|
||||
self.assertListEqual(mock_fork.mock_calls, [mock.call(), mock.call()])
|
||||
self.assertListEqual(mock_exit.mock_calls, [mock.call(0), mock.call(0)])
|
||||
mock_chdir.assert_called_once_with("/")
|
||||
mock_setsid.assert_called_once_with()
|
||||
mock_umask.assert_called_once_with(002)
|
||||
|
||||
@mock.patch("os.fork", create=True)
|
||||
@mock.patch("sys.exit")
|
||||
def test_double_fork_failed_first(self, mock_exit, mock_fork):
|
||||
# setup
|
||||
mock_fork.side_effect = OSError()
|
||||
mock_exit.side_effect = ExpectedExit()
|
||||
|
||||
# test
|
||||
try:
|
||||
self.daemon._double_fork()
|
||||
self.fail("Expected an exit")
|
||||
except ExpectedExit:
|
||||
pass
|
||||
|
||||
# assert
|
||||
self.assertListEqual(mock_fork.mock_calls, [mock.call()])
|
||||
self.assertListEqual(mock_exit.mock_calls, [mock.call(1)])
|
||||
self.assertEquals(len(self.error_method.mock_calls), 1)
|
||||
|
||||
@mock.patch("os.fork", create=True)
|
||||
@mock.patch("os.chdir")
|
||||
@mock.patch("os.setsid", create=True)
|
||||
@mock.patch("os.umask")
|
||||
@mock.patch("sys.exit")
|
||||
def test_double_fork_failed_second(self, mock_exit, mock_umask, mock_setsid, mock_chdir, mock_fork):
|
||||
# setup
|
||||
mock_fork.side_effect = [1234, OSError()]
|
||||
mock_exit.side_effect = [None, ExpectedExit()]
|
||||
|
||||
# test
|
||||
try:
|
||||
self.daemon._double_fork()
|
||||
self.fail("Expected an exit")
|
||||
except ExpectedExit:
|
||||
pass
|
||||
|
||||
# assert
|
||||
self.assertEquals(mock_fork.call_count, 2)
|
||||
self.assertListEqual(mock_exit.mock_calls, [mock.call(0), mock.call(1)])
|
||||
self.assertEquals(self.error_method.call_count, 1)
|
||||
mock_chdir.assert_called_once_with("/")
|
||||
mock_setsid.assert_called_once_with()
|
||||
mock_umask.assert_called_once_with(002)
|
||||
|
||||
@mock.patch("sys.stdin")
|
||||
@mock.patch("sys.stdout")
|
||||
@mock.patch("sys.stderr")
|
||||
@mock.patch("os.devnull")
|
||||
@mock.patch("__builtin__.open")
|
||||
@mock.patch("os.dup2")
|
||||
def test_redirect_io(self, mock_dup2, mock_open, mock_devnull, mock_stderr, mock_stdout, mock_stdin):
|
||||
# setup
|
||||
mock_stdin.fileno.return_value = "stdin"
|
||||
mock_stdout.fileno.return_value = "stdout"
|
||||
mock_stderr.fileno.return_value = "stderr"
|
||||
|
||||
new_stdin = mock.MagicMock()
|
||||
new_stdout = mock.MagicMock()
|
||||
new_stderr = mock.MagicMock()
|
||||
new_stdin.fileno.return_value = "new_stdin"
|
||||
new_stdout.fileno.return_value = "new_stdout"
|
||||
new_stderr.fileno.return_value = "new_stderr"
|
||||
|
||||
mock_open.side_effect = [new_stdin, new_stdout, new_stderr]
|
||||
|
||||
# test
|
||||
self.daemon._redirect_io()
|
||||
|
||||
# assert
|
||||
mock_stdout.flush.assert_called_once_with()
|
||||
mock_stderr.flush.assert_called_once_with()
|
||||
|
||||
self.assertListEqual(mock_open.mock_calls,
|
||||
[mock.call(mock_devnull, "r"),
|
||||
mock.call(mock_devnull, "a+"),
|
||||
mock.call(mock_devnull, "a+")])
|
||||
self.assertListEqual(mock_dup2.mock_calls,
|
||||
[mock.call("new_stdin", "stdin"),
|
||||
mock.call("new_stdout", "stdout"),
|
||||
mock.call("new_stderr", "stderr")])
|
||||
|
||||
@mock.patch("os.getpid")
|
||||
@mock.patch("signal.signal")
|
||||
def test_daemonize(self, mock_signal, mock_getpid):
|
||||
# setup
|
||||
self.daemon._double_fork = mock.MagicMock()
|
||||
self.daemon._redirect_io = mock.MagicMock()
|
||||
self.daemon.set_pid = mock.MagicMock()
|
||||
|
||||
pid = 1234
|
||||
mock_getpid.return_value = pid
|
||||
|
||||
# test
|
||||
self.daemon.start()
|
||||
|
||||
# assert
|
||||
import signal
|
||||
|
||||
self.daemon._double_fork.assert_called_once_with()
|
||||
self.daemon._redirect_io.assert_called_once_with()
|
||||
self.daemon.set_pid.assert_called_once_with(str(pid))
|
||||
mock_signal.assert_called_once_with(signal.SIGTERM, self.daemon._on_sigterm)
|
||||
|
||||
@mock.patch("sys.exit")
|
||||
def test_on_sigterm(self, mock_exit):
|
||||
# setup
|
||||
self.daemon.remove_pidfile = mock.MagicMock()
|
||||
|
||||
mock_exit.side_effect = ExpectedExit
|
||||
|
||||
# test
|
||||
try:
|
||||
self.daemon._on_sigterm("foo", "bar")
|
||||
self.fail("Expected an exit")
|
||||
except ExpectedExit:
|
||||
pass
|
||||
|
||||
# assert
|
||||
self.daemon.remove_pidfile.assert_called_once_with()
|
||||
mock_exit.assert_called_once_with(0)
|
||||
|
||||
def test_start(self):
|
||||
# setup
|
||||
self.daemon._daemonize = mock.MagicMock()
|
||||
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = None
|
||||
|
||||
# test
|
||||
self.daemon.start()
|
||||
|
||||
# assert
|
||||
self.daemon._daemonize.assert_called_once_with()
|
||||
self.daemon.get_pid.assert_called_once_with()
|
||||
self.echo_method.assert_called_once_with("Starting daemon...")
|
||||
self.assertTrue(self.run_method.called)
|
||||
|
||||
@mock.patch("sys.exit")
|
||||
def test_start_running(self, mock_exit):
|
||||
# setup
|
||||
pid = "1234"
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = pid
|
||||
|
||||
mock_exit.side_effect = ExpectedExit()
|
||||
|
||||
# test
|
||||
try:
|
||||
self.daemon.start()
|
||||
self.fail("Expected an exit")
|
||||
except ExpectedExit:
|
||||
pass
|
||||
|
||||
# assert
|
||||
self.daemon.get_pid.assert_called_once_with()
|
||||
self.assertTrue(self.error_method.called)
|
||||
mock_exit.assert_called_once_with(1)
|
||||
|
||||
@mock.patch("os.kill")
|
||||
@mock.patch("time.sleep")
|
||||
def test_stop(self, mock_sleep, mock_kill):
|
||||
import signal
|
||||
|
||||
# setup
|
||||
pid = "1234"
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = pid
|
||||
self.daemon.remove_pidfile = mock.MagicMock()
|
||||
|
||||
mock_kill.side_effect = [None, OSError("No such process")]
|
||||
|
||||
# test
|
||||
self.daemon.stop()
|
||||
|
||||
# assert
|
||||
self.daemon.get_pid.assert_called_once_with()
|
||||
self.assertListEqual(mock_kill.mock_calls,
|
||||
[mock.call(pid, signal.SIGTERM),
|
||||
mock.call(pid, signal.SIGTERM)])
|
||||
mock_sleep.assert_called_once_with(0.1)
|
||||
self.daemon.remove_pidfile.assert_called_once_with()
|
||||
|
||||
@mock.patch("sys.exit")
|
||||
def test_stop_not_running(self, mock_exit):
|
||||
# setup
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = None
|
||||
mock_exit.side_effect = ExpectedExit()
|
||||
|
||||
# test
|
||||
try:
|
||||
self.daemon.stop()
|
||||
self.fail("Expected an exit")
|
||||
except ExpectedExit:
|
||||
pass
|
||||
|
||||
# assert
|
||||
self.daemon.get_pid.assert_called_once_with()
|
||||
self.assertEquals(self.error_method.call_count, 1)
|
||||
mock_exit.assert_called_once_with(1)
|
||||
|
||||
@mock.patch("sys.exit")
|
||||
def test_stop_not_running_no_error(self, mock_exit):
|
||||
# setup
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = None
|
||||
|
||||
# test
|
||||
self.daemon.stop(check_running=False)
|
||||
|
||||
# assert
|
||||
self.daemon.get_pid.assert_called_once_with()
|
||||
self.assertFalse(mock_exit.called)
|
||||
|
||||
@mock.patch("os.kill")
|
||||
@mock.patch("sys.exit")
|
||||
def test_stop_unknown_error(self, mock_exit, mock_kill):
|
||||
# setup
|
||||
pid = "1234"
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = pid
|
||||
|
||||
mock_exit.side_effect = ExpectedExit()
|
||||
mock_kill.side_effect = OSError("Unknown")
|
||||
|
||||
# test
|
||||
try:
|
||||
self.daemon.stop()
|
||||
self.fail("Expected an exit")
|
||||
except ExpectedExit:
|
||||
pass
|
||||
|
||||
# assert
|
||||
self.assertTrue(self.error_method.called)
|
||||
mock_exit.assert_called_once_with(1)
|
||||
|
||||
def test_restart(self):
|
||||
# setup
|
||||
self.daemon.start = mock.MagicMock()
|
||||
self.daemon.stop = mock.MagicMock()
|
||||
|
||||
# test
|
||||
self.daemon.restart()
|
||||
|
||||
# assert
|
||||
self.daemon.stop.assert_called_once_with(check_running=False)
|
||||
self.daemon.start.assert_called_once_with()
|
||||
|
||||
def test_status_running(self):
|
||||
# setup
|
||||
self.daemon.is_running = mock.MagicMock()
|
||||
self.daemon.is_running.return_value = True
|
||||
|
||||
# test
|
||||
self.daemon.status()
|
||||
|
||||
# assert
|
||||
self.echo_method.assert_called_once_with("Daemon is running")
|
||||
|
||||
def test_status_not_running(self):
|
||||
# setup
|
||||
self.daemon.is_running = mock.MagicMock()
|
||||
self.daemon.is_running.return_value = False
|
||||
|
||||
# test
|
||||
self.daemon.status()
|
||||
|
||||
# assert
|
||||
self.echo_method.assert_called_once_with("Daemon is not running")
|
||||
|
||||
@mock.patch("os.kill")
|
||||
def test_is_running_true(self, mock_kill):
|
||||
# setup
|
||||
pid = "1234"
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = pid
|
||||
|
||||
self.daemon.remove_pidfile = mock.MagicMock()
|
||||
|
||||
# test
|
||||
result = self.daemon.is_running()
|
||||
|
||||
# assert
|
||||
self.assertTrue(result)
|
||||
mock_kill.assert_called_once_with(pid, 0)
|
||||
self.assertFalse(self.daemon.remove_pidfile.called)
|
||||
self.assertFalse(self.error_method.called)
|
||||
|
||||
def test_is_running_false_no_pid(self):
|
||||
# setup
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = None
|
||||
|
||||
# test
|
||||
result = self.daemon.is_running()
|
||||
|
||||
# assert
|
||||
self.assertFalse(result)
|
||||
|
||||
@mock.patch("os.kill")
|
||||
def test_is_running_false_pidfile_removed(self, mock_kill):
|
||||
# setup
|
||||
pid = "1234"
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = pid
|
||||
|
||||
mock_kill.side_effect = OSError()
|
||||
|
||||
self.daemon.remove_pidfile = mock.MagicMock()
|
||||
|
||||
# test
|
||||
result = self.daemon.is_running()
|
||||
|
||||
# assert
|
||||
self.assertFalse(result)
|
||||
mock_kill.assert_called_once_with(pid, 0)
|
||||
self.daemon.remove_pidfile.assert_called_once_with()
|
||||
self.assertFalse(self.error_method.called)
|
||||
|
||||
@mock.patch("os.kill")
|
||||
def test_is_running_false_pidfile_error(self, mock_kill):
|
||||
# setup
|
||||
pid = "1234"
|
||||
self.daemon.get_pid = mock.MagicMock()
|
||||
self.daemon.get_pid.return_value = pid
|
||||
|
||||
mock_kill.side_effect = OSError()
|
||||
|
||||
self.daemon.remove_pidfile = mock.MagicMock()
|
||||
self.daemon.remove_pidfile.side_effect = IOError()
|
||||
|
||||
# test
|
||||
result = self.daemon.is_running()
|
||||
|
||||
# assert
|
||||
self.assertFalse(result)
|
||||
mock_kill.assert_called_once_with(pid, 0)
|
||||
self.daemon.remove_pidfile.assert_called_once_with()
|
||||
self.assertTrue(self.error_method.called)
|
||||
|
||||
def test_get_pid(self):
|
||||
# setup
|
||||
pid = 1234
|
||||
|
||||
# test
|
||||
with mock.patch("__builtin__.open", mock.mock_open(read_data="{}\n".format(pid)), create=True) as m:
|
||||
result = self.daemon.get_pid()
|
||||
|
||||
# assert
|
||||
self.assertEquals(result, pid)
|
||||
m.assert_called_once_with(self.pidfile, "r")
|
||||
|
||||
def test_get_pid_ioerror(self):
|
||||
# setup
|
||||
handle = mock.MagicMock()
|
||||
handle.__enter__.side_effect = IOError()
|
||||
|
||||
# test
|
||||
with mock.patch("__builtin__.open", mock.mock_open(), create=True) as m:
|
||||
result = self.daemon.get_pid()
|
||||
|
||||
# assert
|
||||
self.assertIsNone(result)
|
||||
m.assert_called_once_with(self.pidfile, "r")
|
||||
|
||||
def test_get_pid_valueerror(self):
|
||||
# setup
|
||||
pid = "not an integer"
|
||||
|
||||
# test
|
||||
with mock.patch("__builtin__.open", mock.mock_open(read_data="{}\n".format(pid)), create=True) as m:
|
||||
result = self.daemon.get_pid()
|
||||
|
||||
# assert
|
||||
self.assertIsNone(result)
|
||||
m.assert_called_once_with(self.pidfile, "r")
|
||||
|
||||
def test_set_pid(self):
|
||||
# setup
|
||||
pid = "1234"
|
||||
|
||||
# test
|
||||
with mock.patch("__builtin__.open", mock.mock_open(), create=True) as m:
|
||||
self.daemon.set_pid(pid)
|
||||
|
||||
# assert
|
||||
m.assert_called_once_with(self.pidfile, "w+")
|
||||
handle = m()
|
||||
handle.write.assert_called_once_with("{}\n".format(pid))
|
||||
|
||||
def test_set_pid_int(self):
|
||||
# setup
|
||||
pid = 1234
|
||||
|
||||
# test
|
||||
with mock.patch("__builtin__.open", mock.mock_open(), create=True) as m:
|
||||
self.daemon.set_pid(pid)
|
||||
|
||||
# assert
|
||||
m.assert_called_once_with(self.pidfile, "w+")
|
||||
handle = m()
|
||||
handle.write.assert_called_once_with("{}\n".format(pid))
|
||||
|
||||
@mock.patch("os.path.isfile")
|
||||
@mock.patch("os.remove")
|
||||
def test_remove_pidfile_exists(self, mock_remove, mock_isfile):
|
||||
# setup
|
||||
mock_isfile.return_value = True
|
||||
|
||||
# test
|
||||
self.daemon.remove_pidfile()
|
||||
|
||||
# assert
|
||||
mock_isfile.assert_called_once_with(self.pidfile)
|
||||
mock_remove.assert_called_once_with(self.pidfile)
|
||||
|
||||
@mock.patch("os.path.isfile")
|
||||
@mock.patch("os.remove")
|
||||
def test_remove_pidfile_doesnt_exist(self, mock_remove, mock_isfile):
|
||||
# setup
|
||||
mock_isfile.return_value = False
|
||||
|
||||
# test
|
||||
self.daemon.remove_pidfile()
|
||||
|
||||
# assert
|
||||
mock_isfile.assert_called_once_with(self.pidfile)
|
||||
self.assertFalse(mock_remove.called)
|
||||
Loading…
Reference in a new issue