diff --git a/src/adn_server/application/routing/downlink.py b/src/adn_server/application/routing/downlink.py index 1bc6f5e..30126dd 100644 --- a/src/adn_server/application/routing/downlink.py +++ b/src/adn_server/application/routing/downlink.py @@ -100,9 +100,9 @@ def peer_listen_slots(peer: dict[str, Any], tgid: int) -> list[int]: RF timeslots and is expected to key up on both, same as a real repeater configured that way. Simplex peers/bridges always collapse to one slot. """ - from adn_server.application.report.payloads import parse_peer_options_static + from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs - ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + ts1, ts2 = cached_peer_static_tgs(peer) if peer_is_simplex(peer): tg = str(tgid) if tg in ts1 or tg in ts2: diff --git a/src/adn_server/application/routing/helpers.py b/src/adn_server/application/routing/helpers.py index c4ae2f9..587cbcf 100644 --- a/src/adn_server/application/routing/helpers.py +++ b/src/adn_server/application/routing/helpers.py @@ -1318,10 +1318,23 @@ def _system_has_active_bridge_leg( def peer_options_fields(peer: dict[str, Any]) -> dict[str, Any]: - """Parse hotspot OPTIONS into fields used by SINGLE/TIMER resolution.""" + """Parse hotspot OPTIONS into fields used by SINGLE/TIMER resolution. + + Memoized against the OPTIONS blob, the way ``cached_peer_static_tgs`` already + memoizes the static lists: OPTIONS only changes on RPTO, which drops the cache + (``invalidate_peer_options_cache``), while ingress asks for these fields + several times per voice frame through ``peer_single_mode``. + """ + opts = peer.get("OPTIONS") + key = opts if isinstance(opts, bytes) else b"" + cached = peer.get("_CACHED_OPTIONS_FIELDS") + if cached is not None and cached[0] == key: + return cached[1] from adn_server.application.report.payloads import parse_peer_options_fields - return parse_peer_options_fields(peer.get("OPTIONS")) + fields = parse_peer_options_fields(opts) + peer["_CACHED_OPTIONS_FIELDS"] = (key, fields) + return fields def _peer_ua_session_entry( @@ -1381,9 +1394,9 @@ def _peer_static_tg_blocks_slot(peer: dict[str, Any], slot: int, tgid: int) -> b match on one slot must not block genuinely independent dynamic activity on the *other* slot (e.g. TG static on TS2, this same peer separately keying up the same TG on TS1).""" - from adn_server.application.report.payloads import parse_peer_options_static + from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs - ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + ts1, ts2 = cached_peer_static_tgs(peer) tg = str(tgid) if peer_is_simplex(peer): return tg in ts1 or tg in ts2 @@ -1868,9 +1881,9 @@ def peer_single_blocks_foreign_same_tg_downlink( def peer_static_options_tg_count(peer: dict[str, Any]) -> int: """Count distinct static group TGs listed in peer OPTIONS (TS1 ∪ TS2).""" - from adn_server.application.report.payloads import parse_peer_options_static + from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs - ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + ts1, ts2 = cached_peer_static_tgs(peer) return len(set(ts1) | set(ts2)) @@ -1890,18 +1903,18 @@ def peer_receives_group_tgid(peer: dict[str, Any], slot: int, tgid: int) -> bool slot while self-service lists the TG on the other. """ del slot - from adn_server.application.report.payloads import parse_peer_options_static + from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs - ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + ts1, ts2 = cached_peer_static_tgs(peer) tg = str(tgid) return tg in ts1 or tg in ts2 def peer_options_static_tg_slot(peer: dict[str, Any], tgid: int) -> int | None: """Timeslot (1 or 2) where peer OPTIONS list ``tgid``, when unambiguous.""" - from adn_server.application.report.payloads import parse_peer_options_static + from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs - ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + ts1, ts2 = cached_peer_static_tgs(peer) tg = str(tgid) in_ts1 = tg in ts1 in_ts2 = tg in ts2 @@ -1958,9 +1971,9 @@ def peer_downlink_voice_slot( static = peer_options_static_tg_slot(peer, tgid) if static is not None: return static - from adn_server.application.report.payloads import parse_peer_options_static + from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs - ts1, ts2 = parse_peer_options_static(peer.get("OPTIONS")) + ts1, ts2 = cached_peer_static_tgs(peer) tg = str(tgid) if tg in ts1 and tg in ts2: # Static on both slots (peer_options_static_tg_slot returns None diff --git a/src/adn_server/application/routing/peer_downlink_index.py b/src/adn_server/application/routing/peer_downlink_index.py index 69bccb1..b6f5585 100644 --- a/src/adn_server/application/routing/peer_downlink_index.py +++ b/src/adn_server/application/routing/peer_downlink_index.py @@ -37,8 +37,9 @@ from .helpers import peer_single_exclusive_tgid def invalidate_peer_options_cache(peer: dict[str, Any]) -> None: - """Drop cached OPTIONS parse after RPTO.""" + """Drop cached OPTIONS parses after RPTO.""" peer.pop("_CACHED_OPTIONS_STATIC", None) + peer.pop("_CACHED_OPTIONS_FIELDS", None) def cached_peer_static_tgs(peer: dict[str, Any]) -> tuple[tuple[str, ...], tuple[str, ...]]: diff --git a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py index 0e65b0a..32cf760 100644 --- a/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py +++ b/src/adn_server/infrastructure/twisted_adapters/udp_hbp.py @@ -52,6 +52,7 @@ from ...application.routing.downlink import ( from ...application.routing.helpers import ( clear_peer_rx_status_slots, clear_peer_ua_sessions, + derive_peer_rf_mode, hbp_master_ingress_repeat_allowed, is_on_demand_service_dst, is_server_originated_voice, @@ -1267,6 +1268,46 @@ class HBPProtocol(DatagramProtocol): self._refresh_connected_peer_count() self._mark_downlink_index_dirty() + def _acl_rejects_dmrd( + self, + rf_src: bytes, + dst_id: bytes, + slot: int, + stream_id: bytes, + unit_service_dst: bool, + ) -> bool: + """Legacy DMRD ACL gate: global rules, then this system's own. + + The MASTER and PEER ingress paths ran identical copies of this. Order, + the per-slot ``_laststrid`` guard that logs a rejected call once per + stream, and the messages themselves are the legacy ones; on-demand unit + services stay exempt from the TG lists but not from the subscriber ACL. + """ + if not self._router: + return False + for scope, cfg in (("GLOBAL", self._CONFIG.get("GLOBAL", {})), ("SYSTEM", self._config)): + if not cfg.get("USE_ACL"): + continue + if not self._router.acl_check(rf_src, cfg.get("SUB_ACL", (True, []))): + if self._laststrid[slot] != stream_id: + logger.info( + "(%s) CALL DROPPED WITH STREAM ID %s FROM SUBSCRIBER %s BY %s ACL", + self._system, int_id(stream_id), int_id(rf_src), scope, + ) + self._laststrid[slot] = stream_id + return True + if unit_service_dst: + continue + if not self._router.acl_check(dst_id, cfg.get(f"TG{slot}_ACL", (True, []))): + if self._laststrid[slot] != stream_id: + logger.info( + "(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY %s TS%s ACL", + self._system, int_id(stream_id), int_id(dst_id), scope, slot, + ) + self._laststrid[slot] = stream_id + return True + return False + def _master_datagram_received(self, _data: bytes, _sockaddr: tuple[str, int]) -> None: """Direct port of hblink.py master_datagramReceived (lines 888-1146).""" _command = _data[:4] @@ -1311,38 +1352,8 @@ class HBPProtocol(DatagramProtocol): _int_dst_id = int_id(_dst_id) _unit_service_dst = _call_type == "unit" and is_on_demand_service_dst(_int_dst_id) # ACL (legacy order and _laststrid) - if self._router and _global.get("USE_ACL"): - if not self._router.acl_check(_rf_src, _global.get("SUB_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s FROM SUBSCRIBER %s BY GLOBAL ACL", self._system, int_id(_stream_id), int_id(_rf_src)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 1 and not self._router.acl_check(_dst_id, _global.get("TG1_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY GLOBAL TS1 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 2 and not self._router.acl_check(_dst_id, _global.get("TG2_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY GLOBAL TS2 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return - if self._router and self._config.get("USE_ACL"): - if not self._router.acl_check(_rf_src, self._config.get("SUB_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s FROM SUBSCRIBER %s BY SYSTEM ACL", self._system, int_id(_stream_id), int_id(_rf_src)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 1 and not self._router.acl_check(_dst_id, self._config.get("TG1_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY SYSTEM TS1 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 2 and not self._router.acl_check(_dst_id, self._config.get("TG2_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY SYSTEM TS2 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return + if self._acl_rejects_dmrd(_rf_src, _dst_id, _slot, _stream_id, _unit_service_dst): + return # SUB_MAP update (legacy routerHBP.dmrd_received). 4th element # (peer_id) is new — lets same-system private-call repeat target # the exact hotspot instead of broadcasting to every peer. @@ -1732,6 +1743,11 @@ class HBPProtocol(DatagramProtocol): _this_peer["URL"] = _data[98:222] _this_peer["SOFTWARE_ID"] = _data[222:262] _this_peer["PACKAGE_ID"] = _data[262:302] + # RPTC carries the only inputs simplex/duplex depends on (SLOTS and + # the two frequencies), so classify here instead of on every frame: + # peer_rf_mode() reads this and the downlink path asks it three + # times per voice frame. + _this_peer["RF_MODE"] = derive_peer_rf_mode(_this_peer) _sent_call = _rptc_field_str(_this_peer["CALLSIGN"]) if ("ALLOW_UNREG_ID" in self._config and not self._config["ALLOW_UNREG_ID"]) and _sent_call != self.validate_id(_peer_id): self._remove_peer(_peer_id) @@ -1911,40 +1927,9 @@ class HBPProtocol(DatagramProtocol): return pkt_time = time.time() _int_dst_id = int_id(_dst_id) - _global = self._CONFIG.get("GLOBAL", {}) _unit_service_dst = _call_type == "unit" and is_on_demand_service_dst(_int_dst_id) - if self._router and _global.get("USE_ACL"): - if not self._router.acl_check(_rf_src, _global.get("SUB_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s FROM SUBSCRIBER %s BY GLOBAL ACL", self._system, int_id(_stream_id), int_id(_rf_src)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 1 and not self._router.acl_check(_dst_id, _global.get("TG1_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY GLOBAL TS1 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 2 and not self._router.acl_check(_dst_id, _global.get("TG2_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY GLOBAL TS2 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return - if self._router and self._config.get("USE_ACL"): - if not self._router.acl_check(_rf_src, self._config.get("SUB_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s FROM SUBSCRIBER %s BY SYSTEM ACL", self._system, int_id(_stream_id), int_id(_rf_src)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 1 and not self._router.acl_check(_dst_id, self._config.get("TG1_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY SYSTEM TS1 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return - if not _unit_service_dst and _slot == 2 and not self._router.acl_check(_dst_id, self._config.get("TG2_ACL", (True, []))): - if self._laststrid[_slot] != _stream_id: - logger.info("(%s) CALL DROPPED WITH STREAM ID %s ON TGID %s BY SYSTEM TS2 ACL", self._system, int_id(_stream_id), int_id(_dst_id)) - self._laststrid[_slot] = _stream_id - return + if self._acl_rejects_dmrd(_rf_src, _dst_id, _slot, _stream_id, _unit_service_dst): + return # SUB_MAP update (legacy routerHBP.dmrd_received). 4th element # (peer_id) is new — see the MASTER-mode write site for why. sub_map = self._CONFIG.get("_SUB_MAP") diff --git a/tests/hbp/test_acl_gate.py b/tests/hbp/test_acl_gate.py new file mode 100644 index 0000000..ec2a822 --- /dev/null +++ b/tests/hbp/test_acl_gate.py @@ -0,0 +1,134 @@ +# ADN DMR Peer Server - tests hbp ACL gate +# +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY +# +############################################################################### +# 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 +############################################################################### + +"""One ACL gate, shared by the MASTER and PEER DMRD ingress paths.""" + +from __future__ import annotations + +import logging +from dataclasses import replace +from typing import Any + +import pytest + +from tests.harness.deterministic import PacketSpec +from tests.support.hbp_repeat_stack import RecordingTransport, build_hbp_repeat_stack + +from adn_server.domain import bytes_3, bytes_4 +from adn_server.infrastructure.acl_router import InMemoryAclRouter +from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol + +_PEER = (1234567).to_bytes(4, "big") +_ADDR = ("10.0.0.9", 54321) +_RF_SRC = 3120001 +_DENY_SRC = (False, [(_RF_SRC, _RF_SRC)]) +_PERMIT_ALL = (True, [(1, 4294967295)]) +_NO_STREAM = b"\x00" # _make_slot_status seed: no stream seen on this slot yet + + +def _master_stack(**global_acl: Any): + stack = build_hbp_repeat_stack() + # the shared stack ships a permit-everything router; these tests need the real one + stack.hbp._router = InMemoryAclRouter() + # the first thing ingress does after the gate: a frame that got through is in SUB_MAP + stack.config["_SUB_MAP"] = {} + stack.config["GLOBAL"].update(global_acl) + stack.register_peer(_PEER, _ADDR) + return stack + + +def _voice_head(tg: int, slot: int, stream_id: int, *, call_type: str = "group") -> bytes: + return replace( + PacketSpec( + dst_id=tg, slot=slot, stream_id=stream_id, call_type=call_type, + peer_id=int.from_bytes(_PEER, "big"), rf_src=_RF_SRC, + ), + frame_type=2, dtype_vseq=1, + ).data() + + +def test_master_global_subscriber_acl_drops_and_logs_once_per_stream( + caplog: pytest.LogCaptureFixture, +) -> None: + stack = _master_stack(USE_ACL=True, SUB_ACL=_DENY_SRC, TG1_ACL=_PERMIT_ALL, TG2_ACL=_PERMIT_ALL) + with caplog.at_level(logging.INFO): + for _ in range(3): + stack.hbp._master_datagram_received(_voice_head(214, 2, 0x44440001), _ADDR) + + assert stack.config["_SUB_MAP"] == {} + dropped = [r for r in caplog.records if "BY GLOBAL ACL" in r.getMessage()] + assert len(dropped) == 1 + assert "FROM SUBSCRIBER 3120001" in dropped[0].getMessage() + + +def test_master_system_tg_acl_applies_to_the_frame_slot_only() -> None: + stack = _master_stack() + sys_cfg = stack.config["SYSTEMS"][stack.system_name] + sys_cfg["USE_ACL"] = True + sys_cfg["SUB_ACL"] = _PERMIT_ALL + sys_cfg["TG1_ACL"] = (False, [(214, 214)]) + sys_cfg["TG2_ACL"] = _PERMIT_ALL + + stack.hbp._master_datagram_received(_voice_head(214, 1, 0x44440002), _ADDR) + assert stack.config["_SUB_MAP"] == {} + + stack.hbp._master_datagram_received(_voice_head(214, 2, 0x44440003), _ADDR) + assert bytes_3(_RF_SRC) in stack.config["_SUB_MAP"] + + +def test_on_demand_unit_service_skips_tg_acl_but_not_subscriber_acl() -> None: + """Legacy exemption: service destinations bypass the TG lists, never SUB_ACL.""" + stack = _master_stack(USE_ACL=True, SUB_ACL=_PERMIT_ALL, TG1_ACL=(False, [(1, 4294967295)]), + TG2_ACL=(False, [(1, 4294967295)])) + assert not stack.hbp._acl_rejects_dmrd( + bytes_4(_RF_SRC)[1:], bytes_4(9990)[1:], 2, b"\x44\x44\x00\x04", True, + ) + stack.config["GLOBAL"]["SUB_ACL"] = _DENY_SRC + assert stack.hbp._acl_rejects_dmrd( + bytes_4(_RF_SRC)[1:], bytes_4(9990)[1:], 2, b"\x44\x44\x00\x05", True, + ) + + +def test_peer_mode_ingress_uses_the_same_gate() -> None: + """PEER mode (client of another master) rejects on the very same rules.""" + master_sockaddr = ("203.0.113.7", 62031) + config: dict[str, Any] = { + "GLOBAL": {"USE_ACL": True, "SUB_ACL": _DENY_SRC, "TG1_ACL": _PERMIT_ALL, + "TG2_ACL": _PERMIT_ALL}, + "SYSTEMS": { + "PEER-1": { + "MODE": "PEER", + "ENABLED": True, + "LOOSE": True, + "RADIO_ID": _PEER, + "MASTER_SOCKADDR": master_sockaddr, + "USE_ACL": False, + }, + }, + } + hbp = HBPProtocol("PEER-1", config, router=InMemoryAclRouter()) + hbp.transport = RecordingTransport() # type: ignore[assignment] + + hbp._peer_datagram_received(_voice_head(214, 2, 0x44440006), master_sockaddr) + assert hbp.STATUS[2].get("RX_STREAM_ID") == _NO_STREAM + + config["GLOBAL"]["SUB_ACL"] = _PERMIT_ALL + hbp._peer_datagram_received(_voice_head(214, 2, 0x44440007), master_sockaddr) + assert hbp.STATUS[2].get("RX_STREAM_ID") == (0x44440007).to_bytes(4, "big") diff --git a/tests/routing/test_peer_options_cache.py b/tests/routing/test_peer_options_cache.py new file mode 100644 index 0000000..67d50ef --- /dev/null +++ b/tests/routing/test_peer_options_cache.py @@ -0,0 +1,134 @@ +# ADN DMR Peer Server - tests peer OPTIONS / RF mode caching +# +# Copyright (C) 2026 Rodrigo Pérez, CE5RPY +# +############################################################################### +# 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 +############################################################################### + +"""Per-peer OPTIONS fields and RF mode are parsed on change, not per frame.""" + +from __future__ import annotations + +from typing import Any + +from tests.support.hbp_repeat_stack import build_hbp_repeat_stack + +from adn_server.application.routing.helpers import ( + RF_MODE_DUPLEX, + RF_MODE_SIMPLEX, + peer_options_fields, + peer_rf_mode, + peer_single_mode, +) +from adn_server.application.routing.peer_downlink_index import cached_peer_static_tgs + +_PEER = (1234567).to_bytes(4, "big") +_ADDR = ("10.0.0.9", 54321) + + +def _rptc(peer_id: bytes, *, slots: bytes, rx: bytes, tx: bytes) -> bytes: + """MMDVM RPTC frame (field widths per hblink.py / MMDVMHost RPTC layout).""" + return b"".join( + [ + b"RPTC", + peer_id, + b"CE5RPY ", + rx.ljust(9, b"0"), + tx.ljust(9, b"0"), + b"01", + b"01", + b"00.00000", + b"000.00000", + b"000", + b"Andorra".ljust(20), + b"hotspot".ljust(19), + slots, + b"".ljust(124), + b"MMDVM".ljust(40), + b"".ljust(40), + ] + ) + + +def _waiting_peer(stack: Any, peer_id: bytes, addr: tuple[str, int]) -> dict[str, Any]: + peer: dict[str, Any] = { + "CONNECTION": "WAITING_CONFIG", + "SOCKADDR": addr, + "RADIO_ID": str(int.from_bytes(peer_id, "big")), + "CALLSIGN": b"CE5RPY ", + } + stack.hbp._peers[peer_id] = peer + return peer + + +def test_options_fields_reused_while_the_blob_is_unchanged() -> None: + """Same OPTIONS must not be re-parsed: ingress asks several times per frame.""" + stack = build_hbp_repeat_stack() + stack.register_peer(_PEER, _ADDR) + peer = stack.hbp._peers[_PEER] + peer["OPTIONS"] = b"TS2_1=214;SINGLE=1;TIMER=15;" + + first = peer_options_fields(peer) + assert peer_options_fields(peer) is first + + +def test_rpto_makes_the_new_options_take_effect() -> None: + """RPTO drops the cached parse, so SINGLE/TIMER and static TGs follow the new blob.""" + stack = build_hbp_repeat_stack() + sys_cfg = stack.config["SYSTEMS"][stack.system_name] + stack.register_peer(_PEER, _ADDR) + peer = stack.hbp._peers[_PEER] + + stack.hbp._master_datagram_received(b"RPTO" + _PEER + b"TS2_1=214;SINGLE=1;", _ADDR) + assert peer_single_mode(peer, sys_cfg) is True + assert cached_peer_static_tgs(peer) == ((), ("214",)) + + stack.hbp._master_datagram_received(b"RPTO" + _PEER + b"TS1_1=91;SINGLE=0;", _ADDR) + assert peer_single_mode(peer, sys_cfg) is False + assert cached_peer_static_tgs(peer) == (("91",), ()) + assert peer_options_fields(peer).get("SINGLE") == "0" + + +def test_rptc_classifies_rf_mode_at_login() -> None: + """Simplex/duplex is decided from the RPTC fields, not on every frame.""" + stack = build_hbp_repeat_stack() + stack.config["SYSTEMS"][stack.system_name]["ALLOW_UNREG_ID"] = True + peer = _waiting_peer(stack, _PEER, _ADDR) + + stack.hbp._master_datagram_received( + _rptc(_PEER, slots=b"3", rx=b"438500000", tx=b"431100000"), _ADDR, + ) + assert peer["CONNECTION"] == "YES" + assert peer["RF_MODE"] == RF_MODE_DUPLEX + assert peer_rf_mode(peer) == RF_MODE_DUPLEX + + +def test_relogin_with_new_slots_reclassifies_rf_mode() -> None: + """A hotspot that comes back as simplex must not keep the duplex verdict.""" + stack = build_hbp_repeat_stack() + stack.config["SYSTEMS"][stack.system_name]["ALLOW_UNREG_ID"] = True + peer = _waiting_peer(stack, _PEER, _ADDR) + stack.hbp._master_datagram_received( + _rptc(_PEER, slots=b"3", rx=b"438500000", tx=b"431100000"), _ADDR, + ) + assert peer["RF_MODE"] == RF_MODE_DUPLEX + + peer["CONNECTION"] = "WAITING_CONFIG" + stack.hbp._master_datagram_received( + _rptc(_PEER, slots=b"4", rx=b"145500000", tx=b"145500000"), _ADDR, + ) + assert peer["RF_MODE"] == RF_MODE_SIMPLEX + assert peer_rf_mode(peer) == RF_MODE_SIMPLEX