From 0487215804c9025326eca4a2cfd44deb614afc48 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Tue, 30 Nov 2021 13:54:07 +0200 Subject: [PATCH] bpo-13236: Flush the output stream more often in unittest It can prevent some losses when output to buffered stream. --- Lib/unittest/runner.py | 6 ++++ Lib/unittest/test/test_program.py | 16 +++++++-- Lib/unittest/test/test_result.py | 35 +++++++++++++++++-- .../2021-11-30-13-52-02.bpo-13236.FmJIkO.rst | 2 ++ 4 files changed, 54 insertions(+), 5 deletions(-) create mode 100644 Misc/NEWS.d/next/Library/2021-11-30-13-52-02.bpo-13236.FmJIkO.rst diff --git a/Lib/unittest/runner.py b/Lib/unittest/runner.py index b51def8882fc11..120ee6aba35708 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 93a61097e93c74..cc02b227e1da8a 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 c4a49b48752dc9..224a78463edc98 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 00000000000000..bfea8d4fca0e0a --- /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.