Dummy
The dummy interface is the minimum possible interface implementation. It does not actually send or receive any data, but it doesn't crash. It is useful for certain tests where interfaces must be instantiated but don't need to process any data.
The dummy interface has no configuration options.
pax25.interfaces.dummy.DummyInterface
Dummy interface for testing. Allows you to check the precise frames received and
send exact frames as needed.
Source code in pax25/interfaces/dummy.py
| class DummyInterface(Interface[DummySettings]):
"""
Dummy interface for testing. Allows you to check the precise frames received and
send exact frames as needed.
"""
type = "Dummy"
def __init__(self, name: str, settings: DummySettings, station: "Station"):
"""
Just stash the args but don't do anything with them.
"""
self.name = name
self._settings = settings
self.station = station
self.sent_frames: Queue[Frame] = Queue()
self.queue: Queue[Frame] = Queue()
self._loop: Task[None] | None = None
@property
def listening(self) -> bool:
"""
Returns a bool indicating whether the interface is listening.
"""
if not self._loop:
return False
return not self._loop.done()
@property
def sudo(self) -> bool:
"""
Returns if connectsion on this interface should be considered privileged.
"""
return self._settings.get("sudo", False)
async def reload_settings(self, settings: DummySettings) -> None:
"""
Reloads the interface with revised settings.
"""
self._settings = settings
await self.shutdown()
self.start()
@property
def gateway(self) -> bool:
"""
Returns true if we can perform outbound connections via this interface.
"""
return self._settings.get("gateway", False)
def send_frame(self, frame: Frame) -> None:
"""
Dummy send frame function.
"""
self.sent_frames.put_nowait(frame)
def start(self) -> None:
"""
Starts the read loop.
"""
self.queue = Queue()
self._loop = asyncio.ensure_future(self.read_loop())
async def read_loop(self) -> None:
"""
Dummy read loop function. Probably won't be an issue if it just returns
immediately.
"""
while True:
frame = await self.queue.get()
self.station.frame_router.process_frame(self, frame)
self.queue.task_done()
async def shutdown(self) -> None:
"""
Dummy shut down function.
"""
self.queue.shutdown()
if self._loop:
await cancel(self._loop)
self._loop = None
|
gateway: bool
property
Returns true if we can perform outbound connections via this interface.
listening: bool
property
Returns a bool indicating whether the interface is listening.
sudo: bool
property
Returns if connectsion on this interface should be considered privileged.
__init__(name: str, settings: DummySettings, station: Station)
Just stash the args but don't do anything with them.
Source code in pax25/interfaces/dummy.py
| def __init__(self, name: str, settings: DummySettings, station: "Station"):
"""
Just stash the args but don't do anything with them.
"""
self.name = name
self._settings = settings
self.station = station
self.sent_frames: Queue[Frame] = Queue()
self.queue: Queue[Frame] = Queue()
self._loop: Task[None] | None = None
|
read_loop() -> None
async
Dummy read loop function. Probably won't be an issue if it just returns
immediately.
Source code in pax25/interfaces/dummy.py
| async def read_loop(self) -> None:
"""
Dummy read loop function. Probably won't be an issue if it just returns
immediately.
"""
while True:
frame = await self.queue.get()
self.station.frame_router.process_frame(self, frame)
self.queue.task_done()
|
reload_settings(settings: DummySettings) -> None
async
Reloads the interface with revised settings.
Source code in pax25/interfaces/dummy.py
| async def reload_settings(self, settings: DummySettings) -> None:
"""
Reloads the interface with revised settings.
"""
self._settings = settings
await self.shutdown()
self.start()
|
send_frame(frame: Frame) -> None
Dummy send frame function.
Source code in pax25/interfaces/dummy.py
| def send_frame(self, frame: Frame) -> None:
"""
Dummy send frame function.
"""
self.sent_frames.put_nowait(frame)
|
shutdown() -> None
async
Dummy shut down function.
Source code in pax25/interfaces/dummy.py
| async def shutdown(self) -> None:
"""
Dummy shut down function.
"""
self.queue.shutdown()
if self._loop:
await cancel(self._loop)
self._loop = None
|
start() -> None
Starts the read loop.
Source code in pax25/interfaces/dummy.py
| def start(self) -> None:
"""
Starts the read loop.
"""
self.queue = Queue()
self._loop = asyncio.ensure_future(self.read_loop())
|