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
151
152
153
154
155
156
157
158
159
160
|
#!/usr/bin/env python
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
# You can obtain one at http://mozilla.org/MPL/2.0/.
import os
import subprocess
import sys
import unittest
from time import sleep
from mozprocess import processhandler
here = os.path.dirname(os.path.abspath(__file__))
def make_proclaunch(aDir):
"""
Makes the proclaunch executable.
Params:
aDir - the directory in which to issue the make commands
Returns:
the path to the proclaunch executable that is generated
"""
# Ideally make should take care of this, but since it doesn't - on windows,
# anyway, let's just call out both targets explicitly.
p = subprocess.call(["make", "-C", "iniparser"], cwd=aDir)
p = subprocess.call(["make"], cwd=aDir)
if sys.platform == "win32":
exepath = os.path.join(aDir, "proclaunch.exe")
else:
exepath = os.path.join(aDir, "proclaunch")
return exepath
def check_for_process(processName):
"""
Use to determine if process of the given name is still running.
Returns:
detected -- True if process is detected to exist, False otherwise
output -- if process exists, stdout of the process, '' otherwise
"""
output = ''
if sys.platform == "win32":
# On windows we use tasklist
p1 = subprocess.Popen(["tasklist"], stdout=subprocess.PIPE)
output = p1.communicate()[0]
detected = False
for line in output.splitlines():
if processName in line:
detected = True
break
else:
p1 = subprocess.Popen(["ps", "-ef"], stdout=subprocess.PIPE)
p2 = subprocess.Popen(["grep", processName], stdin=p1.stdout, stdout=subprocess.PIPE)
p1.stdout.close()
output = p2.communicate()[0]
detected = False
for line in output.splitlines():
if "grep %s" % processName in line:
continue
elif processName in line and not 'defunct' in line:
detected = True
break
return detected, output
class ProcTest1(unittest.TestCase):
def __init__(self, *args, **kwargs):
# Ideally, I'd use setUpClass but that only exists in 2.7.
# So, we'll do this make step now.
self.proclaunch = make_proclaunch(here)
unittest.TestCase.__init__(self, *args, **kwargs)
def test_process_normal_finish(self):
"""Process is started, runs to completion while we wait for it"""
p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"],
cwd=here)
p.run()
p.processOutput()
p.waitForFinish()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
output,
p.proc.returncode,
p.didTimeout)
def test_process_waittimeout(self):
""" Process is started, runs but we time out waiting on it
to complete
"""
p = processhandler.ProcessHandler([self.proclaunch, "process_waittimeout.ini"],
cwd=here)
p.run()
p.processOutput(timeout=10)
p.waitForFinish()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
output,
p.proc.returncode,
p.didTimeout,
False,
['returncode', 'didtimeout'])
def test_process_kill(self):
""" Process is started, we kill it
"""
p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"],
cwd=here)
p.run()
p.processOutput()
p.kill()
detected, output = check_for_process(self.proclaunch)
self.determine_status(detected,
output,
p.proc.returncode,
p.didTimeout)
def determine_status(self,
detected=False,
output='',
returncode=0,
didtimeout=False,
isalive=False,
expectedfail=[]):
"""
Use to determine if the situation has failed.
Parameters:
detected -- value from check_for_process to determine if the process is detected
output -- string of data from detected process, can be ''
returncode -- return code from process, defaults to 0
didtimeout -- True if process timed out, defaults to False
isalive -- Use True to indicate we pass if the process exists; however, by default
the test will pass if the process does not exist (isalive == False)
expectedfail -- Defaults to [], used to indicate a list of fields that are expected to fail
"""
if 'returncode' in expectedfail:
self.assertTrue(returncode, "Detected an unexpected return code of: %s" % returncode)
elif not isalive:
self.assertTrue(returncode == 0, "Detected non-zero return code of: %d" % returncode)
if 'didtimeout' in expectedfail:
self.assertTrue(didtimeout, "Detected that process didn't time out")
else:
self.assertTrue(not didtimeout, "Detected that process timed out")
if isalive:
self.assertTrue(detected, "Detected process is not running, process output: %s" % output)
else:
self.assertTrue(not detected, "Detected process is still running, process output: %s" % output)
if __name__ == '__main__':
unittest.main()
|