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
}
-def findtestdir(path=None):
+def findtestdir(path: StrPath | None = None) -> StrPath:
return path or os.path.dirname(os.path.dirname(__file__)) or os.curdir
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
# 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
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
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,
--- /dev/null
+# 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
import sys
import warnings
from inspect import isabstract
+from typing import Any
from test import support
from test.support import os_helper
fs = warnings.filters[:]
ps = copyreg.dispatch_table.copy()
pic = sys.path_importer_cache.copy()
+ zdc: dict[str, Any] | None
try:
import zipimport
except ImportError:
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:
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
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
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
**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.
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
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:
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
# 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:
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
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
import faulthandler
+import gc
import os
import random
import signal
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 (
if refleak:
result.state = State.REFLEAK
+ stats: TestStats | None
+
match test_result:
case TestStats():
stats = test_result
import sysconfig
import tempfile
import textwrap
+from collections.abc import Callable
from test import support
from test.support import os_helper
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):
if not os.path.exists(name):
return
+ nuker: Callable[[str], None]
if os.path.isdir(name):
import shutil
kind, nuker = "directory", shutil.rmtree
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
# 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