jmri_nce.transport.loopback

Loopback byte-pair transport for NCE tests.

Two LoopbackPort halves share a pair of asyncio queues; bytes written into one appear on the other's read stream. The test side wires a "fake command station" listener that reads requests, computes a reply, and writes it back.

 1"""Loopback byte-pair transport for NCE tests.
 2
 3Two :class:`LoopbackPort` halves share a pair of asyncio queues; bytes
 4written into one appear on the other's read stream. The test side wires a
 5"fake command station" listener that reads requests, computes a reply,
 6and writes it back.
 7"""
 8from __future__ import annotations
 9
10import asyncio
11
12from jmri_nce.transport.port import BytePort
13
14
15class LoopbackPort(BytePort):
16    """One half of a connected pair."""
17
18    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
19        self._r = read_q
20        self._w = write_q
21        self._open = True
22
23    async def write(self, data: bytes) -> None:
24        if not self._open:
25            raise RuntimeError("port is closed")
26        for b in data:
27            await self._w.put(b)
28
29    async def read_exact(self, n: int) -> bytes:
30        out = bytearray()
31        for _ in range(n):
32            out.append(await self._r.get())
33        return bytes(out)
34
35    async def read_pending(self) -> bytes:
36        """Drain anything currently sitting in the receive queue."""
37        out = bytearray()
38        while not self._r.empty():
39            out.append(self._r.get_nowait())
40        return bytes(out)
41
42    async def aclose(self) -> None:
43        self._open = False
44
45    @property
46    def is_open(self) -> bool:
47        return self._open
48
49
50def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
51    """Two connected loopback ports — A writes to B's read stream and vice versa."""
52    a_to_b: asyncio.Queue[int] = asyncio.Queue()
53    b_to_a: asyncio.Queue[int] = asyncio.Queue()
54    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)
class LoopbackPort(jmri_nce.transport.port.BytePort): 
16class LoopbackPort(BytePort):
17    """One half of a connected pair."""
18
19    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
20        self._r = read_q
21        self._w = write_q
22        self._open = True
23
24    async def write(self, data: bytes) -> None:
25        if not self._open:
26            raise RuntimeError("port is closed")
27        for b in data:
28            await self._w.put(b)
29
30    async def read_exact(self, n: int) -> bytes:
31        out = bytearray()
32        for _ in range(n):
33            out.append(await self._r.get())
34        return bytes(out)
35
36    async def read_pending(self) -> bytes:
37        """Drain anything currently sitting in the receive queue."""
38        out = bytearray()
39        while not self._r.empty():
40            out.append(self._r.get_nowait())
41        return bytes(out)
42
43    async def aclose(self) -> None:
44        self._open = False
45
46    @property
47    def is_open(self) -> bool:
48        return self._open

One half of a connected pair.

LoopbackPort( read_q: asyncio.queues.Queue[int], write_q: asyncio.queues.Queue[int]) 
19    def __init__(self, read_q: asyncio.Queue[int], write_q: asyncio.Queue[int]) -> None:
20        self._r = read_q
21        self._w = write_q
22        self._open = True
async def write(self, data: bytes) -> None: 
24    async def write(self, data: bytes) -> None:
25        if not self._open:
26            raise RuntimeError("port is closed")
27        for b in data:
28            await self._w.put(b)
async def read_exact(self, n: int) -> bytes: 
30    async def read_exact(self, n: int) -> bytes:
31        out = bytearray()
32        for _ in range(n):
33            out.append(await self._r.get())
34        return bytes(out)
async def read_pending(self) -> bytes: 
36    async def read_pending(self) -> bytes:
37        """Drain anything currently sitting in the receive queue."""
38        out = bytearray()
39        while not self._r.empty():
40            out.append(self._r.get_nowait())
41        return bytes(out)

Drain anything currently sitting in the receive queue.

async def aclose(self) -> None: 
43    async def aclose(self) -> None:
44        self._open = False
is_open: bool 
46    @property
47    def is_open(self) -> bool:
48        return self._open
def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]: 
51def loopback_pair() -> tuple[LoopbackPort, LoopbackPort]:
52    """Two connected loopback ports — A writes to B's read stream and vice versa."""
53    a_to_b: asyncio.Queue[int] = asyncio.Queue()
54    b_to_a: asyncio.Queue[int] = asyncio.Queue()
55    return LoopbackPort(b_to_a, a_to_b), LoopbackPort(a_to_b, b_to_a)

Two connected loopback ports — A writes to B's read stream and vice versa.