gh-109413: libregrtest: Add and improve type annotations (#109405)

Author: AlexWaygood

Lib/test/libregrtest/cmdline.py

@@ -161,7 +161,7 @@ def __init__(self, **kwargs) -> None:
         self.trace = False
         self.coverdir = 'coverage'
         self.runleaks = False
-        self.huntrleaks = False
+        self.huntrleaks: tuple[int, int, str] | None = None
         self.rerun = False
         self.verbose3 = False
         self.print_slow = False

Lib/test/libregrtest/findtests.py

@@ -24,7 +24,7 @@
 }
 
 
-def findtestdir(path=None):
+def findtestdir(path: StrPath | None = None) -> StrPath:
     return path or os.path.dirname(os.path.dirname(__file__)) or os.curdir
 
 

Lib/test/libregrtest/logger.py

@@ -14,7 +14,7 @@ def __init__(self, results: TestResults, quiet: bool, pgo: bool):
         self.start_time = time.perf_counter()
         self.test_count_text = ''
         self.test_count_width = 3
-        self.win_load_tracker = None
+        self.win_load_tracker: WindowsLoadTracker | None = None
         self._results: TestResults = results
         self._quiet: bool = quiet
         self._pgo: bool = pgo

Lib/test/libregrtest/main.py

@@ -71,11 +71,11 @@ def __init__(self, ns: Namespace):
 
         # Select tests
         if ns.match_tests:
-            self.match_tests: FilterTuple = tuple(ns.match_tests)
+            self.match_tests: FilterTuple | None = tuple(ns.match_tests)
         else:
             self.match_tests = None
         if ns.ignore_tests:
-            self.ignore_tests: FilterTuple = tuple(ns.ignore_tests)
+            self.ignore_tests: FilterTuple | None = tuple(ns.ignore_tests)
         else:
             self.ignore_tests = None
         self.exclude: bool = ns.exclude
@@ -105,16 +105,16 @@ def __init__(self, ns: Namespace):
         if ns.huntrleaks:
             warmups, runs, filename = ns.huntrleaks
             filename = os.path.abspath(filename)
-            self.hunt_refleak: HuntRefleak = HuntRefleak(warmups, runs, filename)
+            self.hunt_refleak: HuntRefleak | None = HuntRefleak(warmups, runs, filename)
         else:
             self.hunt_refleak = None
         self.test_dir: StrPath | None = ns.testdir
         self.junit_filename: StrPath | None = ns.xmlpath
         self.memory_limit: str | None = ns.memlimit
         self.gc_threshold: int | None = ns.threshold
-        self.use_resources: tuple[str] = tuple(ns.use_resources)
+        self.use_resources: tuple[str, ...] = tuple(ns.use_resources)
         if ns.python:
-            self.python_cmd: tuple[str] = tuple(ns.python)
+            self.python_cmd: tuple[str, ...] | None = tuple(ns.python)
         else:
             self.python_cmd = None
         self.coverage: bool = ns.trace
@@ -389,7 +389,7 @@ def create_run_tests(self, tests: TestTuple):
             match_tests=self.match_tests,
             ignore_tests=self.ignore_tests,
             match_tests_dict=None,
-            rerun=None,
+            rerun=False,
             forever=self.forever,
             pgo=self.pgo,
             pgo_extended=self.pgo_extended,

Lib/test/libregrtest/mypy.ini

@@ -0,0 +1,47 @@
+# Config file for running mypy on libregrtest.
+#
+# Note: mypy can't be run on libregrtest from the CPython repo root.
+# If you try to do so, mypy will complain
+# about the entire `Lib/` directory "shadowing the stdlib".
+# Instead, `cd` into `Lib/test`, then run `mypy --config-file libregrtest/mypy.ini`.
+
+[mypy]
+packages = libregrtest
+python_version = 3.11
+platform = linux
+pretty = True
+
+# Enable most stricter settings
+enable_error_code = ignore-without-code
+strict = True
+
+# Various stricter settings that we can't yet enable
+# Try to enable these in the following order:
+strict_optional = False
+disallow_any_generics = False
+disallow_incomplete_defs = False
+disallow_untyped_calls = False
+disallow_untyped_defs = False
+check_untyped_defs = False
+warn_return_any = False
+
+disable_error_code = return
+
+# Various internal modules that typeshed deliberately doesn't have stubs for:
+[mypy-_abc.*]
+ignore_missing_imports = True
+
+[mypy-_opcode.*]
+ignore_missing_imports = True
+
+[mypy-_overlapped.*]
+ignore_missing_imports = True
+
+[mypy-_testcapi.*]
+ignore_missing_imports = True
+
+[mypy-_testinternalcapi.*]
+ignore_missing_imports = True
+
+[mypy-test.*]
+ignore_missing_imports = True

Lib/test/libregrtest/refleak.py

@@ -1,6 +1,7 @@
 import sys
 import warnings
 from inspect import isabstract
+from typing import Any
 
 from test import support
 from test.support import os_helper
