michael@0: #!/usr/bin/env python
michael@0: 
michael@0: import os
michael@0: import signal
michael@0: import unittest
michael@0: 
michael@0: import mozinfo
michael@0: from mozprocess import processhandler
michael@0: 
michael@0: import proctest
michael@0: 
michael@0: 
michael@0: here = os.path.dirname(os.path.abspath(__file__))
michael@0: 
michael@0: 
michael@0: class ProcTestPoll(proctest.ProcTest):
michael@0:     """ Class to test process poll """
michael@0: 
michael@0:     def test_poll_before_run(self):
michael@0:         """Process is not started, and poll() is called"""
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, self.proclaunch,
michael@0:                                           "process_normal_finish_python.ini"],
michael@0:                                           cwd=here)
michael@0:         self.assertRaises(AttributeError, p.poll)
michael@0: 
michael@0:     def test_poll_while_running(self):
michael@0:         """Process is started, and poll() is called"""
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, self.proclaunch,
michael@0:                                           "process_normal_finish_python.ini"],
michael@0:                                           cwd=here)
michael@0:         p.run()
michael@0:         returncode = p.poll()
michael@0: 
michael@0:         self.assertEqual(returncode, None)
michael@0: 
michael@0:         detected, output = proctest.check_for_process(self.proclaunch)
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               returncode,
michael@0:                               p.didTimeout,
michael@0:                               True)
michael@0:         p.kill()
michael@0: 
michael@0:     def test_poll_after_kill(self):
michael@0:         """Process is killed, and poll() is called"""
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, self.proclaunch,
michael@0:                                           "process_normal_finish_python.ini"],
michael@0:                                           cwd=here)
michael@0:         p.run()
michael@0:         returncode = p.kill()
michael@0: 
michael@0:         # We killed the process, so the returncode should be < 0
michael@0:         self.assertLess(returncode, 0)
michael@0:         self.assertEqual(returncode, p.poll())
michael@0: 
michael@0:         detected, output = proctest.check_for_process(self.proclaunch)
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               returncode,
michael@0:                               p.didTimeout)
michael@0: 
michael@0:     def test_poll_after_kill_no_process_group(self):
michael@0:         """Process (no group) is killed, and poll() is called"""
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, self.proclaunch,
michael@0:                                           "process_normal_finish_no_process_group.ini"],
michael@0:                                           cwd=here,
michael@0:                                           ignore_children=True
michael@0:                                           )
michael@0:         p.run()
michael@0:         returncode = p.kill()
michael@0: 
michael@0:         # We killed the process, so the returncode should be < 0
michael@0:         self.assertLess(returncode, 0)
michael@0:         self.assertEqual(returncode, p.poll())
michael@0: 
michael@0:         detected, output = proctest.check_for_process(self.proclaunch)
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               returncode,
michael@0:                               p.didTimeout)
michael@0: 
michael@0:     def test_poll_after_double_kill(self):
michael@0:         """Process is killed twice, and poll() is called"""
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, self.proclaunch,
michael@0:                                           "process_normal_finish_python.ini"],
michael@0:                                           cwd=here)
michael@0:         p.run()
michael@0:         p.kill()
michael@0:         returncode = p.kill()
michael@0: 
michael@0:         # We killed the process, so the returncode should be < 0
michael@0:         self.assertLess(returncode, 0)
michael@0:         self.assertEqual(returncode, p.poll())
michael@0: 
michael@0:         detected, output = proctest.check_for_process(self.proclaunch)
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               returncode,
michael@0:                               p.didTimeout)
michael@0: 
michael@0:     def test_poll_after_external_kill(self):
michael@0:         """Process is killed externally, and poll() is called"""
michael@0: 
michael@0:         p = processhandler.ProcessHandler([self.python, self.proclaunch,
michael@0:                                           "process_normal_finish_python.ini"],
michael@0:                                           cwd=here)
michael@0:         p.run()
michael@0:         os.kill(p.pid, signal.SIGTERM)
michael@0:         returncode = p.wait()
michael@0: 
michael@0:         # We killed the process, so the returncode should be < 0
michael@0:         self.assertEqual(returncode, -signal.SIGTERM)
michael@0:         self.assertEqual(returncode, p.poll())
michael@0: 
michael@0:         detected, output = proctest.check_for_process(self.proclaunch)
michael@0:         self.determine_status(detected,
michael@0:                               output,
michael@0:                               returncode,
michael@0:                               p.didTimeout)
michael@0: 
michael@0: if __name__ == '__main__':
michael@0:     unittest.main()