|
|
@@ -20,7 +20,6 @@
|
|
|
import bz2
|
|
|
from collections import OrderedDict
|
|
|
import faulthandler
|
|
|
-import io
|
|
|
import json
|
|
|
import logging
|
|
|
import os
|
|
|
@@ -32,7 +31,7 @@
|
|
|
import sys
|
|
|
import time
|
|
|
from typing import (Any, Callable, Dict, Iterable,
|
|
|
- List, Optional, Sequence, Tuple, TypeVar)
|
|
|
+ List, Optional, Sequence, TextIO, Tuple, Type, TypeVar)
|
|
|
import unittest
|
|
|
|
|
|
from contextlib import contextmanager
|
|
|
@@ -1271,37 +1270,50 @@ 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:
|
|
|
+ def __init__(self, stream: TextIO):
|
|
|
+ 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[TextIO] = None,
|
|
|
+ resultclass: Type[unittest.TestResult] = ReproducibleTestResult,
|
|
|
+ **kwargs: Any) -> None:
|
|
|
+ rstream = ReproducibleStreamWrapper(stream or sys.stdout)
|
|
|
+ super().__init__(stream=rstream, # type: ignore
|
|
|
+ descriptions=True,
|
|
|
+ resultclass=resultclass,
|
|
|
+ **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 = ReproducibleTestRunner(verbosity=verbosity)
|
|
|
+ unittest.main(testRunner=runner)
|
|
|
|
|
|
def execute_setup_common(supported_fmts: Sequence[str] = (),
|
|
|
supported_platforms: Sequence[str] = (),
|
Paolo Bonzini