1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150
|
#!/usr/bin/env python
import os
import signal
import sys
import time
import unittest
import mozinfo
import mozunit
import proctest
from mozprocess import processhandler
here = os.path.dirname(os.path.abspath(__file__))
class ProcTestPoll(proctest.ProcTest):
"""Class to test process poll."""
def test_poll_before_run(self):
"""Process is not started, and poll() is called."""
p = processhandler.ProcessHandler(
[self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here
)
self.assertRaises(RuntimeError, p.poll)
def test_poll_while_running(self):
"""Process is started, and poll() is called."""
p = processhandler.ProcessHandler(
[self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here
)
p.run()
returncode = p.poll()
self.assertEqual(returncode, None)
self.determine_status(p, True)
p.kill()
def test_poll_after_kill(self):
"""Process is killed, and poll() is called."""
p = processhandler.ProcessHandler(
[self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here
)
p.run()
returncode = p.kill()
# We killed the process, so the returncode should be non-zero
if mozinfo.isWin:
self.assertGreater(
returncode, 0, 'Positive returncode expected, got "%s"' % returncode
)
else:
self.assertLess(
returncode, 0, 'Negative returncode expected, got "%s"' % returncode
)
self.assertEqual(returncode, p.poll())
self.determine_status(p)
def test_poll_after_kill_no_process_group(self):
"""Process (no group) is killed, and poll() is called."""
p = processhandler.ProcessHandler(
[
self.python,
self.proclaunch,
"process_normal_finish_no_process_group.ini",
],
cwd=here,
ignore_children=True,
)
p.run()
returncode = p.kill()
# We killed the process, so the returncode should be non-zero
if mozinfo.isWin:
self.assertGreater(
returncode, 0, 'Positive returncode expected, got "%s"' % returncode
)
else:
self.assertLess(
returncode, 0, 'Negative returncode expected, got "%s"' % returncode
)
self.assertEqual(returncode, p.poll())
self.determine_status(p)
def test_poll_after_double_kill(self):
"""Process is killed twice, and poll() is called."""
p = processhandler.ProcessHandler(
[self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here
)
p.run()
p.kill()
returncode = p.kill()
# We killed the process, so the returncode should be non-zero
if mozinfo.isWin:
self.assertGreater(
returncode, 0, 'Positive returncode expected, got "%s"' % returncode
)
else:
self.assertLess(
returncode, 0, 'Negative returncode expected, got "%s"' % returncode
)
self.assertEqual(returncode, p.poll())
self.determine_status(p)
@unittest.skipIf(sys.platform.startswith("win"), "Bug 1493796")
def test_poll_after_external_kill(self):
"""Process is killed externally, and poll() is called."""
p = processhandler.ProcessHandler(
[self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here
)
p.run()
os.kill(p.pid, signal.SIGTERM)
# Allow the output reader thread to finish processing remaining data
for i in range(0, 100):
time.sleep(processhandler.INTERVAL_PROCESS_ALIVE_CHECK)
returncode = p.poll()
if returncode is not None:
break
# We killed the process, so the returncode should be non-zero
if mozinfo.isWin:
self.assertEqual(
returncode,
signal.SIGTERM,
'Positive returncode expected, got "%s"' % returncode,
)
else:
self.assertEqual(
returncode,
-signal.SIGTERM,
'%s expected, got "%s"' % (-signal.SIGTERM, returncode),
)
self.assertEqual(returncode, p.wait())
self.determine_status(p)
if __name__ == "__main__":
mozunit.main()
|