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 205 insertions and 6 deletions:
rattail/daemon.py
8
6
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,10 +63,12 @@ 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):

			




	
	
	
		
 
        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)
    





    


  

  







  


    




    








    
        Server instance: tuna-2793187
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2021 by various authors & licensed under GPLv3.
            – Support