126 lines
3.2 KiB
Python
126 lines
3.2 KiB
Python
#!/usr/bin/env python
|
|
|
|
import os
|
|
import signal
|
|
|
|
import mozprocess
|
|
import mozunit
|
|
import proctest
|
|
|
|
here = os.path.dirname(os.path.abspath(__file__))
|
|
|
|
|
|
def kill(proc):
|
|
is_win = os.name == "nt"
|
|
if is_win:
|
|
proc.send_signal(signal.CTRL_BREAK_EVENT)
|
|
else:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
proc.wait()
|
|
|
|
|
|
class ProcTestSimpleRunAndWait(proctest.ProcTest):
|
|
"""Class to test mozprocess.run_and_wait"""
|
|
|
|
def test_normal_finish(self):
|
|
"""Process is started, runs to completion while we wait for it"""
|
|
|
|
p = mozprocess.run_and_wait(
|
|
[self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here
|
|
)
|
|
self.assertEqual(p.returncode, 0)
|
|
|
|
def test_outputhandler(self):
|
|
"""Output handler receives output generated by process"""
|
|
found = False
|
|
|
|
def olh(p, line):
|
|
nonlocal found
|
|
self.assertEqual(line, "XYZ\n")
|
|
found = True
|
|
|
|
p = mozprocess.run_and_wait(
|
|
[self.python, "-c", "print('XYZ')"], cwd=here, output_line_handler=olh
|
|
)
|
|
self.assertTrue(found)
|
|
self.assertEqual(p.returncode, 0)
|
|
|
|
def test_wait(self):
|
|
"""Process is started runs to completion while we wait indefinitely"""
|
|
|
|
p = mozprocess.run_and_wait(
|
|
[self.python, self.proclaunch, "process_waittimeout_10s.ini"], cwd=here
|
|
)
|
|
self.assertEqual(p.returncode, 0)
|
|
|
|
def test_timeout(self):
|
|
"""Process is started, runs but we time out waiting on it
|
|
to complete
|
|
"""
|
|
timed_out = False
|
|
|
|
def th(p):
|
|
nonlocal timed_out
|
|
timed_out = True
|
|
kill(p)
|
|
|
|
mozprocess.run_and_wait(
|
|
[self.python, self.proclaunch, "process_waittimeout.ini"],
|
|
cwd=here,
|
|
timeout=10,
|
|
timeout_handler=th,
|
|
)
|
|
self.assertTrue(timed_out)
|
|
|
|
def test_waitnotimeout(self):
|
|
"""Process is started, runs to completion before our wait times out"""
|
|
p = mozprocess.run_and_wait(
|
|
[self.python, self.proclaunch, "process_waittimeout_10s.ini"],
|
|
cwd=here,
|
|
timeout=30,
|
|
)
|
|
self.assertEqual(p.returncode, 0)
|
|
|
|
def test_outputtimeout(self):
|
|
"""Process produces output, but output stalls and exceeds output timeout"""
|
|
|
|
pgm = """
|
|
import time
|
|
|
|
for i in range(10):
|
|
print(i)
|
|
time.sleep(1)
|
|
time.sleep(10)
|
|
print("survived sleep!")
|
|
"""
|
|
found = False
|
|
found9 = False
|
|
timed_out = False
|
|
|
|
def olh(p, line):
|
|
nonlocal found
|
|
nonlocal found9
|
|
if "9" in line:
|
|
found9 = True
|
|
if "survived" in line:
|
|
found = True
|
|
|
|
def oth(p):
|
|
nonlocal timed_out
|
|
timed_out = True
|
|
kill(p)
|
|
|
|
mozprocess.run_and_wait(
|
|
[self.python, "-u", "-c", pgm],
|
|
cwd=here,
|
|
output_timeout=5,
|
|
output_timeout_handler=oth,
|
|
output_line_handler=olh,
|
|
)
|
|
self.assertFalse(found)
|
|
self.assertTrue(found9)
|
|
self.assertTrue(timed_out)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mozunit.main()
|