bpo-13236: Flush the output stream more often in unittest (GH-29864)

It can prevent some losses when output to buffered stream.

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

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):

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), '.')

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.