Re: [PATCH v2 1/5] qemu-iotests: do not buffer the test output

[Vladimir Sementsov-Ogievskiy]

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] = (),


--
Best regards,
Vladimir