Skip to content

PEL

faststream.redis.testing.PEL #

PEL()
Source code in faststream/redis/testing.py 
74
75
76
77
def __init__(self) -> None:
    self.entries: dict[PELKey, Entry] = {}
    self.put = MagicMock(wraps=self._put)
    self.remove = MagicMock(wraps=self._remove)

entries instance-attribute #

entries: dict[PELKey, Entry] = {}

put instance-attribute #

put = MagicMock(wraps=self._put)

remove instance-attribute #

remove = MagicMock(wraps=self._remove)

get_entry #

get_entry(correlation_id: PELKey) -> Entry | None
Source code in faststream/redis/testing.py 
90
91
def get_entry(self, correlation_id: PELKey) -> Entry | None:
    return self.entries.get(correlation_id)