gador/nixpkgs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

nixos-test-driver: Use ty instead of mypy for test driver package

Browse Source
This commit is contained in:
Jacek Galowicz
2026-04-14 17:37:10 +01:00
parent 1d16486166
commit 85d58871ae
8 changed files with 47 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nixos/lib/test-driver/default.nix
+5 -5
View File
@@ -7,11 +7,11 @@
imagemagick_light,
ipython,
junit-xml,
mypy,
ptpython,
python,
ruff,
remote-pdb,
ruff,
ty,
netpbm,
nixosTests,
@@ -72,13 +72,13 @@ buildPythonApplication {
doCheck = true;
nativeCheckInputs = [
mypy
ruff
ty
];
checkPhase = ''
echo -e "\x1b[32m## run mypy\x1b[0m"
mypy test_driver extract-docstrings.py
echo -e "\x1b[32m## run ty\x1b[0m"
ty check --error-on-warning test_driver extract-docstrings.py
echo -e "\x1b[32m## run ruff check\x1b[0m"
ruff check .
echo -e "\x1b[32m## run ruff format\x1b[0m"
nixos/lib/test-driver/src/pyproject.toml
+1 -20
View File
@@ -17,27 +17,8 @@ find = {}
test_driver = ["py.typed"]
[tool.ruff]
target-version = "py312"
target-version = "py313"
line-length = 88
lint.select = ["E", "F", "I", "U", "N"]
lint.ignore = ["E501", "N818"]
# xxx: we can import https://pypi.org/project/types-colorama/ here
[[tool.mypy.overrides]]
module = "colorama.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "ptpython.*"
ignore_missing_imports = true
[[tool.mypy.overrides]]
module = "junit_xml.*"
ignore_missing_imports = true
[tool.mypy]
warn_redundant_casts = true
disallow_untyped_calls = true
disallow_untyped_defs = true
no_implicit_optional = true
nixos/lib/test-driver/src/test_driver/__init__.py
+3 -3
View File
@@ -22,7 +22,7 @@ class EnvDefault(argparse.Action):
environment variable as the flags default value.
"""
def __init__(self, envvar, required=False, default=None, nargs=None, **kwargs): # type: ignore
def __init__(self, envvar, required=False, default=None, nargs=None, **kwargs):
if not default and envvar:
if envvar in os.environ:
if nargs is not None and (nargs.isdigit() or nargs in ["*", "+"]):
@@ -36,7 +36,7 @@ class EnvDefault(argparse.Action):
required = False
super().__init__(default=default, required=required, nargs=nargs, **kwargs)
def __call__(self, parser, namespace, values, option_string=None): # type: ignore
def __call__(self, parser, namespace, values, option_string=None):
setattr(namespace, self.dest, values)
@@ -206,7 +206,7 @@ def main() -> None:
ptpython.ipython.embed(
user_ns=driver.test_symbols(),
history_filename=history_path,
) # type:ignore
)
else:
tic = time.time()
driver.run_tests()
nixos/lib/test-driver/src/test_driver/debug.py
+3 -3
View File
@@ -6,7 +6,7 @@ import subprocess
import sys
from abc import ABC, abstractmethod
from remote_pdb import RemotePdb # type:ignore
from remote_pdb import RemotePdb
from test_driver.logger import AbstractLogger
@@ -42,12 +42,12 @@ class Debug(DebugAbstract):
self.logger.log_test_error(
f"Breakpoint reached, run 'sudo {self.attach} {pattern}'"
)
os.environ["bashInteractive"] = shutil.which("bash") # type:ignore
os.environ["bashInteractive"] = shutil.which("bash") # ty: ignore[invalid-assignment]
if os.fork() == 0:
subprocess.run(["sleep", pattern])
else:
# RemotePdb writes log messages to both stderr AND the logger,
# which is the same here. Hence, disabling the remote_pdb logger
# to avoid duplicate messages in the build log.
logging.root.manager.loggerDict["remote_pdb"].disabled = True # type:ignore
logging.root.manager.loggerDict["remote_pdb"].disabled = True # ty: ignore[invalid-assignment]
RemotePdb(host=host, port=port).set_trace(sys._getframe().f_back)
nixos/lib/test-driver/src/test_driver/driver.py
+1 -1
View File
@@ -418,7 +418,7 @@ class Driver:
def __enter__(self) -> None:
driver.polling_conditions.append(self.condition)
def __exit__(self, a, b, c) -> None: # type: ignore
def __exit__(self, a, b, c) -> None:
res = driver.polling_conditions.pop()
assert res is self.condition
nixos/lib/test-driver/src/test_driver/logger.py
+25 -25
View File
@@ -1,5 +1,4 @@
import atexit
import codecs
import os
import sys
import time
@@ -33,19 +32,19 @@ class AbstractLogger(ABC):
pass
@abstractmethod
def info(self, *args, **kwargs) -> None: # type: ignore
def info(self, *args, **kwargs) -> None:
pass
@abstractmethod
def warning(self, *args, **kwargs) -> None: # type: ignore
def warning(self, *args, **kwargs) -> None:
pass
@abstractmethod
def error(self, *args, **kwargs) -> None: # type: ignore
def error(self, *args, **kwargs) -> None:
pass
@abstractmethod
def log_test_error(self, *args, **kwargs) -> None: # type:ignore
def log_test_error(self, *args, **kwargs) -> None:
pass
@abstractmethod
@@ -91,17 +90,17 @@ class JunitXMLLogger(AbstractLogger):
self.log(message)
yield
def info(self, *args, **kwargs) -> None: # type: ignore
def info(self, *args, **kwargs) -> None:
self.tests[self.currentSubtest].stdout += args[0] + os.linesep
def warning(self, *args, **kwargs) -> None: # type: ignore
def warning(self, *args, **kwargs) -> None:
self.tests[self.currentSubtest].stdout += args[0] + os.linesep
def error(self, *args, **kwargs) -> None: # type: ignore
def error(self, *args, **kwargs) -> None:
self.tests[self.currentSubtest].stderr += args[0] + os.linesep
self.tests[self.currentSubtest].failure = True
def log_test_error(self, *args, **kwargs) -> None: # type: ignore
def log_test_error(self, *args, **kwargs) -> None:
self.error(*args, **kwargs)
def log_serial(self, message: str, machine: str) -> None:
@@ -155,19 +154,19 @@ class CompositeLogger(AbstractLogger):
stack.enter_context(logger.nested(message, attributes))
yield
def info(self, *args, **kwargs) -> None: # type: ignore
def info(self, *args, **kwargs) -> None:
for logger in self.logger_list:
logger.info(*args, **kwargs)
def warning(self, *args, **kwargs) -> None: # type: ignore
def warning(self, *args, **kwargs) -> None:
for logger in self.logger_list:
logger.warning(*args, **kwargs)
def log_test_error(self, *args, **kwargs) -> None: # type: ignore
def log_test_error(self, *args, **kwargs) -> None:
for logger in self.logger_list:
logger.log_test_error(*args, **kwargs)
def error(self, *args, **kwargs) -> None: # type: ignore
def error(self, *args, **kwargs) -> None:
for logger in self.logger_list:
logger.error(*args, **kwargs)
sys.exit(1)
@@ -206,7 +205,8 @@ class TerminalLogger(AbstractLogger):
def nested(self, message: str, attributes: dict[str, str] = {}) -> Iterator[None]:
self._eprint(
self.maybe_prefix(
Style.BRIGHT + Fore.GREEN + message + Style.RESET_ALL, attributes
Style.BRIGHT + Fore.GREEN + message + Style.RESET_ALL, # ty: ignore[unsupported-operator]
attributes,
)
)
@@ -215,13 +215,13 @@ class TerminalLogger(AbstractLogger):
toc = time.time()
self.log(f"(finished: {message}, in {toc - tic:.2f} seconds)", attributes)
def info(self, *args, **kwargs) -> None: # type: ignore
def info(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def warning(self, *args, **kwargs) -> None: # type: ignore
def warning(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def error(self, *args, **kwargs) -> None: # type: ignore
def error(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def print_serial_logs(self, enable: bool) -> None:
@@ -231,17 +231,17 @@ class TerminalLogger(AbstractLogger):
if not self._print_serial_logs:
return
self._eprint(Style.DIM + f"{machine} # {message}" + Style.RESET_ALL)
self._eprint(Style.DIM + f"{machine} # {message}" + Style.RESET_ALL) # ty: ignore[unsupported-operator]
def log_test_error(self, *args, **kwargs) -> None: # type: ignore
prefix = Fore.RED + "!!! " + Style.RESET_ALL
def log_test_error(self, *args, **kwargs) -> None:
prefix = Fore.RED + "!!! " + Style.RESET_ALL # ty: ignore[unsupported-operator]
# NOTE: using `warning` instead of `error` to ensure it does not exit after printing the first log
self.warning(f"{prefix}{args[0]}", *args[1:], **kwargs)
class XMLLogger(AbstractLogger):
def __init__(self, outfile: str) -> None:
self.logfile_handle = codecs.open(outfile, "wb")
self.logfile_handle = open(outfile, "wb")
self.xml = XMLGenerator(self.logfile_handle, encoding="utf-8")
self.queue: Queue[dict[str, str]] = Queue()
@@ -268,16 +268,16 @@ class XMLLogger(AbstractLogger):
self.xml.characters(message)
self.xml.endElement("line")
def info(self, *args, **kwargs) -> None: # type: ignore
def info(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def warning(self, *args, **kwargs) -> None: # type: ignore
def warning(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def error(self, *args, **kwargs) -> None: # type: ignore
def error(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def log_test_error(self, *args, **kwargs) -> None: # type: ignore
def log_test_error(self, *args, **kwargs) -> None:
self.log(*args, **kwargs)
def log(self, message: str, attributes: dict[str, str] = {}) -> None:
nixos/lib/test-driver/src/test_driver/machine/__init__.py
+5 -3
View File
@@ -24,9 +24,11 @@ from typing import Any
from test_driver.errors import MachineError, RequestedAssertionFailed
from test_driver.logger import AbstractLogger
from .ocr import perform_ocr_on_screenshot, perform_ocr_variants_on_screenshot
from .qmp import QMPSession
from test_driver.machine.ocr import (
perform_ocr_on_screenshot,
perform_ocr_variants_on_screenshot,
)
from test_driver.machine.qmp import QMPSession
CHAR_TO_KEY = {
"A": "shift-a",
nixos/lib/test-driver/src/test_driver/polling_condition.py
+4 -4
View File
@@ -25,7 +25,7 @@ class PollingCondition:
seconds_interval: float = 2.0,
description: str | None = None,
):
self.condition = condition # type: ignore
self.condition = condition # ty: ignore[invalid-assignment]
self.seconds_interval = seconds_interval
self.logger = logger
@@ -33,7 +33,7 @@ class PollingCondition:
if condition.__doc__:
self.description = condition.__doc__
else:
self.description = condition.__name__
self.description = condition.__name__ # ty: ignore[unresolved-attribute]
else:
self.description = str(description)
@@ -54,7 +54,7 @@ class PollingCondition:
self.logger.info(last_message)
try:
res = self.condition() # type: ignore
res = self.condition()
except Exception:
res = False
res = res is None or res
@@ -89,7 +89,7 @@ class PollingCondition:
def __enter__(self) -> None:
self.entry_count += 1
def __exit__(self, exc_type, exc_value, traceback) -> None: # type: ignore
def __exit__(self, exc_type, exc_value, traceback) -> None:
assert self.entered
self.entry_count -= 1
self.last_called = time.monotonic()