| Message ID | 20210323181928.311862-2-pbonzini@redhat.com (mailing list archive) |
|---|---|
| State | New, archived |
| Headers | show |
| Series | qemu-iotests: quality of life improvements | expand |
23.03.2021 21:19, Paolo Bonzini wrote: > 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 <pbonzini@redhat.com> Reviewed-by: Vladimir Sementsov-Ogievskiy <vsementsov@virtuozzo.com> some non-critical notes below > --- > 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 > You don't use typing everywhere.. But iotests.py doesn't use it everywhere as well > +# 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__'): why __getstate__ is here? It's something about pickle.. > + raise AttributeError(attr) > + return getattr(self.stream,attr) s/,/, / > + > + 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): actually, neither stream nor resultclass arguments are used in the code, so it seems we don't need them? > + 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] = (), >
23.03.2021 21:54, Vladimir Sementsov-Ogievskiy wrote: >> +class ReproducibleTestRunner(unittest.TextTestRunner): >> + def __init__(self, stream: Optional[io.TextIOBase] = None, >> + resultclass: Type = ReproducibleTestResult, *args, **kwargs): > > actually, neither stream nor resultclass arguments are used in the code, so it seems we don't need them? Ah, we probably should support them to be prepared for the following patch.
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] = (),
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 <pbonzini@redhat.com> --- tests/qemu-iotests/iotests.py | 68 ++++++++++++++++++++--------------- 1 file changed, 40 insertions(+), 28 deletions(-)