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
|
#!/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()
|
