Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 72 additions & 0 deletions test/color_runner.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,72 @@
# Copyright 2025 The Emscripten Authors. All rights reserved.
# Emscripten is available under two separate licenses, the MIT license and the
# University of Illinois/NCSA Open Source License. Both these licenses can be
# found in the LICENSE file.

import logging
import unittest

from tools.colored_logger import CYAN, GREEN, RED, with_color


class BufferingMixin:
"""This class takes care of redirectting `logging` output in `buffer=True` mode.
To use this class inherit from it along with a one of the standard unittest result
classe.
"""
def _setupStdout(self):
super()._setupStdout()
# In addition to redirecting sys.stderr and sys.stdout, also update the python
# loggers have cached versions of these.
if self.buffer:
for handler in logging.root.handlers:
if handler.stream == self._original_stderr:
handler.stream = self._stderr_buffer

def _restoreStdout(self):
super()._restoreStdout()
if self.buffer:
for handler in logging.root.handlers:
if handler.stream == self._stderr_buffer:
handler.stream = self._original_stderr


class ProgressMixin:
test_count = 0
progress_counter = 0

def startTest(self, test):
assert self.test_count > 0
self.progress_counter += 1
if self.showAll:
progress = f'[{self.progress_counter}/{self.test_count}] '
self.stream.write(with_color(CYAN, progress))
super().startTest(test)


class ColorTextResult(BufferingMixin, ProgressMixin, unittest.TextTestResult):
"""Adds color the printed test result."""
def _write_status(self, test, status):
# Add some color to the status message
if status == 'ok':
color = GREEN
elif status.isupper():
color = RED
else:
color = CYAN
super()._write_status(test, with_color(color, status))


class ColorTextRunner(unittest.TextTestRunner):
"""Subclass of TextTestRunner that uses ColorTextResult"""
resultclass = ColorTextResult

def _makeResult(self):
result = super()._makeResult()
result.test_count = self.test_count
return result

def run(self, test):
self.test_count = test.countTestCases()
return super().run(test)
5 changes: 2 additions & 3 deletions test/parallel_testsuite.py
Original file line number Diff line number Diff line change
Expand Up @@ -125,9 +125,8 @@ def addTest(self, test):
test.is_parallel = True

def printOneResult(self, res):
percent = int(self.progress_counter * 100 / self.num_tests)
progress = f'[{percent:2d}%] '
self.progress_counter += 1
progress = f'[{self.progress_counter}/{self.num_tests}] '

if res.test_result == 'success':
msg = 'ok'
Expand Down Expand Up @@ -165,7 +164,7 @@ def run(self, result):
# multiprocessing.set_start_method('spawn')

tests = self.get_sorted_tests()
self.num_tests = len(tests)
self.num_tests = self.countTestCases()
contains_browser_test = any(test.is_browser_test() for test in tests)
use_cores = cap_max_workers_in_pool(min(self.max_cores, len(tests), num_cores()), contains_browser_test)
errlog(f'Using {use_cores} parallel test processes')
Expand Down
21 changes: 19 additions & 2 deletions test/runner.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,9 +40,11 @@
import common
import jsrun
import parallel_testsuite
from color_runner import ColorTextRunner
from common import errlog
from single_line_runner import SingleLineTestRunner

from tools import config, shared, utils
from tools import colored_logger, config, shared, utils

logger = logging.getLogger("runner")

Expand Down Expand Up @@ -427,8 +429,12 @@ def run_tests(options, suites):
testRunner = xmlrunner.XMLTestRunner(output=output, verbosity=2,
failfast=options.failfast)
print('Writing XML test output to ' + os.path.abspath(output.name))
elif options.buffer and options.ansi and not options.verbose:
# And buffering is enabled and ansi color output is available use our nice single-line
# result display.
testRunner = SingleLineTestRunner(verbosity=2, failfast=options.failfast)
else:
testRunner = unittest.TextTestRunner(verbosity=2, buffer=options.buffer, failfast=options.failfast)
testRunner = ColorTextRunner(verbosity=2, failfast=options.failfast)

