KB8PMY
/
ADN-DMR-Peer-Server
mirror of https://github.com/Amateur-Digital-Network/ADN-DMR-Peer-Server.git
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity
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.
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'develop'
${ noResults }
ADN-DMR-Peer-Server/tests/routing/test_unit_data_routing.py

428 lines
18 KiB
Raw Permalink Blame History

# ADN DMR Peer Server - tests routing unit data routing
#
# 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
###############################################################################
"""Unit data routing: SUB_MAP, hotspot, gateway, OBP fanout."""
from __future__ import annotations
import pytest
from tests.harness.assertions import assert_forwarded, assert_not_forwarded
from tests.harness.deterministic import (
DeterministicScenario,
FakeClock,
PacketSpec,
add_openbridge_system,
minimal_config,
parse_dmr_fields,
patch_routing_wall_time,
)
from tests.routing.unit_data_helpers import idle_hbp_slot
from adn_server.domain import bytes_3, bytes_4
from adn_server.domain.hbp_protocol import HBPF_SLT_VHEAD, HBPF_SLT_VTERM
DAPRS_GATEWAY_ID = 900999
HOTSPOT_SUB_ID = 7300392
_HARNESS_T0 = FakeClock().time()
@pytest.mark.behavior
def test_unit_data_hbp_forward_wire_format_matches_legacy_send_data_to_hbp() -> None:
"""Server is a transparent bridge: only TS bit may change; DMR payload [20:53] is untouched."""
config = minimal_config(("SYSTEM", "D-APRS"))
config["SYSTEMS"]["D-APRS"]["PEERS"] = {
DAPRS_GATEWAY_ID: {"CALLSIGN": "D-APRS", "IP": "127.0.0.1", "PORT": 52555},
}
config["SYSTEMS"]["D-APRS"]["GROUP_HANGTIME"] = 0
scenario = DeterministicScenario(config=config)
scenario.protocols["D-APRS"].STATUS[2] = idle_hbp_slot()
payload = bytes(range(33)) # distinct pattern for dmrpkt [20:53]
base = PacketSpec(
call_type="unit",
rf_src=HOTSPOT_SUB_ID,
dst_id=DAPRS_GATEWAY_ID,
stream_id=0xDEADBEEF,
slot=1,
dtype_vseq=6,
payload=payload,
)
spec = DeterministicScenario.unit_data_header_spec(base)
ingress = spec.data()
scenario.inject_unit("SYSTEM", spec)
forwarded = scenario.capture.for_system("D-APRS")
assert len(forwarded) == 1
out = forwarded[0].packet
assert out[:15] == ingress[:15]
assert out[16:20] == ingress[16:20]
assert out[20:53] == ingress[20:53]
assert out[8:11] == ingress[8:11]
assert out[5:8] == ingress[5:8]
assert (out[15] ^ ingress[15]) == 0x80
@pytest.mark.behavior
def test_unit_data_sub_map_forwards_to_idle_hbp() -> None:
config = minimal_config(("MASTER-A", "MASTER-B"))
# 6-digit dst: only unit_data SUB_MAP path (pvt_call runs for 7-digit only, legacy ~3253).
dst_sub = 712345
config["_SUB_MAP"] = {bytes_3(dst_sub): ("MASTER-B", 2, 1000.0)}
scenario = DeterministicScenario(config=config)
scenario.protocols["MASTER-B"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(call_type="unit", dst_id=dst_sub, stream_id=0x52525252, slot=2)
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert_forwarded(scenario, "MASTER-B", count=1, call_type="unit")
@pytest.mark.behavior
def test_unit_data_7digit_sub_map_idle_forwards_via_both_legacy_paths() -> None:
"""Legacy parity: dtype 6/7/8 to 7-digit dst runs unit_data + pvt_call_received."""
config = minimal_config(("MASTER-A", "MASTER-B"))
dst_sub = 7123456
config["_SUB_MAP"] = {bytes_3(dst_sub): ("MASTER-B", 2, 1000.0)}
scenario = DeterministicScenario(config=config)
scenario.protocols["MASTER-B"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(call_type="unit", dst_id=dst_sub, stream_id=0x52525252, slot=2)
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert_forwarded(scenario, "MASTER-B", count=2, call_type="unit")
def test_unit_data_sub_map_skips_busy_hbp_target() -> None:
config = minimal_config(("MASTER-A", "MASTER-B"))
# 6-digit dst isolates SUB_MAP busy check (no pvt_call fallback).
dst_sub = 712345
config["_SUB_MAP"] = {bytes_3(dst_sub): ("MASTER-B", 2, 1000.0)}
scenario = DeterministicScenario(config=config)
scenario.protocols["MASTER-B"].STATUS[2] = {
"RX_TYPE": HBPF_SLT_VHEAD,
"TX_TYPE": HBPF_SLT_VTERM,
"TX_TIME": scenario.clock.time(),
}
base = PacketSpec(call_type="unit", dst_id=dst_sub, stream_id=0x53535353, slot=2)
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert_not_forwarded(scenario, "MASTER-B")
@pytest.mark.behavior
def test_unit_data_hotspot_peer_match_forwards_to_idle_hbp() -> None:
"""Regression: hotspot peer ID match forwards unit data to idle MASTER slot."""
config = minimal_config(("MASTER-A", "MASTER-B"))
config["SYSTEMS"]["MASTER-B"]["PEERS"] = {
1234567890: {"CALLSIGN": "HS1", "IP": "127.0.0.1", "PORT": 62040},
}
config["SYSTEMS"]["MASTER-B"]["GROUP_HANGTIME"] = 0
scenario = DeterministicScenario(config=config)
scenario.protocols["MASTER-B"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(call_type="unit", dst_id=123456, stream_id=0x54545454, slot=2)
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert_forwarded(scenario, "MASTER-B", count=1, call_type="unit", dst_id=123456)
def test_unit_data_forwards_to_data_gateway_when_enabled() -> None:
config = minimal_config(("MASTER-A",))
config["GLOBAL"]["DATA_GATEWAY"] = True
add_openbridge_system(config, "DATA-GATEWAY")
config["SYSTEMS"]["DATA-GATEWAY"]["ENABLED"] = True
scenario = DeterministicScenario(config=config)
base = PacketSpec(call_type="unit", dst_id=5005, stream_id=0x65656565, slot=2)
with patch_routing_wall_time(scenario.clock):
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert len(scenario.capture.for_system("DATA-GATEWAY")) == 1
def test_unit_data_fanout_to_other_obp_with_ver_gt_1() -> None:
config = minimal_config(("MASTER-A",))
add_openbridge_system(config, "OBP-FAN")
scenario = DeterministicScenario(config=config)
base = PacketSpec(call_type="unit", dst_id=1000001, stream_id=0x66666666, slot=2)
with patch_routing_wall_time(scenario.clock):
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert len(scenario.capture.for_system("OBP-FAN")) == 1
assert parse_dmr_fields(scenario.capture.for_system("OBP-FAN")[0].packet)["call_type"] == "unit"
@pytest.mark.behavior
def test_unit_data_to_fresh_local_sub_is_not_fanned_out_to_obp() -> None:
"""Unit data to a subscriber recently seen on a local system goes only via SUB_MAP."""
config = minimal_config(("D-APRS", "SYSTEM"))
add_openbridge_system(config, "OBP-FAN")
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, _HARNESS_T0 - 60)}
scenario = DeterministicScenario(config=config)
scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(call_type="unit", rf_src=DAPRS_GATEWAY_ID, dst_id=HOTSPOT_SUB_ID, stream_id=0x67676767, slot=2)
with patch_routing_wall_time(scenario.clock):
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
assert len(scenario.capture.for_system("OBP-FAN")) == 0
assert len(scenario.capture.for_system("SYSTEM")) >= 1
@pytest.mark.behavior
def test_unit_data_to_stale_local_sub_is_still_fanned_out_to_obp() -> None:
config = minimal_config(("D-APRS", "SYSTEM"))
add_openbridge_system(config, "OBP-FAN")
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, _HARNESS_T0 - 3600)}
scenario = DeterministicScenario(config=config)
scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(call_type="unit", rf_src=DAPRS_GATEWAY_ID, dst_id=HOTSPOT_SUB_ID, stream_id=0x68686868, slot=2)
with patch_routing_wall_time(scenario.clock):
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
assert len(scenario.capture.for_system("OBP-FAN")) == 1
@pytest.mark.behavior
def test_unit_data_to_sub_learned_via_obp_is_still_fanned_out_to_obp() -> None:
config = minimal_config(("D-APRS",))
add_openbridge_system(config, "OBP-FAN")
add_openbridge_system(config, "OBP-OTHER")
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("OBP-OTHER", 1, _HARNESS_T0 - 60)}
scenario = DeterministicScenario(config=config)
base = PacketSpec(call_type="unit", rf_src=DAPRS_GATEWAY_ID, dst_id=HOTSPOT_SUB_ID, stream_id=0x69696969, slot=2)
with patch_routing_wall_time(scenario.clock):
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
assert len(scenario.capture.for_system("OBP-FAN")) == 1
@pytest.mark.behavior
def test_unit_data_daprs_peer_id_match_forwards() -> None:
"""Legacy parity: unit data routes when dst matches a PEER on D-APRS (runtime or static)."""
for gateway_id in (DAPRS_GATEWAY_ID, 730999):
config = minimal_config(("SYSTEM", "D-APRS"))
config["SYSTEMS"]["D-APRS"]["PEERS"] = {
gateway_id: {"CALLSIGN": "D-APRS", "IP": "127.0.0.1", "PORT": 52555},
}
config["SYSTEMS"]["D-APRS"]["GROUP_HANGTIME"] = 0
scenario = DeterministicScenario(config=config)
scenario.protocols["D-APRS"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(
call_type="unit",
rf_src=HOTSPOT_SUB_ID,
dst_id=gateway_id,
stream_id=0xAABBCCDD,
slot=2,
)
scenario.inject_unit("SYSTEM", DeterministicScenario.unit_data_header_spec(base))
assert_forwarded(scenario, "D-APRS", count=1, call_type="unit", dst_id=gateway_id)
@pytest.mark.behavior
def test_daprs_uplink_then_downlink_while_hotspot_transmitting() -> None:
"""D-APRS: uplink unit data must not mark SYSTEM slot busy; reply via SUB_MAP forwards."""
config = minimal_config(("SYSTEM", "D-APRS"))
config["SYSTEMS"]["D-APRS"]["PEERS"] = {
DAPRS_GATEWAY_ID: {"CALLSIGN": "D-APRS", "IP": "127.0.0.1", "PORT": 52555},
}
config["SYSTEMS"]["D-APRS"]["GROUP_HANGTIME"] = 0
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, 1000.0)}
scenario = DeterministicScenario(config=config)
scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
scenario.protocols["D-APRS"].STATUS[2] = idle_hbp_slot()
uplink = PacketSpec(
call_type="unit",
rf_src=HOTSPOT_SUB_ID,
dst_id=DAPRS_GATEWAY_ID,
stream_id=0x41381248,
slot=2,
)
scenario.inject_unit("SYSTEM", DeterministicScenario.unit_data_header_spec(uplink))
assert scenario.protocols["SYSTEM"].STATUS[2]["RX_TYPE"] == HBPF_SLT_VTERM
assert_forwarded(scenario, "D-APRS", count=1, call_type="unit")
scenario.capture.packets.clear()
downlink = PacketSpec(
call_type="unit",
rf_src=DAPRS_GATEWAY_ID,
dst_id=HOTSPOT_SUB_ID,
stream_id=0x33416424,
slot=2,
)
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(downlink))
# Idle SYSTEM slot: legacy sends via sendDataToHBP and pvt_call send_system (7-digit dst).
assert_forwarded(scenario, "SYSTEM", count=2, call_type="unit", dst_id=HOTSPOT_SUB_ID)
@pytest.mark.behavior
def test_unit_data_7digit_dst_uses_pvt_call_when_sub_map_idle_check_fails() -> None:
"""Legacy parity: dtype 6/7/8 to 7-digit dst also runs pvt_call_received (routerHBP ~3253)."""
from adn_server.domain.hbp_protocol import HBPF_SLT_VHEAD
config = minimal_config(("D-APRS", "SYSTEM"))
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, 1000.0)}
scenario = DeterministicScenario(config=config)
# SUB_MAP idle check fails (strict RX/TX VTERM), but pvt_call uses STREAM_TO contention.
scenario.protocols["SYSTEM"].STATUS[2] = {
"RX_TYPE": HBPF_SLT_VHEAD,
"TX_TYPE": HBPF_SLT_VHEAD,
"RX_TGID": bytes_3(900999),
"TX_TGID": bytes_3(0),
"RX_TIME": scenario.clock.time(),
"TX_TIME": 0.0,
"RX_STREAM_ID": bytes_3(0),
}
base = PacketSpec(
call_type="unit",
rf_src=900999,
dst_id=HOTSPOT_SUB_ID,
stream_id=0x86673819,
slot=2,
)
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
assert_forwarded(scenario, "SYSTEM", count=1, call_type="unit", dst_id=HOTSPOT_SUB_ID)
@pytest.mark.behavior
def test_unit_data_7digit_multi_frame_pierces_rx_tgid_match() -> None:
"""Unit data downlink must forward all frames when target RX_TGID equals private dst."""
from adn_server.domain.hbp_protocol import HBPF_DATA_SYNC, HBPF_SLT_VHEAD
config = minimal_config(("D-APRS", "SYSTEM"))
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, 1000.0)}
scenario = DeterministicScenario(config=config)
scenario.protocols["SYSTEM"].STATUS[2] = {
"RX_TYPE": HBPF_SLT_VHEAD,
"TX_TYPE": HBPF_SLT_VHEAD,
"RX_TGID": bytes_3(HOTSPOT_SUB_ID),
"TX_TGID": bytes_3(0),
"RX_TIME": scenario.clock.time(),
"TX_TIME": 0.0,
"RX_STREAM_ID": bytes_3(0),
}
stream = 0xAABBCCDD
for i, dtype in enumerate([3, 6, 8, 8, 8]):
spec = PacketSpec(
call_type="unit",
rf_src=900999,
dst_id=HOTSPOT_SUB_ID,
stream_id=stream,
slot=2,
dtype_vseq=dtype,
frame_type=HBPF_DATA_SYNC,
payload=bytes([dtype]) * 33,
seq=i,
)
scenario.inject_unit("D-APRS", spec)
assert_forwarded(scenario, "SYSTEM", count=5, call_type="unit", dst_id=HOTSPOT_SUB_ID)
@pytest.mark.behavior
def test_unit_data_pvt_call_does_not_emit_private_voice_monitor_events() -> None:
"""Unit data downlink must not light monitor TS chips (no PRIVATE VOICE START/END)."""
config = minimal_config(("D-APRS", "SYSTEM"))
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, 1000.0)}
scenario = DeterministicScenario(config=config, enable_reporting=True)
scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
base = PacketSpec(
call_type="unit",
rf_src=900999,
dst_id=HOTSPOT_SUB_ID,
stream_id=0x11223344,
slot=2,
)
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
assert_forwarded(scenario, "SYSTEM", count=2, call_type="unit", dst_id=HOTSPOT_SUB_ID)
assert scenario.report_factory is not None
private_voice = [e for e in scenario.report_factory.events if e.startswith("PRIVATE VOICE")]
assert private_voice == []
@pytest.mark.behavior
def test_unit_data_sub_map_with_peer_id_targets_only_that_peer() -> None:
"""Regression: cross-talk -- unit DATA to a known subscriber must reach only the
one hotspot it was last heard on, not every peer of the destination system
(send_to_system/send_peers broadcasts to all peers when no peer is targeted)."""
dst_sub = 712345
dst_peer = bytes_4(730039101)
config = minimal_config(("MASTER-A", "MASTER-B"))
config["_SUB_MAP"] = {bytes_3(dst_sub): ("MASTER-B", 2, 1000.0, dst_peer)}
scenario = DeterministicScenario(config=config)
scenario.protocols["MASTER-B"].STATUS[2] = idle_hbp_slot()
scenario.protocols["MASTER-B"]._peers[dst_peer] = {}
base = PacketSpec(call_type="unit", dst_id=dst_sub, stream_id=0x52525253, slot=2)
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert scenario.protocols["MASTER-B"].sent_to_peer, "must deliver directly to the known peer"
assert scenario.protocols["MASTER-B"].sent_to_peer[0][0] == dst_peer
assert_not_forwarded(scenario, "MASTER-B")
@pytest.mark.behavior
def test_unit_data_sub_map_peer_not_connected_falls_back_to_broadcast() -> None:
"""When the known peer isn't (or is no longer) connected, fall back to the old
broadcast-to-system behavior rather than silently dropping the packet."""
dst_sub = 712345
dst_peer = bytes_4(730039101)
config = minimal_config(("MASTER-A", "MASTER-B"))
config["_SUB_MAP"] = {bytes_3(dst_sub): ("MASTER-B", 2, 1000.0, dst_peer)}
scenario = DeterministicScenario(config=config)
scenario.protocols["MASTER-B"].STATUS[2] = idle_hbp_slot()
# dst_peer intentionally NOT registered in scenario.protocols["MASTER-B"]._peers.
base = PacketSpec(call_type="unit", dst_id=dst_sub, stream_id=0x52525254, slot=2)
scenario.inject_unit("MASTER-A", DeterministicScenario.unit_data_header_spec(base))
assert not scenario.protocols["MASTER-B"].sent_to_peer
assert_forwarded(scenario, "MASTER-B", count=1, call_type="unit")
@pytest.mark.behavior
def test_pvt_call_sub_map_with_peer_id_targets_only_that_peer() -> None:
"""Same cross-talk fix for the 7-digit pvt_call_received cross-system path (private
voice and ARS/LRRP downlink use this, independent of the unit-data SUB_MAP branch)."""
dst_peer = bytes_4(730039101)
config = minimal_config(("D-APRS", "SYSTEM"))
config["_SUB_MAP"] = {bytes_3(HOTSPOT_SUB_ID): ("SYSTEM", 2, 1000.0, dst_peer)}
scenario = DeterministicScenario(config=config)
scenario.protocols["SYSTEM"].STATUS[2] = idle_hbp_slot()
scenario.protocols["SYSTEM"]._peers[dst_peer] = {}
base = PacketSpec(
call_type="unit",
rf_src=DAPRS_GATEWAY_ID,
dst_id=HOTSPOT_SUB_ID,
stream_id=0x33416425,
slot=2,
)
scenario.inject_unit("D-APRS", DeterministicScenario.unit_data_header_spec(base))
assert scenario.protocols["SYSTEM"].sent_to_peer, "must deliver directly to the known peer"
assert scenario.protocols["SYSTEM"].sent_to_peer[0][0] == dst_peer
assert_not_forwarded(scenario, "SYSTEM")
Reference in new issue View Git Blame Copy Permalink

Powered by TurnKey Linux.