-
-
Notifications
You must be signed in to change notification settings - Fork 274
driver: add initial type hints for common driver code #1946
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Changes from 1 commit
427d66f
3829631
a5d7218
636b5c2
94e55ff
b513092
0cd4874
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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] | ||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. So we need to mark each instantiation of an attr.s class like that? I don't think that scales.. Doesn't mypy work better with that?
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. AFAIR
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. In the separate PR that will add At the moment, |
||
|
|
||
| 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,29 +64,32 @@ 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 | ||
| return True here. | ||
| """ | ||
| 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: | ||
|
rpoisel marked this conversation as resolved.
Outdated
|
||
| if command_prefix is None: | ||
| command_prefix: list[str] = [] | ||
| if subprocess.call(command_prefix + ["test", "-r", filename]) != 0: | ||
|
rpoisel marked this conversation as resolved.
Outdated
|
||
| raise ExecutionError(f"File {filename} is not readable") # ty: ignore[too-many-positional-arguments] | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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") |
Uh oh!
There was an error while loading. Please reload this page.