69
68
OVERRIDE_PYTHON = None # to run with alternative python 'python'
72
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
73
a_callable=None, *args, **kwargs):
74
"""Call callable with redirected std io pipes.
76
Returns the return code."""
77
from StringIO import StringIO
78
if not callable(a_callable):
79
raise ValueError("a_callable must be callable.")
83
stdout = self.TEST_LOG
85
stderr = self.TEST_LOG
86
real_stdin = sys.stdin
87
real_stdout = sys.stdout
88
real_stderr = sys.stderr
94
result = a_callable(*args, **kwargs)
96
sys.stdout = real_stdout
97
sys.stderr = real_stderr
98
sys.stdin = real_stdin
102
72
super(TestCase, self).setUp()
103
73
# setup a temporary log for the test
105
77
self.TEST_LOG = tempfile.NamedTemporaryFile(mode='wt', bufsize=0)
78
# save stdout & stderr so there's no leakage from code-under-test
79
self.real_stdout = sys.stdout
80
self.real_stderr = sys.stderr
81
sys.stdout = sys.stderr = self.TEST_LOG
106
82
self.log("%s setup" % self.id())
108
84
def tearDown(self):
85
sys.stdout = self.real_stdout
86
sys.stderr = self.real_stderr
109
87
self.log("%s teardown" % self.id())
111
89
super(TestCase, self).tearDown()
232
210
except ImportError, e:
233
211
_need_subprocess()
235
214
cmd = self.formcmd(cmd)
236
215
child = Popen(cmd, stdout=PIPE, stderr=self.TEST_LOG)
237
216
outd, errd = child.communicate()
239
218
actual_retcode = child.wait()
240
220
outd = outd.replace('\r', '')
241
222
if retcode != actual_retcode:
242
223
raise CommandFailed("test failed: %r returned %d, expected %d"
243
224
% (cmd, actual_retcode, retcode))
246
230
def build_tree(self, shape):
247
231
"""Build a test tree according to a pattern.
265
249
InTempDir = FunctionalTestCase
268
class EarlyStoppingTestResultAdapter(object):
269
"""An adapter for TestResult to stop at the first first failure or error"""
271
def __init__(self, result):
272
self._result = result
274
def addError(self, test, err):
275
self._result.addError(test, err)
278
def addFailure(self, test, err):
279
self._result.addFailure(test, err)
282
def __getattr__(self, name):
283
return getattr(self._result, name)
285
def __setattr__(self, name, value):
286
if name == '_result':
287
object.__setattr__(self, name, value)
288
return setattr(self._result, name, value)
291
252
class _MyResult(unittest._TextTestResult):
293
254
Custom TestResult.
337
298
class TextTestRunner(unittest.TextTestRunner):
339
300
def _makeResult(self):
340
result = _MyResult(self.stream, self.descriptions, self.verbosity)
341
return EarlyStoppingTestResultAdapter(result)
344
class filteringVisitor(TestUtil.TestVisitor):
345
"""I accruse all the testCases I visit that pass a regexp filter on id
349
def __init__(self, filter):
351
TestUtil.TestVisitor.__init__(self)
353
self.filter=re.compile(filter)
356
"""answer the suite we are building"""
357
if self._suite is None:
358
self._suite=TestUtil.TestSuite()
361
def visitCase(self, aCase):
362
if self.filter.match(aCase.id()):
363
self.suite().addTest(aCase)
366
def run_suite(suite, name='test', verbose=False, pattern=".*"):
301
return _MyResult(self.stream, self.descriptions, self.verbosity)
304
def run_suite(suite, name='test', verbose=False):
368
306
FunctionalTestCase._TEST_NAME = name
373
311
runner = TextTestRunner(stream=sys.stdout,
375
313
verbosity=verbosity)
376
visitor = filteringVisitor(pattern)
378
result = runner.run(visitor.suite())
314
result = runner.run(suite)
379
315
# This is still a little bogus,
380
316
# but only a little. Folk not using our testrunner will
381
317
# have to delete their temp directories themselves.
382
318
if result.wasSuccessful():
383
if FunctionalTestCase.TEST_ROOT is not None:
384
shutil.rmtree(FunctionalTestCase.TEST_ROOT)
319
shutil.rmtree(FunctionalTestCase.TEST_ROOT)
386
321
print "Failed tests working directories are in '%s'\n" % FunctionalTestCase.TEST_ROOT
387
322
return result.wasSuccessful()