You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
195 lines
7.8 KiB
195 lines
7.8 KiB
# ADN DMR Peer Server - tests plugin bus event interest and batching
|
|
#
|
|
# Copyright (C) 2026 Rodrigo Pérez, CE5RPY <ce5rpy@qmd.cl>
|
|
#
|
|
###############################################################################
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program; if not, write to the Free Software Foundation,
|
|
# Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
|
|
###############################################################################
|
|
|
|
"""Plugins pay only for the events they take; deferred events cost one call_later per tick."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from adn_server.application.plugins.application import voice_bridge as vb
|
|
from adn_server.application.plugins.application.bus import PluginBus
|
|
from adn_server.application.plugins.application.data_bridge import DataPluginBridge
|
|
from adn_server.application.plugins.application.voice_bridge import VoicePluginBridge
|
|
from adn_server.application.plugins.domain.events import (
|
|
UnitDataEnd,
|
|
UnitDataFrame,
|
|
UnitDataStart,
|
|
VoiceCallEnd,
|
|
VoiceCallFrame,
|
|
VoiceCallStart,
|
|
)
|
|
from adn_server.domain import HBPF_DATA_SYNC, HBPF_VOICE
|
|
|
|
|
|
class _Plugin:
|
|
def __init__(self, name: str, events=None) -> None:
|
|
self.name = name
|
|
if events is not None:
|
|
self.events = events
|
|
self.got: list[object] = []
|
|
|
|
def on_event(self, event: object) -> None:
|
|
self.got.append(event)
|
|
|
|
def on_shutdown(self) -> None:
|
|
pass
|
|
|
|
|
|
CONFIG = {"GLOBAL": {"SERVER_ID": 2131}, "SYSTEMS": {"SYSTEM": {"MODE": "MASTER"}}}
|
|
|
|
|
|
def _voice_frame(bridge: VoicePluginBridge, stream: int = 7) -> None:
|
|
bridge.notify_group_after_forward(
|
|
system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\0\0\xd5", slot=2,
|
|
frame_type=HBPF_VOICE, dtype_vseq=1, stream_id=stream.to_bytes(4, "big"), data=b"DMRD" + bytes(51),
|
|
pkt_time=1.0, source_is_obp=False, obp_hops=b"", obp_source_server=None, obp_ber=b"\0",
|
|
obp_rssi=b"\0", obp_source_rptr=b"\0\0\0\0", synthetic_announcement=False, forwarded=[],
|
|
)
|
|
|
|
|
|
def _data_frame(bridge: DataPluginBridge, dtype: int = 7) -> None:
|
|
bridge.notify_after_forward(
|
|
route="unit", system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\x0d\xbb\xa7",
|
|
seq=1, slot=2, frame_type=HBPF_DATA_SYNC, dtype_vseq=dtype, stream_id=b"\0\0\0\x09",
|
|
data=b"DMRD" + bytes(51), pkt_time=1.0, source_is_obp=False, obp_hops=b"", obp_source_server=None,
|
|
obp_ber=b"\0", obp_rssi=b"\0", obp_source_rptr=b"\0\0\0\0", forwarded=[],
|
|
)
|
|
|
|
|
|
def test_a_plugin_without_events_still_gets_everything() -> None:
|
|
bus = PluginBus()
|
|
legacy = _Plugin("legacy")
|
|
bus.register_plugin(legacy)
|
|
assert bus.wants(VoiceCallFrame) and bus.wants(UnitDataEnd)
|
|
_voice_frame(VoicePluginBridge(bus, CONFIG))
|
|
assert [type(e) for e in legacy.got] == [VoiceCallFrame]
|
|
|
|
|
|
def test_declared_events_are_the_only_ones_delivered() -> None:
|
|
bus = PluginBus()
|
|
data = _Plugin("d-aprs", events=(UnitDataFrame,))
|
|
voice = _Plugin("igate", events=(VoiceCallFrame, VoiceCallEnd))
|
|
bus.register_plugin(data)
|
|
bus.register_plugin(voice)
|
|
_voice_frame(VoicePluginBridge(bus, CONFIG))
|
|
_data_frame(DataPluginBridge(bus, CONFIG))
|
|
assert [type(e) for e in voice.got] == [VoiceCallFrame]
|
|
assert [type(e) for e in data.got] == [UnitDataFrame]
|
|
|
|
|
|
def test_voice_frames_are_not_even_built_when_nobody_takes_them(monkeypatch) -> None:
|
|
bus = PluginBus()
|
|
bus.register_plugin(_Plugin("d-aprs", events=(UnitDataStart, UnitDataFrame, UnitDataEnd)))
|
|
bridge = VoicePluginBridge(bus, CONFIG)
|
|
|
|
def boom(*a, **k):
|
|
raise AssertionError("voice event built for nobody")
|
|
|
|
monkeypatch.setattr(bridge, "_group_ctx_base", boom)
|
|
monkeypatch.setattr(vb, "CallLegContext", boom)
|
|
_voice_frame(bridge)
|
|
assert not bridge.has_subscribers()
|
|
|
|
|
|
def test_unregistering_the_last_interested_plugin_stops_the_work() -> None:
|
|
bus = PluginBus()
|
|
igate = _Plugin("igate", events=(VoiceCallFrame,))
|
|
bus.register_plugin(igate)
|
|
assert bus.wants(VoiceCallFrame)
|
|
bus.unregister_plugin(igate)
|
|
assert not bus.wants(VoiceCallFrame) and not bus.wants_any(VoiceCallStart, UnitDataFrame)
|
|
|
|
|
|
def test_an_internal_handler_takes_everything() -> None:
|
|
bus = PluginBus()
|
|
bus.subscribe(lambda e: None)
|
|
assert bus.wants(VoiceCallFrame) and bus.wants(UnitDataStart)
|
|
|
|
|
|
def test_deferred_events_of_a_tick_share_one_call_later_and_keep_their_order() -> None:
|
|
scheduled: list = []
|
|
bus = PluginBus()
|
|
bus.set_call_later(lambda delay, fn, *a: scheduled.append((fn, a)))
|
|
plugin = _Plugin("p")
|
|
bus.register_plugin(plugin)
|
|
events = [object(), object(), object()]
|
|
for e in events:
|
|
bus.emit_deferred(e)
|
|
assert len(scheduled) == 1 and plugin.got == []
|
|
fn, args = scheduled.pop()
|
|
fn(*args)
|
|
assert plugin.got == events
|
|
bus.emit_deferred(events[0]) # next tick schedules again
|
|
assert len(scheduled) == 1
|
|
|
|
|
|
def test_a_deferred_event_nobody_takes_is_not_scheduled() -> None:
|
|
scheduled: list = []
|
|
bus = PluginBus()
|
|
bus.set_call_later(lambda delay, fn, *a: scheduled.append(fn))
|
|
bus.register_plugin(_Plugin("d-aprs", events=(UnitDataFrame,)))
|
|
bus.emit_deferred(VoiceCallFrame(context=None, dmrpkt=b"", frame_type=0, dtype_vseq=0))
|
|
assert scheduled == []
|
|
|
|
|
|
@pytest.mark.parametrize("calls", [3, vb._ENDED_KEEP + 10])
|
|
def test_stream_bookkeeping_does_not_grow_without_bound(calls: int) -> None:
|
|
bus = PluginBus()
|
|
bus.register_plugin(_Plugin("p"))
|
|
bridge = VoicePluginBridge(bus, CONFIG)
|
|
kw = dict(system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\0\0\xd5", slot=2,
|
|
pkt_time=1.0, source_is_obp=False)
|
|
for n in range(calls):
|
|
sid = n.to_bytes(4, "big")
|
|
assert bridge.emit_group_voice_start(stream_id=sid, **kw)
|
|
assert bridge.emit_group_voice_end(stream_id=sid, duration_s=1.0, **kw)
|
|
assert not bridge.emit_group_voice_end(stream_id=sid, duration_s=1.0, **kw) # still once
|
|
assert bridge._plugin_started == set()
|
|
assert len(bridge._plugin_ended) == min(calls, vb._ENDED_KEEP)
|
|
|
|
|
|
def test_a_stream_resolves_its_aliases_once_and_each_event_owns_its_extra(monkeypatch) -> None:
|
|
calls: list[int] = []
|
|
real = vb.alias_extra
|
|
monkeypatch.setattr(vb, "alias_extra", lambda *a: calls.append(1) or real(*a))
|
|
bus = PluginBus()
|
|
plugin = _Plugin("igate", events=(VoiceCallFrame,))
|
|
bus.register_plugin(plugin)
|
|
bridge = VoicePluginBridge(bus, {**CONFIG, "_SUB_IDS": {1: "C31AG"}})
|
|
for _ in range(6):
|
|
_voice_frame(bridge)
|
|
assert len(calls) == 1
|
|
assert plugin.got[0].context.extra == {"src_callsign": "C31AG"}
|
|
plugin.got[0].context.extra["mutated"] = True
|
|
assert "mutated" not in plugin.got[1].context.extra
|
|
|
|
|
|
def test_the_cached_context_is_dropped_when_the_stream_ends() -> None:
|
|
bus = PluginBus()
|
|
bus.register_plugin(_Plugin("p"))
|
|
bridge = VoicePluginBridge(bus, CONFIG)
|
|
_voice_frame(bridge, stream=7)
|
|
assert bridge._stream_ctx
|
|
bridge.emit_group_voice_end(system_name="SYSTEM", peer_id=b"\0\0\0\1", rf_src=b"\0\0\1", dst_id=b"\0\0\xd5",
|
|
slot=2, stream_id=(7).to_bytes(4, "big"), pkt_time=2.0, duration_s=1.0,
|
|
source_is_obp=False)
|
|
assert bridge._stream_ctx == {}
|