total_core_time = 0
run_start_time = time.perf_counter()
Expand Down Expand Up @@ -467,6 +473,9 @@ def parse_args():
parser.add_argument('--no-clean', action='store_true',
help='Do not clean the temporary directory before each test run')
parser.add_argument('--verbose', '-v', action='store_true')
# TODO: Replace with BooleanOptionalAction once we can depend on python3.9
parser.add_argument('--ansi', action='store_true', default=None)
parser.add_argument('--no-ansi', action='store_false', dest='ansi', default=None)
parser.add_argument('--all-engines', action='store_true')
parser.add_argument('--detect-leaks', action='store_true')
parser.add_argument('--skip-slow', action='store_true', help='Skip tests marked as slow')
Expand Down Expand Up @@ -499,6 +508,14 @@ def parse_args():

options = parser.parse_args()

if options.ansi is None:
options.ansi = colored_logger.ansi_color_available()
else:
if options.ansi:
colored_logger.enable(force=True)
else:
colored_logger.disable()

if options.failfast:
if options.max_failures != 2**31 - 1:
utils.exit_with_error('--failfast and --max-failures are mutually exclusive!')
Expand Down
95 changes: 95 additions & 0 deletions test/single_line_runner.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,95 @@
# Copyright 2025 The Emscripten Authors. All rights reserved.
# Emscripten is available under two separate licenses, the MIT license and the
# University of Illinois/NCSA Open Source License. Both these licenses can be
# found in the LICENSE file.

import shutil
import unittest

from color_runner import BufferingMixin, ColorTextRunner

from tools.colored_logger import CYAN, GREEN, RED, with_color


def clearline(stream):
stream.write('\r\033[K')
stream.flush()


def term_width():
return shutil.get_terminal_size()[0]


class SingleLineTestResult(BufferingMixin, unittest.TextTestResult):
"""Similar to the standard TextTestResult but uses ANSI escape codes
for color output and reusing a single line on the terminal.
"""

def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.progress_counter = 0

def writeStatusLine(self, line):
clearline(self._original_stderr)
self._original_stderr.write(line)
self._original_stderr.flush()

def updateStatus(self, test, msg, color):
progress = f'[{self.progress_counter}/{self.test_count}] '
# Format the line so that it fix within the terminal width, unless its less then min_len
# in which case there is not much we can do, and we just overflow the line.
min_len = len(progress) + len(msg) + 5
test_name = str(test)
if term_width() > min_len:
max_name = term_width() - min_len
test_name = test_name[:max_name]
line = f'{with_color(CYAN, progress)}{test_name} ... {with_color(color, msg)}'
self.writeStatusLine(line)

def startTest(self, test):
self.progress_counter += 1
assert self.test_count > 0
# Note: We explicitly do not use `super()` here but instead call `unittest.TestResult`. i.e.
# we skip the superclass (since we don't want its specific behaviour) and instead call its
# superclass.
unittest.TestResult.startTest(self, test)
if self.progress_counter == 1:
self.updateStatus(test, '', GREEN)

def addSuccess(self, test):
unittest.TestResult.addSuccess(self, test)
self.updateStatus(test, 'ok', GREEN)

def addFailure(self, test, err):
unittest.TestResult.addFailure(self, test, err)
self.updateStatus(test, 'FAIL', RED)

def addError(self, test, err):
unittest.TestResult.addError(self, test, err)
self.updateStatus(test, 'ERROR', RED)

def addExpectedFailure(self, test, err):
unittest.TestResult.addExpectedFailure(self, test, err)
self.updateStatus(test, 'expected failure', RED)

def addUnexpectedSuccess(self, test, err):
unittest.TestResult.addUnexpectedSuccess(self, test, err)
self.updateStatus(test, 'UNEXPECTED SUCCESS', RED)

def addSkip(self, test, reason):
unittest.TestResult.addSkip(self, test, reason)
self.updateStatus(test, f"skipped '{reason}'", CYAN)

def printErrors(self):
# All tests have been run at this point so print a final newline
# to end out status line
self._original_stderr.write('\n')
super().printErrors()


class SingleLineTestRunner(ColorTextRunner):
"""Subclass of TextTestResult that uses SingleLineTestResult"""
resultclass = SingleLineTestResult

def __init__(self, *args, **kwargs):
super().__init__(*args, buffer=True, **kwargs)