test_run_and_wait.py (3240B)
1 #!/usr/bin/env python 2 3 import os 4 import signal 5 6 import mozprocess 7 import mozunit 8 import proctest 9 10 here = os.path.dirname(os.path.abspath(__file__)) 11 12 13 def kill(proc): 14 is_win = os.name == "nt" 15 if is_win: 16 proc.send_signal(signal.CTRL_BREAK_EVENT) 17 else: 18 os.killpg(proc.pid, signal.SIGKILL) 19 proc.wait() 20 21 22 class ProcTestSimpleRunAndWait(proctest.ProcTest): 23 """Class to test mozprocess.run_and_wait""" 24 25 def test_normal_finish(self): 26 """Process is started, runs to completion while we wait for it""" 27 28 p = mozprocess.run_and_wait( 29 [self.python, self.proclaunch, "process_normal_finish.ini"], cwd=here 30 ) 31 self.assertEqual(p.returncode, 0) 32 33 def test_outputhandler(self): 34 """Output handler receives output generated by process""" 35 found = False 36 37 def olh(p, line): 38 nonlocal found 39 self.assertEqual(line, "XYZ\n") 40 found = True 41 42 p = mozprocess.run_and_wait( 43 [self.python, "-c", "print('XYZ')"], cwd=here, output_line_handler=olh 44 ) 45 self.assertTrue(found) 46 self.assertEqual(p.returncode, 0) 47 48 def test_wait(self): 49 """Process is started runs to completion while we wait indefinitely""" 50 51 p = mozprocess.run_and_wait( 52 [self.python, self.proclaunch, "process_waittimeout_10s.ini"], cwd=here 53 ) 54 self.assertEqual(p.returncode, 0) 55 56 def test_timeout(self): 57 """Process is started, runs but we time out waiting on it 58 to complete 59 """ 60 timed_out = False 61 62 def th(p): 63 nonlocal timed_out 64 timed_out = True 65 kill(p) 66 67 mozprocess.run_and_wait( 68 [self.python, self.proclaunch, "process_waittimeout.ini"], 69 cwd=here, 70 timeout=10, 71 timeout_handler=th, 72 ) 73 self.assertTrue(timed_out) 74 75 def test_waitnotimeout(self): 76 """Process is started, runs to completion before our wait times out""" 77 p = mozprocess.run_and_wait( 78 [self.python, self.proclaunch, "process_waittimeout_10s.ini"], 79 cwd=here, 80 timeout=30, 81 ) 82 self.assertEqual(p.returncode, 0) 83 84 def test_outputtimeout(self): 85 """Process produces output, but output stalls and exceeds output timeout""" 86 87 pgm = """ 88 import time 89 90 for i in range(10): 91 print(i) 92 time.sleep(1) 93 time.sleep(10) 94 print("survived sleep!") 95 """ 96 found = False 97 found9 = False 98 timed_out = False 99 100 def olh(p, line): 101 nonlocal found 102 nonlocal found9 103 if "9" in line: 104 found9 = True 105 if "survived" in line: 106 found = True 107 108 def oth(p): 109 nonlocal timed_out 110 timed_out = True 111 kill(p) 112 113 mozprocess.run_and_wait( 114 [self.python, "-u", "-c", pgm], 115 cwd=here, 116 output_timeout=5, 117 output_timeout_handler=oth, 118 output_line_handler=olh, 119 ) 120 self.assertFalse(found) 121 self.assertTrue(found9) 122 self.assertTrue(timed_out) 123 124 125 if __name__ == "__main__": 126 mozunit.main()