From 427d66fb77315e4e6098a4a1cda59582e6afa654 Mon Sep 17 00:00:00 2001 From: Rainer Poisel Date: Tue, 28 Jul 2026 23:28:18 +0200 Subject: [PATCH] driver: add initial type hints for common driver code Add `ty` as a development dependency and configure it to check one initial driver module. Annotate `labgrid.driver.common` while leaving existing attrs-based exception behavior unchanged. Signed-off-by: Rainer Poisel --- labgrid/driver/common.py | 33 ++++++++++++++++----------- pyproject.toml | 7 ++++++ tests/test_driver_common.py | 45 +++++++++++++++++++++++++++++++++++++ 3 files changed, 72 insertions(+), 13 deletions(-) create mode 100644 tests/test_driver_common.py diff --git a/labgrid/driver/common.py b/labgrid/driver/common.py index 2adf3f893..3ac33f319 100644 --- a/labgrid/driver/common.py +++ b/labgrid/driver/common.py @@ -1,7 +1,11 @@ import logging import subprocess +from typing import cast + import attr +from labgrid.resource.common import Resource + from ..binding import BindingError, BindingMixin from .exception import ExecutionError @@ -21,17 +25,17 @@ class Driver(BindingMixin): - deactivate """ - def __attrs_post_init__(self): + def __attrs_post_init__(self) -> None: super().__attrs_post_init__() if self.target is None: - raise BindingError("Drivers can only be created on a valid target") + raise BindingError("Drivers can only be created on a valid target") # ty: ignore[too-many-positional-arguments] logger_name = f"{self.__class__.__name__}({self.target.name})" if self.name: logger_name += f":{self.name}" self.logger = logging.getLogger(logger_name) - def get_priority(self, protocol): + def get_priority(self, protocol) -> int: """Retrieve the priority for a given protocol Arguments: @@ -41,17 +45,17 @@ def get_priority(self, protocol): Int: value of the priority if it is found, 0 otherwise. """ for cls in self.__class__.__mro__: - prios = getattr(cls, 'priorities', {}) + prios = getattr(cls, "priorities", {}) # we found a matching parent priorities attribute with the matching protocol if prios and protocol in prios: - return prios.get(protocol) + return cast(int, prios[protocol]) # If we find the parent protocol, set the priority to 0 if cls.__name__ == protocol.__name__: return 0 return 0 - def get_export_name(self): + def get_export_name(self) -> str: """Get the name to be used for exported variables. Falls back to the class name if the driver has no name. @@ -60,12 +64,12 @@ def get_export_name(self): return self.name return self.__class__.__name__ - def get_export_vars(self): + def get_export_vars(self) -> dict[str, str]: """Get a dictionary of variables to be exported.""" return {} @property - def skip_deactivate_on_export(self): + def skip_deactivate_on_export(self) -> bool: """Drivers are deactivated on export by default. If the driver can handle external accesses even while active, it can @@ -73,16 +77,19 @@ def skip_deactivate_on_export(self): """ return False - def get_bound_resources(self): + def get_bound_resources(self) -> set[Resource]: """Return the bound resources for a driver This recursively calls all suppliers and combines the sets of returned resources. """ - res = set() + res: set[Resource] = set() for supplier in self.suppliers: res |= supplier.get_bound_resources() return res -def check_file(filename, *, command_prefix=[]): - if subprocess.call(command_prefix + ['test', '-r', filename]) != 0: - raise ExecutionError(f"File {filename} is not readable") + +def check_file(filename, *, command_prefix=None) -> None: + if command_prefix is None: + command_prefix: list[str] = [] + if subprocess.call(command_prefix + ["test", "-r", filename]) != 0: + raise ExecutionError(f"File {filename} is not readable") # ty: ignore[too-many-positional-arguments] diff --git a/pyproject.toml b/pyproject.toml index 499d1f087..c8d22cd7d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -91,6 +91,7 @@ dev = [ "pytest-mock>=3.6.1", "pylint>=3.0.0", "ruff>=0.5.7", + "ty>=0.0.64", "pystuck", "requests[socks]>=2.26.0", @@ -138,6 +139,11 @@ testpaths = [ ] addopts = "-p no:labgrid" +[tool.ty.src] +include = [ + "labgrid/driver/common.py", +] + [tool.pylint.imports] ignored-modules = ["gi"] @@ -224,6 +230,7 @@ include = [ "examples/**/*.py", "labgrid/driver/httpvideodriver.py", "labgrid/driver/manualswitchdriver.py", + "labgrid/driver/common.py", "labgrid/driver/power/gude8031.py", "labgrid/driver/power/pe6216.py", "labgrid/driver/power/poe_netgear_plus.py", diff --git a/tests/test_driver_common.py b/tests/test_driver_common.py new file mode 100644 index 000000000..4eec50145 --- /dev/null +++ b/tests/test_driver_common.py @@ -0,0 +1,45 @@ +from unittest.mock import Mock + +import pytest + +from labgrid import Target +from labgrid.binding import BindingError +from labgrid.driver.common import Driver, check_file +from labgrid.driver.exception import ExecutionError + + +def test_driver_requires_target() -> None: + with pytest.raises(BindingError): + Driver(None, None) + + +def test_driver_get_priority_returns_zero_for_unknown_protocol(target: Target) -> None: + class Protocol: + pass + + driver = Driver(target, None) + + assert driver.get_priority(Protocol) == 0 + + +def test_driver_get_export_vars_returns_empty_dict(target: Target) -> None: + driver = Driver(target, None) + + assert driver.get_export_vars() == {} + + +def test_check_file_uses_default_command_prefix(monkeypatch: pytest.MonkeyPatch) -> None: + call = Mock(return_value=0) + monkeypatch.setattr("labgrid.driver.common.subprocess.call", call) + + check_file("/tmp/file") + + call.assert_called_once_with(["test", "-r", "/tmp/file"]) + + +def test_check_file_raises_execution_error(monkeypatch: pytest.MonkeyPatch) -> None: + call = Mock(return_value=1) + monkeypatch.setattr("labgrid.driver.common.subprocess.call", call) + + with pytest.raises(ExecutionError, match="File /tmp/file is not readable"): + check_file("/tmp/file")