gh-109413: libregrtest: Add and improve type annotations

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 (

Lib/test/libregrtest/single.py

@@ -51,6 +51,8 @@ def regrtest_runner(result: TestResult, test_func, runtests: RunTests) -> None:
  if refleak:
  result.state = State.REFLEAK
 
+ stats: TestStats | None
+
  match test_result:
  case TestStats():
  stats = test_result

Lib/test/libregrtest/utils.py

@@ -10,6 +10,7 @@
 import sysconfig
 import tempfile
 import textwrap
+from collections.abc import Callable
 
 from test import support
 from test.support import os_helper
@@ -67,7 +68,7 @@ def format_duration(seconds):
  return ' '.join(parts)
 
 
-def strip_py_suffix(names: list[str]):
+def strip_py_suffix(names: list[str] | None) -> None:
  if not names:
  return
  for idx, name in enumerate(names):
@@ -441,6 +442,7 @@ def remove_testfn(test_name: TestName, verbose: int) -> None:
  if not os.path.exists(name):
  return
 
+ nuker: Callable[[str], None]
  if os.path.isdir(name):
  import shutil
  kind, nuker = "directory", shutil.rmtree

Lib/test/libregrtest/worker.py

@@ -1,7 +1,7 @@
 import subprocess
 import sys
 import os
-from typing import NoReturn
+from typing import Any, NoReturn
 
 from test import support
 from test.support import os_helper
@@ -45,7 +45,7 @@ def create_worker_process(runtests: RunTests, output_fd: int,
  # Running the child from the same working directory as regrtest's original
  # invocation ensures that TEMPDIR for the child is the same when
  # sysconfig.is_python_build() is true. See issue 15300.
- kwargs = dict(
+ kwargs: dict[str, Any] = dict(
  env=env,
  stdout=output_fd,
  # bpo-45410: Write stderr into stdout to keep messages order