Instead of buffering the test output into a StringIO, patch it on
the fly by wrapping sys.stdout's write method. This can be
done unconditionally, even if using -d, which makes execute_unittest
a bit simpler.
Signed-off-by: Paolo Bonzini <pbonz...@redhat.com>
---
tests/qemu-iotests/iotests.py | 68 ++++++++++++++++++++---------------
1 file changed, 40 insertions(+), 28 deletions(-)
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index 90d0b62523..0521235030 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -32,7 +32,7 @@
import sys
import time
from typing import (Any, Callable, Dict, Iterable,
- List, Optional, Sequence, Tuple, TypeVar)
+ List, Optional, Sequence, Tuple, Type, TypeVar)
import unittest
from contextlib import contextmanager
@@ -1271,37 +1271,49 @@ def func_wrapper(*args, **kwargs):
return func(*args, **kwargs)
return func_wrapper
+# We need to filter out the time taken from the output so that
+# qemu-iotest can reliably diff the results against master output,
+# and hide skipped tests from the reference output.
+
+class ReproducibleTestResult(unittest.TextTestResult):
+ def addSkip(self, test, reason):
+ # Same as TextTestResult, but print dot instead of "s"
+ unittest.TestResult.addSkip(self, test, reason)
+ if self.showAll:
+ self.stream.writeln("skipped {0!r}".format(reason))
+ elif self.dots:
+ self.stream.write(".")
+ self.stream.flush()
+
+class ReproducibleStreamWrapper(object):
+ def __init__(self, stream):
+ self.stream = stream
+
+ def __getattr__(self, attr):
+ if attr in ('stream', '__getstate__'):
+ raise AttributeError(attr)
+ return getattr(self.stream,attr)
+
+ def write(self, arg=None):
+ arg = re.sub(r'Ran (\d+) tests? in [\d.]+s', r'Ran \1 tests', arg)
+ arg = re.sub(r' \(skipped=\d+\)', r'', arg)
+ self.stream.write(arg)
+
+class ReproducibleTestRunner(unittest.TextTestRunner):
+ def __init__(self, stream: Optional[io.TextIOBase] = None,
+ resultclass: Type = ReproducibleTestResult, *args, **kwargs):
+ rstream = ReproducibleStreamWrapper(stream or sys.stdout)
+ super().__init__(stream=rstream, # type: ignore
+ descriptions=True,
+ resultclass=resultclass,
+ *args, **kwargs)
+
def execute_unittest(debug=False):
"""Executes unittests within the calling module."""
verbosity = 2 if debug else 1
-
- if debug:
- output = sys.stdout
- else:
- # We need to filter out the time taken from the output so that
- # qemu-iotest can reliably diff the results against master output.
- output = io.StringIO()
-
- runner = unittest.TextTestRunner(stream=output, descriptions=True,
- verbosity=verbosity)
- try:
- # unittest.main() will use sys.exit(); so expect a SystemExit
- # exception
- unittest.main(testRunner=runner)
- finally:
- # We need to filter out the time taken from the output so that
- # qemu-iotest can reliably diff the results against master output.
- if not debug:
- out = output.getvalue()
- out = re.sub(r'Ran (\d+) tests? in [\d.]+s', r'Ran \1 tests', out)
-
- # Hide skipped tests from the reference output
- out = re.sub(r'OK \(skipped=\d+\)', 'OK', out)
- out_first_line, out_rest = out.split('\n', 1)
- out = out_first_line.replace('s', '.') + '\n' + out_rest
-
- sys.stderr.write(out)
+ runner = unittest.ReproducibleTestRunner(verbosity=verbosity)
+ unittest.main(testRunner=runner)
def execute_setup_common(supported_fmts: Sequence[str] = (),
supported_platforms: Sequence[str] = (),
--
2.30.1
