[Python-checkins] bpo-13236: Flush the output stream more often in unittest (GH-29864)
serhiy-storchaka
webhook-mailer at python.org
Sat Dec 4 13:57:28 EST 2021
https://github.com/python/cpython/commit/f42a06ba279c916fb67289e47f9bc60dc5dee4ee
commit: f42a06ba279c916fb67289e47f9bc60dc5dee4ee
branch: main
author: Serhiy Storchaka <storchaka at gmail.com>
committer: serhiy-storchaka <storchaka at gmail.com>
date: 2021-12-04T20:57:20+02:00
summary:
bpo-13236: Flush the output stream more often in unittest (GH-29864)
It can prevent some losses when output to buffered stream.
files:
A Misc/NEWS.d/next/Library/2021-11-30-13-52-02.bpo-13236.FmJIkO.rst
M Lib/unittest/runner.py
M Lib/unittest/test/test_program.py
M Lib/unittest/test/test_result.py
diff --git a/Lib/unittest/runner.py b/Lib/unittest/runner.py
index b51def8882fc1..120ee6aba3570 100644
--- a/Lib/unittest/runner.py
+++ b/Lib/unittest/runner.py
@@ -68,6 +68,7 @@ def _write_status(self, test, status):
self.stream.write(self.getDescription(test))
self.stream.write(" ... ")
self.stream.writeln(status)
+ self.stream.flush()
self._newline = True
def addSubTest(self, test, subtest, err):
@@ -121,6 +122,7 @@ def addExpectedFailure(self, test, err):
super(TextTestResult, self).addExpectedFailure(test, err)
if self.showAll:
self.stream.writeln("expected failure")
+ self.stream.flush()
elif self.dots:
self.stream.write("x")
self.stream.flush()
@@ -129,6 +131,7 @@ def addUnexpectedSuccess(self, test):
super(TextTestResult, self).addUnexpectedSuccess(test)
if self.showAll:
self.stream.writeln("unexpected success")
+ self.stream.flush()
elif self.dots:
self.stream.write("u")
self.stream.flush()
@@ -136,6 +139,7 @@ def addUnexpectedSuccess(self, test):
def printErrors(self):
if self.dots or self.showAll:
self.stream.writeln()
+ self.stream.flush()
self.printErrorList('ERROR', self.errors)
self.printErrorList('FAIL', self.failures)
@@ -145,6 +149,7 @@ def printErrorList(self, flavour, errors):
self.stream.writeln("%s: %s" % (flavour,self.getDescription(test)))
self.stream.writeln(self.separator2)
self.stream.writeln("%s" % err)
+ self.stream.flush()
class TextTestRunner(object):
@@ -239,4 +244,5 @@ def run(self, test):
self.stream.writeln(" (%s)" % (", ".join(infos),))
else:
self.stream.write("\n")
+ self.stream.flush()
return result
diff --git a/Lib/unittest/test/test_program.py b/Lib/unittest/test/test_program.py
index 93a61097e93c7..cc02b227e1da8 100644
--- a/Lib/unittest/test/test_program.py
+++ b/Lib/unittest/test/test_program.py
@@ -6,6 +6,7 @@
from test import support
import unittest
import unittest.test
+from .test_result import BufferedWriter
class Test_TestProgram(unittest.TestCase):
@@ -104,30 +105,39 @@ def run(self, test):
program.testNames)
def test_NonExit(self):
+ stream = BufferedWriter()
program = unittest.main(exit=False,
argv=["foobar"],
- testRunner=unittest.TextTestRunner(stream=io.StringIO()),
+ testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.FooBarLoader())
self.assertTrue(hasattr(program, 'result'))
+ self.assertIn('\nFAIL: testFail ', stream.getvalue())
+ self.assertTrue(stream.getvalue().endswith('\n\nFAILED (failures=1)\n'))
def test_Exit(self):
+ stream = BufferedWriter()
self.assertRaises(
SystemExit,
unittest.main,
argv=["foobar"],
- testRunner=unittest.TextTestRunner(stream=io.StringIO()),
+ testRunner=unittest.TextTestRunner(stream=stream),
exit=True,
testLoader=self.FooBarLoader())
+ self.assertIn('\nFAIL: testFail ', stream.getvalue())
+ self.assertTrue(stream.getvalue().endswith('\n\nFAILED (failures=1)\n'))
def test_ExitAsDefault(self):
+ stream = BufferedWriter()
self.assertRaises(
SystemExit,
unittest.main,
argv=["foobar"],
- testRunner=unittest.TextTestRunner(stream=io.StringIO()),
+ testRunner=unittest.TextTestRunner(stream=stream),
testLoader=self.FooBarLoader())
+ self.assertIn('\nFAIL: testFail ', stream.getvalue())
+ self.assertTrue(stream.getvalue().endswith('\n\nFAILED (failures=1)\n'))
class InitialisableProgram(unittest.TestProgram):
diff --git a/Lib/unittest/test/test_result.py b/Lib/unittest/test/test_result.py
index c4a49b48752dc..224a78463edc9 100644
--- a/Lib/unittest/test/test_result.py
+++ b/Lib/unittest/test/test_result.py
@@ -33,6 +33,22 @@ def bad_cleanup2():
raise ValueError('bad cleanup2')
+class BufferedWriter:
+ def __init__(self):
+ self.result = ''
+ self.buffer = ''
+
+ def write(self, arg):
+ self.buffer += arg
+
+ def flush(self):
+ self.result += self.buffer
+ self.buffer = ''
+
+ def getvalue(self):
+ return self.result
+
+
class Test_TestResult(unittest.TestCase):
# Note: there are not separate tests for TestResult.wasSuccessful(),
# TestResult.errors, TestResult.failures, TestResult.testsRun or
@@ -335,10 +351,13 @@ def testFailFast(self):
self.assertTrue(result.shouldStop)
def testFailFastSetByRunner(self):
- runner = unittest.TextTestRunner(stream=io.StringIO(), failfast=True)
+ stream = BufferedWriter()
+ runner = unittest.TextTestRunner(stream=stream, failfast=True)
def test(result):
self.assertTrue(result.failfast)
result = runner.run(test)
+ stream.flush()
+ self.assertTrue(stream.getvalue().endswith('\n\nOK\n'))
class Test_TextTestResult(unittest.TestCase):
@@ -462,6 +481,12 @@ def testFail(self):
self.fail('fail')
def testError(self):
raise Exception('error')
+ @unittest.expectedFailure
+ def testExpectedFailure(self):
+ self.fail('fail')
+ @unittest.expectedFailure
+ def testUnexpectedSuccess(self):
+ pass
def testSubTestSuccess(self):
with self.subTest('one', a=1):
pass
@@ -483,7 +508,7 @@ def tearDown(self):
raise self.tearDownError
def _run_test(self, test_name, verbosity, tearDownError=None):
- stream = io.StringIO()
+ stream = BufferedWriter()
stream = unittest.runner._WritelnDecorator(stream)
result = unittest.TextTestResult(stream, True, verbosity)
test = self.Test(test_name)
@@ -496,6 +521,8 @@ def testDotsOutput(self):
self.assertEqual(self._run_test('testSkip', 1), 's')
self.assertEqual(self._run_test('testFail', 1), 'F')
self.assertEqual(self._run_test('testError', 1), 'E')
+ self.assertEqual(self._run_test('testExpectedFailure', 1), 'x')
+ self.assertEqual(self._run_test('testUnexpectedSuccess', 1), 'u')
def testLongOutput(self):
classname = f'{__name__}.{self.Test.__qualname__}'
@@ -507,6 +534,10 @@ def testLongOutput(self):
f'testFail ({classname}) ... FAIL\n')
self.assertEqual(self._run_test('testError', 2),
f'testError ({classname}) ... ERROR\n')
+ self.assertEqual(self._run_test('testExpectedFailure', 2),
+ f'testExpectedFailure ({classname}) ... expected failure\n')
+ self.assertEqual(self._run_test('testUnexpectedSuccess', 2),
+ f'testUnexpectedSuccess ({classname}) ... unexpected success\n')
def testDotsOutputSubTestSuccess(self):
self.assertEqual(self._run_test('testSubTestSuccess', 1), '.')
diff --git a/Misc/NEWS.d/next/Library/2021-11-30-13-52-02.bpo-13236.FmJIkO.rst b/Misc/NEWS.d/next/Library/2021-11-30-13-52-02.bpo-13236.FmJIkO.rst
new file mode 100644
index 0000000000000..bfea8d4fca0e0
--- /dev/null
+++ b/Misc/NEWS.d/next/Library/2021-11-30-13-52-02.bpo-13236.FmJIkO.rst
@@ -0,0 +1,2 @@
+:class:`unittest.TextTestResult` and :class:`unittest.TextTestRunner` flush
+now the output stream more often.
More information about the Python-checkins
mailing list