2 # subunit: extensions to python unittest to get test results from subprocesses.
3 # Copyright (C) 2005 Robert Collins <robertc@robertcollins.net>
4 # Copyright (C) 2007 Jelmer Vernooij <jelmer@samba.org>
6 # This program is free software; you can redistribute it and/or modify
7 # it under the terms of the GNU General Public License as published by
8 # the Free Software Foundation; either version 2 of the License, or
9 # (at your option) any later version.
11 # This program is distributed in the hope that it will be useful,
12 # but WITHOUT ANY WARRANTY; without even the implied warranty of
13 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14 # GNU General Public License for more details.
16 # You should have received a copy of the GNU General Public License
17 # along with this program; if not, write to the Free Software
18 # Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
22 from StringIO import StringIO
28 return subunit.tests.test_suite()
31 def join_dir(base_path, path):
33 Returns an absolute path to C{path}, calculated relative to the parent
36 @param base_path: A path to a file or directory.
37 @param path: An absolute path, or a path relative to the containing
38 directory of C{base_path}.
40 @return: An absolute path to C{path}.
42 return os.path.join(os.path.dirname(os.path.abspath(base_path)), path)
45 class TestProtocolServer(object):
46 """A class for receiving results from a TestProtocol client."""
53 def __init__(self, client, stream=sys.stdout):
54 """Create a TestProtocol server instance.
56 client should be an object that provides
62 methods, i.e. a TestResult.
64 self.state = TestProtocolServer.OUTSIDE_TEST
68 def _addError(self, offset, line):
69 if (self.state == TestProtocolServer.TEST_STARTED and
70 self.current_test_description == line[offset:-1]):
71 self.state = TestProtocolServer.OUTSIDE_TEST
72 self.current_test_description = None
73 self.client.addError(self._current_test, RemoteError(""))
74 self.client.stopTest(self._current_test)
75 self._current_test = None
76 elif (self.state == TestProtocolServer.TEST_STARTED and
77 self.current_test_description + " [" == line[offset:-1]):
78 self.state = TestProtocolServer.READING_ERROR
81 self.stdOutLineReceived(line)
83 def _addFailure(self, offset, line):
84 if (self.state == TestProtocolServer.TEST_STARTED and
85 self.current_test_description == line[offset:-1]):
86 self.state = TestProtocolServer.OUTSIDE_TEST
87 self.current_test_description = None
88 self.client.addFailure(self._current_test, RemoteError())
89 self.client.stopTest(self._current_test)
90 elif (self.state == TestProtocolServer.TEST_STARTED and
91 self.current_test_description + " [" == line[offset:-1]):
92 self.state = TestProtocolServer.READING_FAILURE
95 self.stdOutLineReceived(line)
97 def _addSuccess(self, offset, line):
98 if (self.state == TestProtocolServer.TEST_STARTED and
99 self.current_test_description == line[offset:-1]):
100 self.client.addSuccess(self._current_test)
101 self.client.stopTest(self._current_test)
102 self.current_test_description = None
103 self._current_test = None
104 self.state = TestProtocolServer.OUTSIDE_TEST
106 self.stdOutLineReceived(line)
108 def _appendMessage(self, line):
109 if line[0:2] == " ]":
111 self._message += line[1:]
113 self._message += line
115 def endQuote(self, line):
116 if self.state == TestProtocolServer.READING_FAILURE:
117 self.state = TestProtocolServer.OUTSIDE_TEST
118 self.current_test_description = None
119 self.client.addFailure(self._current_test,
120 RemoteError(self._message))
121 self.client.stopTest(self._current_test)
122 elif self.state == TestProtocolServer.READING_ERROR:
123 self.state = TestProtocolServer.OUTSIDE_TEST
124 self.current_test_description = None
125 self.client.addError(self._current_test,
126 RemoteError(self._message))
127 self.client.stopTest(self._current_test)
129 self.stdOutLineReceived(line)
131 def lineReceived(self, line):
132 """Call the appropriate local method for the received line."""
135 elif (self.state == TestProtocolServer.READING_FAILURE or
136 self.state == TestProtocolServer.READING_ERROR):
137 self._appendMessage(line)
139 parts = line.split(None, 1)
142 offset = len(cmd) + 1
144 if cmd in ('test', 'testing'):
145 self._startTest(offset, line)
147 self._addError(offset, line)
148 elif cmd == 'failure':
149 self._addFailure(offset, line)
150 elif cmd in ('success', 'successful'):
151 self._addSuccess(offset, line)
153 self.stdOutLineReceived(line)
155 self.stdOutLineReceived(line)
157 def lostConnection(self):
158 """The input connection has finished."""
159 if self.state == TestProtocolServer.TEST_STARTED:
160 self.client.addError(self._current_test,
161 RemoteError("lost connection during test '%s'"
162 % self.current_test_description))
163 self.client.stopTest(self._current_test)
164 elif self.state == TestProtocolServer.READING_ERROR:
165 self.client.addError(self._current_test,
166 RemoteError("lost connection during "
167 "error report of test "
169 self.current_test_description))
170 self.client.stopTest(self._current_test)
171 elif self.state == TestProtocolServer.READING_FAILURE:
172 self.client.addError(self._current_test,
173 RemoteError("lost connection during "
174 "failure report of test "
176 self.current_test_description))
177 self.client.stopTest(self._current_test)
179 def readFrom(self, pipe):
180 for line in pipe.readlines():
181 self.lineReceived(line)
182 self.lostConnection()
184 def _startTest(self, offset, line):
185 """Internal call to change state machine. Override startTest()."""
186 if self.state == TestProtocolServer.OUTSIDE_TEST:
187 self.state = TestProtocolServer.TEST_STARTED
188 self._current_test = RemotedTestCase(line[offset:-1])
189 self.current_test_description = line[offset:-1]
190 self.client.startTest(self._current_test)
192 self.stdOutLineReceived(line)
194 def stdOutLineReceived(self, line):
195 self._stream.write(line)
198 class RemoteException(Exception):
199 """An exception that occured remotely to python."""
201 def __eq__(self, other):
203 return self.args == other.args
204 except AttributeError:
208 class TestProtocolClient(unittest.TestResult):
209 """A class that looks like a TestResult and informs a TestProtocolServer."""
211 def __init__(self, stream):
212 super(TestProtocolClient, self).__init__()
213 self._stream = stream
215 def addError(self, test, error):
216 """Report an error in test test."""
217 self._stream.write("error: %s [\n" % (test.shortDescription() or str(test)))
218 for line in self._exc_info_to_string(error, test).splitlines():
219 self._stream.write("%s\n" % line)
220 self._stream.write("]\n")
221 super(TestProtocolClient, self).addError(test, error)
223 def addFailure(self, test, error):
224 """Report a failure in test test."""
225 self._stream.write("failure: %s [\n" % (test.shortDescription() or str(test)))
226 for line in self._exc_info_to_string(error, test).splitlines():
227 self._stream.write("%s\n" % line)
228 self._stream.write("]\n")
229 super(TestProtocolClient, self).addFailure(test, error)
231 def addSuccess(self, test):
232 """Report a success in a test."""
233 self._stream.write("successful: %s\n" % (test.shortDescription() or str(test)))
234 super(TestProtocolClient, self).addSuccess(test)
236 def startTest(self, test):
237 """Mark a test as starting its test run."""
238 self._stream.write("test: %s\n" % (test.shortDescription() or str(test)))
239 super(TestProtocolClient, self).startTest(test)
242 def RemoteError(description=""):
243 if description == "":
245 return (RemoteException, RemoteException(description), None)
248 class RemotedTestCase(unittest.TestCase):
249 """A class to represent test cases run in child processes."""
251 def __eq__ (self, other):
253 return self.__description == other.__description
254 except AttributeError:
257 def __init__(self, description):
258 """Create a psuedo test case with description description."""
259 self.__description = description
261 def error(self, label):
262 raise NotImplementedError("%s on RemotedTestCases is not permitted." %
269 self.error("tearDown")
271 def shortDescription(self):
272 return self.__description
275 return "%s.%s" % (self._strclass(), self.__description)
278 return "%s (%s)" % (self.__description, self._strclass())
281 return "<%s description='%s'>" % \
282 (self._strclass(), self.__description)
284 def run(self, result=None):
285 if result is None: result = self.defaultTestResult()
286 result.startTest(self)
287 result.addError(self, RemoteError("Cannot run RemotedTestCases.\n"))
288 result.stopTest(self)
292 return "%s.%s" % (cls.__module__, cls.__name__)
295 class ExecTestCase(unittest.TestCase):
296 """A test case which runs external scripts for test fixtures."""
298 def __init__(self, methodName='runTest'):
299 """Create an instance of the class that will use the named test
300 method when executed. Raises a ValueError if the instance does
301 not have a method with the specified name.
303 unittest.TestCase.__init__(self, methodName)
304 testMethod = getattr(self, methodName)
305 self.script = join_dir(sys.modules[self.__class__.__module__].__file__,
308 def countTestCases(self):
311 def run(self, result=None):
312 if result is None: result = self.defaultTestResult()
316 """Run the test without collecting errors in a TestResult"""
317 self._run(unittest.TestResult())
319 def _run(self, result):
320 protocol = TestProtocolServer(result)
321 output = os.popen(self.script, mode='r')
322 protocol.readFrom(output)
325 class IsolatedTestCase(unittest.TestCase):
326 """A TestCase which runs its tests in a forked process."""
328 def run(self, result=None):
329 if result is None: result = self.defaultTestResult()
330 run_isolated(unittest.TestCase, self, result)
333 class IsolatedTestSuite(unittest.TestSuite):
334 """A TestCase which runs its tests in a forked process."""
336 def run(self, result=None):
337 if result is None: result = unittest.TestResult()
338 run_isolated(unittest.TestSuite, self, result)
341 def run_isolated(klass, self, result):
342 """Run a test suite or case in a subprocess, using the run method on klass.
344 c2pread, c2pwrite = os.pipe()
345 # fixme - error -> result
350 # Close parent's pipe ends
357 # at this point, sys.stdin is redirected, now we want
358 # to filter it to escape ]'s.
359 ### XXX: test and write that bit.
361 result = TestProtocolClient(sys.stdout)
362 klass.run(self, result)
365 # exit HARD, exit NOW.
369 # Close child pipe ends
371 # hookup a protocol engine
372 protocol = TestProtocolServer(result)
373 protocol.readFrom(os.fdopen(c2pread, 'rU'))
375 # TODO return code evaluation.
379 class SubunitTestRunner:
380 def __init__(self, stream=sys.stdout):
384 "Run the given test case or test suite."
385 result = TestProtocolClient(self.stream)