Skip to content

Commit

Permalink
Adjust locking actor to new development
Browse files Browse the repository at this point in the history
  • Loading branch information
BenediktBurger committed Jun 19, 2024
1 parent 6aa25bb commit ff8b159
Show file tree
Hide file tree
Showing 2 changed files with 6 additions and 14 deletions.
12 changes: 2 additions & 10 deletions pyleco/actors/locking_actor.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,23 +27,19 @@

from zmq import Context

from ..core.message import Message
from .actor import Actor


Device = TypeVar("Device")


class AccessError(BaseException):
# TODO name TBD
class AccessDeniedError(BaseException):
pass


class LockingActor(Actor, Generic[Device]):
"""An Actor which allows to lock the device or parts of it."""

current_message: Message

def __init__(
self,
name: str,
Expand Down Expand Up @@ -84,10 +80,6 @@ def force_unlock(self, resource: Optional[str] = None) -> None:
self._locks.pop(resource, None)

# modified methods for device access
def process_json_message(self, message: Message) -> Message:
self.current_message = message
return super().process_json_message(message=message)

def get_parameters(self, parameters: Union[list[str], tuple[str, ...]]) -> dict[str, Any]:
# `parameters` should be `Iterable[str]`, however, openrpc does not like that.
for parameter in parameters:
Expand Down Expand Up @@ -120,4 +112,4 @@ def check_access_rights(self, resource: Optional[str]) -> bool:

def _check_access_rights_raising(self, resource: str) -> None:
if self.check_access_rights(resource=resource) is False:
raise AccessError(f"Resource '{resource}' is locked by someone else.")
raise AccessDeniedError(f"Resource '{resource}' is locked by someone else.")
8 changes: 4 additions & 4 deletions tests/actors/test_locking_actor.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@
import pytest

from pyleco.core.message import Message
from pyleco.actors.locking_actor import LockingActor, AccessError
from pyleco.actors.locking_actor import LockingActor, AccessDeniedError
from pyleco.core.leco_protocols import LockingActorProtocol


Expand Down Expand Up @@ -264,7 +264,7 @@ def test_get_parameters_successfully(locked_actor: LockingActor):

def test_get_parameters_unsuccessfully(locked_actor: LockingActor):
locked_actor.current_message = Message("rec", "requester")
with pytest.raises(AccessError, match="'l_prop'"):
with pytest.raises(AccessDeniedError, match="'l_prop'"):
locked_actor.get_parameters(["o_prop", "l_prop"])


Expand All @@ -276,7 +276,7 @@ def test_set_parameters_successfully(locked_actor: LockingActor):

def test_set_parameters_unsuccessfully(locked_actor: LockingActor):
locked_actor.current_message = Message("rec", "requester")
with pytest.raises(AccessError, match="'l_prop'"):
with pytest.raises(AccessDeniedError, match="'l_prop'"):
locked_actor.set_parameters({"o_prop": 5, "l_prop": 6})


Expand All @@ -288,5 +288,5 @@ def test_call_action_successfully(locked_actor: LockingActor):

def test_call_action_unsuccessfully(locked_actor: LockingActor):
locked_actor.current_message = Message("rec", "requester")
with pytest.raises(AccessError, match="'l_method'"):
with pytest.raises(AccessDeniedError, match="'l_method'"):
locked_actor.call_action("l_method", [5])

0 comments on commit ff8b159

Please sign in to comment.