import os import errno import stat import tempfile from unittest import TestCase from mock import patch, mock_open, call, DEFAULT from rattail import daemon class DaemonTests(TestCase): pidfile = '/tmp/test_rattail_daemon.pid' def test_init(self): d = daemon.Daemon(self.pidfile) self.assertEqual(d.pidfile, self.pidfile) self.assertEqual(d.stdin, os.devnull) self.assertEqual(d.stdout, os.devnull) self.assertEqual(d.stderr, os.devnull) def test_delpid(self): # normal delete with open(self.pidfile, 'w') as f: f.write('foo') self.assertTrue(os.path.exists(self.pidfile)) d = daemon.Daemon(self.pidfile) d.delpid() self.assertFalse(os.path.exists(self.pidfile)) # file doesn't exist (no error) d.delpid() @patch('sys.exit') @patch('sys.stderr') def test_start(self, stderr, exit_): d = daemon.Daemon(self.pidfile) with patch.object(d, 'daemonize') as daemonize: with patch.object(d, 'run') as run: # pidfile exists with open(self.pidfile, 'w') as f: f.write('42\n') self.assertTrue(os.path.exists(self.pidfile)) exit_.side_effect = RuntimeError self.assertRaises(RuntimeError, d.start) stderr.write.assert_called_once_with("pidfile /tmp/test_rattail_daemon.pid already exist. Daemon already running?\n") exit_.assert_called_once_with(1) self.assertFalse(daemonize.called) self.assertFalse(run.called) stderr.reset_mock() exit_.reset_mock() # no pidfile, no daemonize os.remove(self.pidfile) self.assertFalse(os.path.exists(self.pidfile)) d.start(daemonize=False) self.assertFalse(stderr.write.called) self.assertFalse(exit_.called) self.assertFalse(d.daemonize.called) d.run.assert_called_once_with() d.run.reset_mock() # no pidfile, with daemonize self.assertFalse(os.path.exists(self.pidfile)) d.start(daemonize=True) self.assertFalse(stderr.write.called) self.assertFalse(exit_.called) d.daemonize.assert_called_once_with() d.run.assert_called_once_with() @patch('os.chdir') @patch('os.chmod') @patch('os.dup2') @patch('os.fork') @patch('os.getpid') @patch('os.setsid') @patch('os.umask') @patch('sys.exit') @patch('sys.stderr') @patch('sys.stdin') @patch('sys.stdout') @patch('atexit.register') class DaemonDaemonizeTests(TestCase): pidfile = '/tmp/test_rattail_daemon.pid' def tearDown(self): if os.path.exists(self.pidfile): os.remove(self.pidfile) def fake_open(self, *args): # Return mock values for stdin etc. if args[0] in ('/tmp/stdin', '/tmp/stdout', '/tmp/stderr'): return DEFAULT # Real value is returned for PID file. return open(*args) def test_first_fork_failure(self, register, stdout, stdin, stderr, exit_, umask, setsid, getpid, fork, dup2, chmod, chdir): fork.side_effect = OSError(errno.EIO, "I/O error") exit_.side_effect = RuntimeError d = daemon.Daemon(self.pidfile) self.assertRaises(RuntimeError, d.daemonize) stderr.write.assert_called_once_with("fork #1 failed: 5 (I/O error)\n") exit_.assert_called_once_with(1) def test_first_fork_success(self, register, stdout, stdin, stderr, exit_, umask, setsid, getpid, fork, dup2, chmod, chdir): fork.return_value = 42 exit_.side_effect = RuntimeError d = daemon.Daemon(self.pidfile) self.assertRaises(RuntimeError, d.daemonize) self.assertFalse(stderr.write.called) exit_.assert_called_once_with(0) def test_second_fork_failure(self, register, stdout, stdin, stderr, exit_, umask, setsid, getpid, fork, dup2, chmod, chdir): fork.side_effect = [0, OSError(errno.EIO, "I/O error")] exit_.side_effect = RuntimeError d = daemon.Daemon(self.pidfile) self.assertRaises(RuntimeError, d.daemonize) stderr.write.assert_called_once_with("fork #2 failed: 5 (I/O error)\n") exit_.assert_called_once_with(1) def test_second_fork_success_parent(self, register, stdout, stdin, stderr, exit_, umask, setsid, getpid, fork, dup2, chmod, chdir): fork.side_effect = [0, 42] exit_.side_effect = RuntimeError d = daemon.Daemon(self.pidfile) self.assertRaises(RuntimeError, d.daemonize) self.assertFalse(stderr.write.called) exit_.assert_called_once_with(0) def test_second_fork_success_child(self, register, stdout, stdin, stderr, exit_, umask, setsid, getpid, fork, dup2, chmod, chdir): with patch('rattail.daemon.open', new=mock_open(), create=True) as open_: fork.side_effect = [0, 0] open_.side_effect = self.fake_open getpid.return_value = 42 d = daemon.Daemon(self.pidfile, stdin='/tmp/stdin', stdout='/tmp/stdout', stderr='/tmp/stderr') d.daemonize() stdout.flush.assert_called_once_with() stderr.flush.assert_called_once_with() self.assertEqual(open_.call_args_list, [ call('/tmp/stdin', 'r'), call('/tmp/stdout', 'a+'), call('/tmp/stderr', 'a+', 0), call(self.pidfile, 'w+'), ]) stdin.fileno.assert_called_once_with() stdout.fileno.assert_called_once_with() stderr.fileno.assert_called_once_with() self.assertEqual(open_.return_value.fileno.call_count, 3) self.assertEqual(dup2.call_args_list, [ call(open_.return_value.fileno.return_value, stdin.fileno.return_value), call(open_.return_value.fileno.return_value, stdout.fileno.return_value), call(open_.return_value.fileno.return_value, stderr.fileno.return_value), ]) register.assert_called_once_with(d.delpid) getpid.assert_called_once_with() chmod.assert_called_once_with('/tmp/test_rattail_daemon.pid', stat.S_IRUSR | stat.S_IWUSR) with open(self.pidfile) as f: pid = f.read() self.assertEqual(pid, '42\n') def test_pidfile_error(self, register, stdout, stdin, stderr, exit_, umask, setsid, getpid, fork, dup2, chmod, chdir): tempdir = tempfile.mkdtemp() os.rmdir(tempdir) self.pidfile = os.path.join(tempdir, 'test_rattail_daemon.pid') self.assertFalse(os.path.exists(tempdir)) self.assertFalse(os.path.exists(self.pidfile)) with patch('rattail.daemon.open', new=mock_open(), create=True) as open_: fork.side_effect = [0, 0] open_.side_effect = self.fake_open d = daemon.Daemon(self.pidfile, stdin='/tmp/stdin', stdout='/tmp/stdout', stderr='/tmp/stderr') self.assertRaises(IOError, d.daemonize) open_.assert_called_with(self.pidfile, 'w+') self.assertFalse(chmod.called)