Use `from __future__ import annotations`

Allows us to use more modern typing.
This commit is contained in:
Ran Benita 2024-04-03 09:53:40 +03:00
parent 059c1bcc8c
commit e7a971b167
6 changed files with 23 additions and 24 deletions

View File

@ -85,6 +85,7 @@ select = [
"W", # pycodestyle
"T10", # flake8-debugger
"PIE", # flake8-pie
"FA", # flake8-future-annotations
"PGH", # pygrep-hooks
"PLE", # pylint error
"PLW", # pylint warning

View File

@ -7,11 +7,12 @@ processes) otherwise changes to source code can crash
the controlling process which should best never happen.
"""
from __future__ import annotations
import os
from pathlib import Path
import sys
import time
from typing import Dict
from typing import Sequence
from _pytest._io import TerminalWriter
@ -45,7 +46,7 @@ def pytest_cmdline_main(config):
return 2 # looponfail only can get stop with ctrl-C anyway
def looponfail_main(config: "pytest.Config") -> None:
def looponfail_main(config: pytest.Config) -> None:
remotecontrol = RemoteControl(config)
config_roots = config.getini("looponfailroots")
if not config_roots:
@ -238,7 +239,7 @@ class WorkerFailSession:
class StatRecorder:
def __init__(self, rootdirlist: Sequence[Path]) -> None:
self.rootdirlist = rootdirlist
self.statcache: Dict[Path, os.stat_result] = {}
self.statcache: dict[Path, os.stat_result] = {}
self.check() # snapshot state
def fil(self, p: Path) -> bool:
@ -256,7 +257,7 @@ class StatRecorder:
def check(self, removepycfiles: bool = True) -> bool:
changed = False
newstat: Dict[Path, os.stat_result] = {}
newstat: dict[Path, os.stat_result] = {}
for rootdir in self.rootdirlist:
for path in visit_path(rootdir, filter=self.fil, recurse=self.rec):
oldstat = self.statcache.pop(path, None)

View File

@ -7,11 +7,7 @@ from pathlib import Path
import re
import sys
from typing import Any
from typing import List
from typing import Optional
from typing import Sequence
from typing import Set
from typing import Tuple
from typing import Union
import uuid
@ -63,7 +59,7 @@ class NodeManager:
self.specs.append(spec)
self.roots = self._getrsyncdirs()
self.rsyncoptions = self._getrsyncoptions()
self._rsynced_specs: Set[Tuple[Any, Any]] = set()
self._rsynced_specs: set[tuple[Any, Any]] = set()
def rsync_roots(self, gateway):
"""Rsync the set of roots to the node's gateway cwd."""
@ -92,7 +88,7 @@ class NodeManager:
def _getxspecs(self):
return [execnet.XSpec(x) for x in parse_spec_config(self.config)]
def _getrsyncdirs(self) -> List[Path]:
def _getrsyncdirs(self) -> list[Path]:
for spec in self.specs:
if not spec.popen or spec.chdir:
break
@ -177,7 +173,7 @@ class HostRSync(execnet.RSync):
self,
sourcedir: PathLike,
*,
ignores: Optional[Sequence[PathLike]] = None,
ignores: Sequence[PathLike] | None = None,
verbose: bool = True,
) -> None:
if ignores is None:
@ -204,7 +200,7 @@ class HostRSync(execnet.RSync):
print(f"{gateway.spec}:{remotepath} <= {path}")
def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
def make_reltoroot(roots: Sequence[Path], args: list[str]) -> list[str]:
# XXX introduce/use public API for splitting pytest args
splitcode = "::"
result = []
@ -219,7 +215,7 @@ def make_reltoroot(roots: Sequence[Path], args: List[str]) -> List[str]:
result.append(arg)
continue
for root in roots:
x: Optional[Path]
x: Path | None
try:
x = fspath.relative_to(root)
except ValueError:

View File

@ -1,9 +1,8 @@
from __future__ import annotations
import os
import re
import shutil
from typing import Dict
from typing import List
from typing import Tuple
import pytest
@ -1528,7 +1527,7 @@ class TestLocking:
result.assert_outcomes(passed=(48 if scope != "each" else 48 * 2))
def parse_tests_and_workers_from_output(lines: List[str]) -> List[Tuple[str, str, str]]:
def parse_tests_and_workers_from_output(lines: list[str]) -> list[tuple[str, str, str]]:
result = []
for line in lines:
# example match: "[gw0] PASSED test_a.py::test[7]"
@ -1550,9 +1549,9 @@ def parse_tests_and_workers_from_output(lines: List[str]) -> List[Tuple[str, str
def get_workers_and_test_count_by_prefix(
prefix: str, lines: List[str], expected_status: str = "PASSED"
) -> Dict[str, int]:
result: Dict[str, int] = {}
prefix: str, lines: list[str], expected_status: str = "PASSED"
) -> dict[str, int]:
result: dict[str, int] = {}
for worker, status, nodeid in parse_tests_and_workers_from_output(lines):
if expected_status == status and nodeid.startswith(prefix):
result[worker] = result.get(worker, 0) + 1

View File

@ -1,5 +1,6 @@
from __future__ import annotations
import shutil
from typing import List
import execnet
import pytest
@ -41,7 +42,7 @@ def specssh(request) -> str:
# configuration information for tests
def getgspecs(config) -> List[execnet.XSpec]:
def getgspecs(config) -> list[execnet.XSpec]:
return [execnet.XSpec(spec) for spec in config.getvalueorskip("gspecs")]

View File

@ -1,9 +1,10 @@
from __future__ import annotations
import pathlib
from pathlib import Path
import shutil
import tempfile
import textwrap
from typing import List
import unittest.mock
import pytest
@ -75,7 +76,7 @@ class TestStatRecorder:
# make check()'s visit() call return our just removed
# path as if we were in a race condition
dirname = str(tmp)
dirnames: List[str] = []
dirnames: list[str] = []
filenames = [str(p)]
with unittest.mock.patch(
"os.walk", return_value=[(dirname, dirnames, filenames)], autospec=True