gh-109162: libregrtest: add Logger class (#109212)

* Add Logger class in a new logger.py file.
* Move Regrtest attributes to Logger:

  * start_time
  * test_count_text
  * test_count_width
  * win_load_tracker

* Move Regrtest method to Logger:

  * log()
  * getloadavg(): rename to get_load_avg()
  * set_tests()

* Add methods to the Logger class:

  * start_load_tracker()
  * stop_load_tracker()

Author: vstinner

Lib/test/libregrtest/logger.py

@@ -0,0 +1,69 @@
+import os
+import time
+
+from test.libregrtest.runtest import RunTests
+from test.libregrtest.utils import print_warning, MS_WINDOWS
+
+if MS_WINDOWS:
+    from test.libregrtest.win_utils import WindowsLoadTracker
+
+
+class Logger:
+    def __init__(self):
+        self.start_time = time.perf_counter()
+        self.test_count_text = ''
+        self.test_count_width = 3
+        self.win_load_tracker = None
+
+    def log(self, line: str = '') -> None:
+        empty = not line
+
+        # add the system load prefix: "load avg: 1.80 "
+        load_avg = self.get_load_avg()
+        if load_avg is not None:
+            line = f"load avg: {load_avg:.2f} {line}"
+
+        # add the timestamp prefix:  "0:01:05 "
+        test_time = time.perf_counter() - self.start_time
+
+        mins, secs = divmod(int(test_time), 60)
+        hours, mins = divmod(mins, 60)
+        test_time = "%d:%02d:%02d" % (hours, mins, secs)
+
+        line = f"{test_time} {line}"
+        if empty:
+            line = line[:-1]
+
+        print(line, flush=True)
+
+    def get_load_avg(self) -> float | None:
+        if hasattr(os, 'getloadavg'):
+            return os.getloadavg()[0]
+        if self.win_load_tracker is not None:
+            return self.win_load_tracker.getloadavg()
+        return None
+
+    def set_tests(self, runtests: RunTests) -> None:
+        if runtests.forever:
+            self.test_count_text = ''
+            self.test_count_width = 3
+        else:
+            self.test_count_text = '/{}'.format(len(runtests.tests))
+            self.test_count_width = len(self.test_count_text) - 1
+
+    def start_load_tracker(self) -> None:
+        if not MS_WINDOWS:
+            return
+
+        try:
+            self.win_load_tracker = WindowsLoadTracker()
+        except PermissionError as error:
+            # Standard accounts may not have access to the performance
+            # counters.
+            print_warning(f'Failed to create WindowsLoadTracker: {error}')
+
+    def stop_load_tracker(self) -> None:
+        if self.win_load_tracker is None:
+            return
+        self.win_load_tracker.close()
+        self.win_load_tracker = None

Lib/test/libregrtest/main.py

@@ -10,6 +10,7 @@
 import time
 import unittest
 from test.libregrtest.cmdline import _parse_args, Namespace
+from test.libregrtest.logger import Logger
 from test.libregrtest.runtest import (
     findtests, split_test_packages, run_single_test, abs_module_name,
     PROGRESS_MIN_TIME, State, RunTests, HuntRefleak,
@@ -54,6 +55,8 @@ class Regrtest:
     on the command line.
     """
     def __init__(self, ns: Namespace):
+        self.logger = Logger()
+
         # Actions
         self.want_header: bool = ns.header
         self.want_list_tests: bool = ns.list_tests
@@ -137,29 +140,8 @@ def __init__(self, ns: Namespace):
         self.next_single_test: TestName | None = None
         self.next_single_filename: StrPath | None = None
 
-        # misc
-        self.win_load_tracker = None
-
     def log(self, line=''):
-        empty = not line
-
-        # add the system load prefix: "load avg: 1.80 "
-        load_avg = self.getloadavg()
-        if load_avg is not None:
-            line = f"load avg: {load_avg:.2f} {line}"
-
-        # add the timestamp prefix:  "0:01:05 "
-        test_time = time.perf_counter() - self.start_time
-
-        mins, secs = divmod(int(test_time), 60)
-        hours, mins = divmod(mins, 60)
-        test_time = "%d:%02d:%02d" % (hours, mins, secs)
-
-        line = f"{test_time} {line}"
-        if empty:
-            line = line[:-1]
-
-        print(line, flush=True)
+        self.logger.log(line)
 
     def display_progress(self, test_index, text):
         if self.quiet:
@@ -293,7 +275,7 @@ def _rerun_failed_tests(self, runtests: RunTests):
             fail_fast=False,
             match_tests_dict=match_tests_dict,
             output_on_failure=False)
-        self.set_tests(runtests)
+        self.logger.set_tests(runtests)
         self._run_tests_mp(runtests, self.num_workers)
         return runtests
 
@@ -437,44 +419,7 @@ def get_state(self):
 
     def _run_tests_mp(self, runtests: RunTests, num_workers: int) -> None:
         from test.libregrtest.runtest_mp import RunWorkers
-
-        # If we're on windows and this is the parent runner (not a worker),
-        # track the load average.
-        if sys.platform == 'win32':
-            from test.libregrtest.win_utils import WindowsLoadTracker
-
-            try:
-                self.win_load_tracker = WindowsLoadTracker()
-            except PermissionError as error:
-                # Standard accounts may not have access to the performance
-                # counters.
-                print(f'Failed to create WindowsLoadTracker: {error}')
-
-        try:
-            RunWorkers(self, runtests, num_workers).run()
-        finally:
-            if self.win_load_tracker is not None:
-                self.win_load_tracker.close()
-                self.win_load_tracker = None
-
-    def set_tests(self, runtests: RunTests):
-        self.tests = runtests.tests
-        if runtests.forever:
-            self.test_count_text = ''
-            self.test_count_width = 3
-        else:
-            self.test_count_text = '/{}'.format(len(self.tests))
-            self.test_count_width = len(self.test_count_text) - 1
-
-    def run_tests(self, runtests: RunTests):
-        self.first_runtests = runtests
-        self.set_tests(runtests)
-        if self.num_workers:
-            self._run_tests_mp(runtests, self.num_workers)
-            tracer = None
-        else:
-            tracer = self.run_tests_sequentially(runtests)
-        return tracer
+        RunWorkers(self, runtests, num_workers).run()
 
     def finalize_tests(self, tracer):
         if self.next_single_filename:
@@ -496,7 +441,7 @@ def finalize_tests(self, tracer):
             self.results.write_junit(self.junit_filename)
 
     def display_summary(self):
-        duration = time.perf_counter() - self.start_time
+        duration = time.perf_counter() - self.logger.start_time
         filtered = bool(self.match_tests) or bool(self.ignore_tests)
 
         # Total duration
@@ -619,35 +564,8 @@ def main(self, tests: TestList | None = None):
 
             sys.exit(exc.code)
 
-    def getloadavg(self):
-        if self.win_load_tracker is not None:
-            return self.win_load_tracker.getloadavg()
-
-        if hasattr(os, 'getloadavg'):
-            return os.getloadavg()[0]
-
-        return None
-
-    def action_run_tests(self):
-        if self.hunt_refleak and self.hunt_refleak.warmups < 3:
-            msg = ("WARNING: Running tests with --huntrleaks/-R and "
-                   "less than 3 warmup repetitions can give false positives!")
-            print(msg, file=sys.stdout, flush=True)
-
-        # For a partial run, we do not need to clutter the output.
-        if (self.want_header
-            or not(self.pgo or self.quiet or self.single_test_run
-                   or self.tests or self.cmdline_args)):
-            self.display_header()
-
-        if self.randomize:
-            print("Using random seed", self.random_seed)
-
-        if self.num_workers < 0:
-            # Use all cores + extras for tests that like to sleep
-            self.num_workers = 2 + (os.cpu_count() or 1)
-
-        runtests = RunTests(
+    def create_run_tests(self):
+        return RunTests(
             tuple(self.selected),
             fail_fast=self.fail_fast,
             match_tests=self.match_tests,
@@ -668,17 +586,53 @@ def action_run_tests(self):
             python_cmd=self.python_cmd,
         )
 
+    def run_tests(self) -> int:
+        if self.hunt_refleak and self.hunt_refleak.warmups < 3:
+            msg = ("WARNING: Running tests with --huntrleaks/-R and "
+                   "less than 3 warmup repetitions can give false positives!")
+            print(msg, file=sys.stdout, flush=True)
+
+        if self.num_workers < 0:
+            # Use all CPUs + 2 extra worker processes for tests
+            # that like to sleep
+            self.num_workers = (os.cpu_count() or 1) + 2
+
+        # For a partial run, we do not need to clutter the output.
+        if (self.want_header
+            or not(self.pgo or self.quiet or self.single_test_run
+                   or self.tests or self.cmdline_args)):
+            self.display_header()
+
+        if self.randomize:
+            print("Using random seed", self.random_seed)
+
+        runtests = self.create_run_tests()
+        self.first_runtests = runtests
+        self.logger.set_tests(runtests)
+
         setup_tests(runtests)
 
-        tracer = self.run_tests(runtests)
-        self.display_result(runtests)
+        self.logger.start_load_tracker()
+        try:
+            if self.num_workers:
+                self._run_tests_mp(runtests, self.num_workers)
+                tracer = None
+            else:
+                tracer = self.run_tests_sequentially(runtests)
 
-        if self.want_rerun and self.results.need_rerun():
-            self.rerun_failed_tests(runtests)
+            self.display_result(runtests)
+
+            if self.want_rerun and self.results.need_rerun():
+                self.rerun_failed_tests(runtests)
+        finally:
+            self.logger.stop_load_tracker()
 
         self.display_summary()
         self.finalize_tests(tracer)
 
+        return self.results.get_exitcode(self.fail_env_changed,
+                                         self.fail_rerun)
+
     def _main(self):
         if self.is_worker():
             from test.libregrtest.runtest_mp import worker_process
@@ -697,9 +651,7 @@ def _main(self):
         elif self.want_list_cases:
             self.list_cases()
         else:
-            self.action_run_tests()
-            exitcode = self.results.get_exitcode(self.fail_env_changed,
-                                                 self.fail_rerun)
+            exitcode = self.run_tests()
 
         sys.exit(exitcode)
 

Lib/test/libregrtest/runtest_mp.py

@@ -433,7 +433,7 @@ def get_running(workers: list[WorkerThread]) -> list[str]:
 class RunWorkers:
     def __init__(self, regrtest: Regrtest, runtests: RunTests, num_workers: int) -> None:
         self.results: TestResults = regrtest.results
-        self.log = regrtest.log
+        self.log = regrtest.logger.log
         self.display_progress = regrtest.display_progress
         self.num_workers = num_workers
         self.runtests = runtests

Lib/test/libregrtest/utils.py

@@ -6,6 +6,9 @@
 from test import support
 
 
+MS_WINDOWS = (sys.platform == 'win32')
+
+
 def format_duration(seconds):
     ms = math.ceil(seconds * 1e3)
     seconds, ms = divmod(ms, 1000)