jmri_ncemonitor.transport.loopback

In-memory loopback for tests — same shape as jmri-nce.

 1"""In-memory loopback for tests — same shape as jmri-nce."""
 2from __future__ import annotations
 3
 4import asyncio
 5
 6from jmri_ncemonitor.transport.port import BytePort
 7
 8
 9class LoopbackPort(BytePort):
10    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
11        self._r = read_q
12        self._w = write_q
13        self._open = True
14
15    async def write(self, data: bytes) -> None:
16        if not self._open:
17            raise RuntimeError("port is closed")
18        for b in data:
19            await self._w.put(b)
20
21    async def read_byte(self) -> int:
22        return await self._r.get()
23
24    async def aclose(self) -> None:
25        self._open = False
26
27    @property
28    def is_open(self) -> bool:
29        return self._open
30
31
32def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
33    a_to_b: asyncio.Queue[int] = asyncio.Queue()
34    b_to_a: asyncio.Queue[int] = asyncio.Queue()
35    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)
class LoopbackPort(jmri_ncemonitor.transport.port.BytePort): 
10class LoopbackPort(BytePort):
11    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
12        self._r = read_q
13        self._w = write_q
14        self._open = True
15
16    async def write(self, data: bytes) -> None:
17        if not self._open:
18            raise RuntimeError("port is closed")
19        for b in data:
20            await self._w.put(b)
21
22    async def read_byte(self) -> int:
23        return await self._r.get()
24
25    async def aclose(self) -> None:
26        self._open = False
27
28    @property
29    def is_open(self) -> bool:
30        return self._open

Helper class that provides a standard way to create an ABC using inheritance.

LoopbackPort( read_q: asyncio.queues.Queue[int], write_q: asyncio.queues.Queue[int]) 
11    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
12        self._r = read_q
13        self._w = write_q
14        self._open = True
async def write(self, data: bytes) -> None: 
16    async def write(self, data: bytes) -> None:
17        if not self._open:
18            raise RuntimeError("port is closed")
19        for b in data:
20            await self._w.put(b)
async def read_byte(self) -> int: 
22    async def read_byte(self) -> int:
23        return await self._r.get()
async def aclose(self) -> None: 
25    async def aclose(self) -> None:
26        self._open = False
is_open: bool 
28    @property
29    def is_open(self) -> bool:
30        return self._open
def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]: 
33def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
34    a_to_b: asyncio.Queue[int] = asyncio.Queue()
35    b_to_a: asyncio.Queue[int] = asyncio.Queue()
36    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)