|
0.11.6
by Chris Coulson
Import upstream version 1.4.6 |
1 |
#!/usr/bin/env python
|
2 |
||
3 |
# This Source Code Form is subject to the terms of the Mozilla Public
|
|
4 |
# License, v. 2.0. If a copy of the MPL was not distributed with this file,
|
|
5 |
# You can obtain one at http://mozilla.org/MPL/2.0/.
|
|
6 |
||
7 |
import os |
|
8 |
import subprocess |
|
9 |
import sys |
|
10 |
import unittest |
|
11 |
from time import sleep |
|
12 |
||
13 |
from mozprocess import processhandler |
|
14 |
||
15 |
here = os.path.dirname(os.path.abspath(__file__)) |
|
16 |
||
17 |
def make_proclaunch(aDir): |
|
18 |
"""
|
|
19 |
Makes the proclaunch executable.
|
|
20 |
Params:
|
|
21 |
aDir - the directory in which to issue the make commands
|
|
22 |
Returns:
|
|
23 |
the path to the proclaunch executable that is generated
|
|
24 |
"""
|
|
25 |
# Ideally make should take care of this, but since it doesn't - on windows,
|
|
26 |
# anyway, let's just call out both targets explicitly.
|
|
27 |
p = subprocess.call(["make", "-C", "iniparser"], cwd=aDir) |
|
28 |
p = subprocess.call(["make"], cwd=aDir) |
|
29 |
if sys.platform == "win32": |
|
30 |
exepath = os.path.join(aDir, "proclaunch.exe") |
|
31 |
else: |
|
32 |
exepath = os.path.join(aDir, "proclaunch") |
|
33 |
return exepath |
|
34 |
||
35 |
def check_for_process(processName): |
|
36 |
"""
|
|
37 |
Use to determine if process of the given name is still running.
|
|
38 |
||
39 |
Returns:
|
|
40 |
detected -- True if process is detected to exist, False otherwise
|
|
41 |
output -- if process exists, stdout of the process, '' otherwise
|
|
42 |
"""
|
|
43 |
output = '' |
|
44 |
if sys.platform == "win32": |
|
45 |
# On windows we use tasklist
|
|
46 |
p1 = subprocess.Popen(["tasklist"], stdout=subprocess.PIPE) |
|
47 |
output = p1.communicate()[0] |
|
48 |
detected = False |
|
49 |
for line in output.splitlines(): |
|
50 |
if processName in line: |
|
51 |
detected = True |
|
52 |
break
|
|
53 |
else: |
|
54 |
p1 = subprocess.Popen(["ps", "-ef"], stdout=subprocess.PIPE) |
|
55 |
p2 = subprocess.Popen(["grep", processName], stdin=p1.stdout, stdout=subprocess.PIPE) |
|
56 |
p1.stdout.close() |
|
57 |
output = p2.communicate()[0] |
|
58 |
detected = False |
|
59 |
for line in output.splitlines(): |
|
60 |
if "grep %s" % processName in line: |
|
61 |
continue
|
|
62 |
elif processName in line and not 'defunct' in line: |
|
63 |
detected = True |
|
64 |
break
|
|
65 |
||
66 |
return detected, output |
|
67 |
||
68 |
||
69 |
class ProcTest1(unittest.TestCase): |
|
70 |
||
71 |
def __init__(self, *args, **kwargs): |
|
72 |
||
73 |
# Ideally, I'd use setUpClass but that only exists in 2.7.
|
|
74 |
# So, we'll do this make step now.
|
|
75 |
self.proclaunch = make_proclaunch(here) |
|
76 |
unittest.TestCase.__init__(self, *args, **kwargs) |
|
77 |
||
78 |
def test_process_normal_finish(self): |
|
79 |
"""Process is started, runs to completion while we wait for it"""
|
|
80 |
||
81 |
p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
|
82 |
cwd=here) |
|
83 |
p.run() |
|
84 |
p.processOutput() |
|
85 |
p.waitForFinish() |
|
86 |
||
87 |
detected, output = check_for_process(self.proclaunch) |
|
88 |
self.determine_status(detected, |
|
89 |
output, |
|
90 |
p.proc.returncode, |
|
91 |
p.didTimeout) |
|
92 |
||
93 |
def test_process_waittimeout(self): |
|
94 |
""" Process is started, runs but we time out waiting on it
|
|
95 |
to complete
|
|
96 |
"""
|
|
97 |
p = processhandler.ProcessHandler([self.proclaunch, "process_waittimeout.ini"], |
|
98 |
cwd=here) |
|
99 |
p.run() |
|
100 |
p.processOutput(timeout=10) |
|
101 |
p.waitForFinish() |
|
102 |
||
103 |
detected, output = check_for_process(self.proclaunch) |
|
104 |
self.determine_status(detected, |
|
105 |
output, |
|
106 |
p.proc.returncode, |
|
107 |
p.didTimeout, |
|
108 |
False, |
|
109 |
['returncode', 'didtimeout']) |
|
110 |
||
111 |
def test_process_kill(self): |
|
112 |
""" Process is started, we kill it
|
|
113 |
"""
|
|
114 |
p = processhandler.ProcessHandler([self.proclaunch, "process_normal_finish.ini"], |
|
115 |
cwd=here) |
|
116 |
p.run() |
|
117 |
p.processOutput() |
|
118 |
p.kill() |
|
119 |
||
120 |
detected, output = check_for_process(self.proclaunch) |
|
121 |
self.determine_status(detected, |
|
122 |
output, |
|
123 |
p.proc.returncode, |
|
124 |
p.didTimeout) |
|
125 |
||
126 |
def determine_status(self, |
|
127 |
detected=False, |
|
128 |
output='', |
|
129 |
returncode=0, |
|
130 |
didtimeout=False, |
|
131 |
isalive=False, |
|
132 |
expectedfail=[]): |
|
133 |
"""
|
|
134 |
Use to determine if the situation has failed.
|
|
135 |
Parameters:
|
|
136 |
detected -- value from check_for_process to determine if the process is detected
|
|
137 |
output -- string of data from detected process, can be ''
|
|
138 |
returncode -- return code from process, defaults to 0
|
|
139 |
didtimeout -- True if process timed out, defaults to False
|
|
140 |
isalive -- Use True to indicate we pass if the process exists; however, by default
|
|
141 |
the test will pass if the process does not exist (isalive == False)
|
|
142 |
expectedfail -- Defaults to [], used to indicate a list of fields that are expected to fail
|
|
143 |
"""
|
|
144 |
if 'returncode' in expectedfail: |
|
145 |
self.assertTrue(returncode, "Detected an unexpected return code of: %s" % returncode) |
|
146 |
elif not isalive: |
|
147 |
self.assertTrue(returncode == 0, "Detected non-zero return code of: %d" % returncode) |
|
148 |
||
149 |
if 'didtimeout' in expectedfail: |
|
150 |
self.assertTrue(didtimeout, "Detected that process didn't time out") |
|
151 |
else: |
|
152 |
self.assertTrue(not didtimeout, "Detected that process timed out") |
|
153 |
||
154 |
if isalive: |
|
155 |
self.assertTrue(detected, "Detected process is not running, process output: %s" % output) |
|
156 |
else: |
|
157 |
self.assertTrue(not detected, "Detected process is still running, process output: %s" % output) |
|
158 |
||
159 |
if __name__ == '__main__': |
|
160 |
unittest.main() |