@@ -45,6 +46,7 @@ def runtest_refleak(test_name, test_func,
     fs = warnings.filters[:]
     ps = copyreg.dispatch_table.copy()
     pic = sys.path_importer_cache.copy()
+    zdc: dict[str, Any] | None
     try:
         import zipimport
     except ImportError:

Lib/test/libregrtest/results.py

@@ -111,7 +111,7 @@ def accumulate_result(self, result: TestResult, runtests: RunTests):
     def need_rerun(self):
         return bool(self.bad_results)
 
-    def prepare_rerun(self) -> (TestTuple, FilterDict):
+    def prepare_rerun(self) -> tuple[TestTuple, FilterDict]:
         tests: TestList = []
         match_tests_dict = {}
         for result in self.bad_results:

Lib/test/libregrtest/run_workers.py

@@ -21,7 +21,7 @@
 from .runtests import RunTests, JsonFile, JsonFileType
 from .single import PROGRESS_MIN_TIME
 from .utils import (
-    StrPath, StrJSON, TestName, MS_WINDOWS,
+    StrPath, TestName, MS_WINDOWS,
     format_duration, print_warning, count, plural)
 from .worker import create_worker_process, USE_PROCESS_GROUP
 
@@ -104,9 +104,9 @@ def __init__(self, worker_id: int, runner: "RunWorkers") -> None:
         self.output = runner.output
         self.timeout = runner.worker_timeout
         self.log = runner.log
-        self.test_name = None
-        self.start_time = None
-        self._popen = None
+        self.test_name: TestName | None = None
+        self.start_time: float | None = None
+        self._popen: subprocess.Popen[str] | None = None
         self._killed = False
         self._stopped = False
 
@@ -160,7 +160,7 @@ def stop(self) -> None:
         self._kill()
 
     def _run_process(self, runtests: RunTests, output_fd: int,
-                     tmp_dir: StrPath | None = None) -> int:
+                     tmp_dir: StrPath | None = None) -> int | None:
         popen = create_worker_process(runtests, output_fd, tmp_dir)
         self._popen = popen
         self._killed = False
@@ -260,7 +260,7 @@ def create_worker_runtests(self, test_name: TestName, json_file: JsonFile) -> Ru
             **kwargs)
 
     def run_tmp_files(self, worker_runtests: RunTests,
-                      stdout_fd: int) -> (int, list[StrPath]):
+                      stdout_fd: int) -> tuple[int | None, list[StrPath]]:
         # gh-93353: Check for leaked temporary files in the parent process,
         # since the deletion of temporary files can happen late during
         # Python finalization: too late for libregrtest.
@@ -297,13 +297,13 @@ def read_json(self, json_file: JsonFile, json_tmpfile: TextIO | None,
         try:
             if json_tmpfile is not None:
                 json_tmpfile.seek(0)
-                worker_json: StrJSON = json_tmpfile.read()
+                worker_json = json_tmpfile.read()
             elif json_file.file_type == JsonFileType.STDOUT:
                 stdout, _, worker_json = stdout.rpartition("\n")
                 stdout = stdout.rstrip()
             else:
                 with json_file.open(encoding='utf8') as json_fp:
-                    worker_json: StrJSON = json_fp.read()
+                    worker_json = json_fp.read()
         except Exception as exc:
             # gh-101634: Catch UnicodeDecodeError if stdout cannot be
             # decoded from encoding
@@ -414,8 +414,8 @@ def wait_stopped(self, start_time: float) -> None:
                 break
 
 
-def get_running(workers: list[WorkerThread]) -> list[str]:
-    running = []
+def get_running(workers: list[WorkerThread]) -> str | None:
+    running: list[str] = []
     for worker in workers:
         test_name = worker.test_name
         if not test_name:
@@ -431,7 +431,7 @@ def get_running(workers: list[WorkerThread]) -> list[str]:
 
 class RunWorkers:
     def __init__(self, num_workers: int, runtests: RunTests,
-                 logger: Logger, results: TestResult) -> None:
+                 logger: Logger, results: TestResults) -> None:
         self.num_workers = num_workers
         self.runtests = runtests
         self.log = logger.log
@@ -446,10 +446,10 @@ def __init__(self, num_workers: int, runtests: RunTests,
             # Rely on faulthandler to kill a worker process. This timouet is
             # when faulthandler fails to kill a worker process. Give a maximum
             # of 5 minutes to faulthandler to kill the worker.
-            self.worker_timeout = min(self.timeout * 1.5, self.timeout + 5 * 60)
+            self.worker_timeout: float | None = min(self.timeout * 1.5, self.timeout + 5 * 60)
         else:
             self.worker_timeout = None
-        self.workers = None
+        self.workers: list[WorkerThread] | None = None
 
         jobs = self.runtests.get_jobs()
         if jobs is not None:
@@ -529,7 +529,7 @@ def display_result(self, mp_result: MultiprocessResult) -> None:
                 text += f' -- {running}'
         self.display_progress(self.test_index, text)
 
-    def _process_result(self, item: QueueOutput) -> bool:
+    def _process_result(self, item: QueueOutput) -> TestResult:
         """Returns True if test runner must stop."""
         if item[0]:
             # Thread got an exception

Lib/test/libregrtest/runtests.py

@@ -88,8 +88,8 @@ class RunTests:
     use_junit: bool
     memory_limit: str | None
     gc_threshold: int | None
-    use_resources: tuple[str]
-    python_cmd: tuple[str] | None
+    use_resources: tuple[str, ...]
+    python_cmd: tuple[str, ...] | None
     randomize: bool
     random_seed: int | None
     json_file: JsonFile | None

Lib/test/libregrtest/setup.py

@@ -1,15 +1,12 @@
 import faulthandler
+import gc
 import os
 import random
 import signal
 import sys
 import unittest
 from test import support
 from test.support.os_helper import TESTFN_UNDECODABLE, FS_NONASCII
-try:
-    import gc
-except ImportError:
-    gc = None
 
 from .runtests import RunTests
 from .utils import (