rattail-project/rattail Changeset

Changeset - 9a8d5a1d2664

Parent rev.

Child rev.

[Not reviewed]

0 1 1

Lance Edgar (lance) - 11 years ago 2013-12-21 20:13:27
lance@edbob.org

Added some daemon tests.

2 files changed with 206 insertions and 7 deletions:

rattail/daemon.py

tests/test_daemon.py

197

0 comments (0 inline, 0 general)

rattail/daemon.py

➞

Show inline comments

@@ @@ -53,9 +53,9 @@ class Daemon: @@
         # redirect standard file descriptors
         sys.stdout.flush()
         sys.stderr.flush()
         si = file(self.stdin, 'r')
         so = file(self.stdout, 'a+')
         se = file(self.stderr, 'a+', 0)
         si = open(self.stdin, 'r')
         so = open(self.stdout, 'a+')
         se = open(self.stderr, 'a+', 0)
         os.dup2(si.fileno(), sys.stdin.fileno())
         os.dup2(so.fileno(), sys.stdout.fileno())
         os.dup2(se.fileno(), sys.stderr.fileno())
@@ @@ -63,11 +63,13 @@ class Daemon: @@
         # write pidfile
         atexit.register(self.delpid)
         pid = str(os.getpid())
         file(self.pidfile,'w+').write("%s\n" % pid)
         with open(self.pidfile, 'w+') as f:
             f.write('{0}\n'.format(pid))
         os.chmod(self.pidfile, stat.S_IRUSR|stat.S_IWUSR)
     def delpid(self):
         os.remove(self.pidfile)
         if os.path.exists(self.pidfile):
             os.remove(self.pidfile)
     def start(self, daemonize=True):
         """
@@ @@ -75,7 +77,7 @@ class Daemon: @@
         """
         # Check for a pidfile to see if the daemon already runs
         try:
-            pf = file(self.pidfile,'r')
+            pf = open(self.pidfile,'r')
             pid = int(pf.read().strip())
             pf.close()
         except IOError:
@@ @@ -97,7 +99,7 @@ class Daemon: @@
         """
         # Get the pid from the pidfile
         try:
-            pf = file(self.pidfile,'r')
+            pf = open(self.pidfile,'r')
             pid = int(pf.read().strip())
             pf.close()
         except IOError:

tests/test_daemon.py

➞

Show inline comments

@@ new file 100644 @@
 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)

0 comments (0 inline, 0 general)