diff --git a/nixos/lib/test-driver/default.nix b/nixos/lib/test-driver/default.nix index 72aa6b11bfe5..88fd70ee7f47 100644 --- a/nixos/lib/test-driver/default.nix +++ b/nixos/lib/test-driver/default.nix @@ -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" diff --git a/nixos/lib/test-driver/src/pyproject.toml b/nixos/lib/test-driver/src/pyproject.toml index fa4e6a2de127..c348bf1e4dfb 100644 --- a/nixos/lib/test-driver/src/pyproject.toml +++ b/nixos/lib/test-driver/src/pyproject.toml @@ -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 diff --git a/nixos/lib/test-driver/src/test_driver/__init__.py b/nixos/lib/test-driver/src/test_driver/__init__.py index 0f42f2842c77..ded14748b64a 100644 --- a/nixos/lib/test-driver/src/test_driver/__init__.py +++ b/nixos/lib/test-driver/src/test_driver/__init__.py @@ -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() diff --git a/nixos/lib/test-driver/src/test_driver/debug.py b/nixos/lib/test-driver/src/test_driver/debug.py index 7f783fe96de8..25423c35a3d8 100644 --- a/nixos/lib/test-driver/src/test_driver/debug.py +++ b/nixos/lib/test-driver/src/test_driver/debug.py @@ -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) diff --git a/nixos/lib/test-driver/src/test_driver/driver.py b/nixos/lib/test-driver/src/test_driver/driver.py index 9188e195cb35..abaa2cc71955 100644 --- a/nixos/lib/test-driver/src/test_driver/driver.py +++ b/nixos/lib/test-driver/src/test_driver/driver.py @@ -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 diff --git a/nixos/lib/test-driver/src/test_driver/logger.py b/nixos/lib/test-driver/src/test_driver/logger.py index a218d234fe3f..62d1a99d371e 100644 --- a/nixos/lib/test-driver/src/test_driver/logger.py +++ b/nixos/lib/test-driver/src/test_driver/logger.py @@ -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: diff --git a/nixos/lib/test-driver/src/test_driver/machine/__init__.py b/nixos/lib/test-driver/src/test_driver/machine/__init__.py index 365826dc22ff..7edb72b76145 100644 --- a/nixos/lib/test-driver/src/test_driver/machine/__init__.py +++ b/nixos/lib/test-driver/src/test_driver/machine/__init__.py @@ -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", diff --git a/nixos/lib/test-driver/src/test_driver/polling_condition.py b/nixos/lib/test-driver/src/test_driver/polling_condition.py index 1a8091cf4471..bebdeca29702 100644 --- a/nixos/lib/test-driver/src/test_driver/polling_condition.py +++ b/nixos/lib/test-driver/src/test_driver/polling_condition.py @@ -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()