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.
142 lines
5.5 KiB
142 lines
5.5 KiB
"""Talker Alias bridge/repeat injection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from tests.harness.deterministic import (
|
|
DeterministicScenario,
|
|
PacketSpec,
|
|
active_bridge,
|
|
add_openbridge_system,
|
|
)
|
|
from tests.harness.scenarios import talker_alias_config
|
|
|
|
from adn_server.domain import bytes_3, bytes_4
|
|
from adn_server.domain.talker_alias import DMRA_BLOCK_COUNT, DMRA_OPCODE
|
|
|
|
|
|
def test_talker_alias_inject_dmra_on_bridge_vhead() -> None:
|
|
"""Bridge forward sends DMRA blocks once per target stream on VHEAD."""
|
|
bridges = active_bridge(91, (("MASTER-A", 2), ("MASTER-B", 2)))
|
|
scenario = DeterministicScenario(config=talker_alias_config(), bridges=bridges)
|
|
base = PacketSpec(dst_id=91, rf_src=3120001, stream_id=0x90909090)
|
|
|
|
scenario.inject_hbp("MASTER-A", DeterministicScenario.voice_head_spec(base))
|
|
|
|
assert len(scenario.dmra_capture) == 1
|
|
dmra = scenario.dmra_capture[0]
|
|
assert dmra.target_system == "MASTER-B"
|
|
assert 1 <= len(dmra.packets) <= DMRA_BLOCK_COUNT
|
|
assert all(p[:4] == DMRA_OPCODE for p in dmra.packets)
|
|
payload = b"".join(p[8:15] for p in dmra.packets)
|
|
assert b"CE5RPY" in payload
|
|
|
|
|
|
def test_talker_alias_vhead_sent_once_per_target_stream() -> None:
|
|
bridges = active_bridge(91, (("MASTER-A", 2), ("MASTER-B", 2)))
|
|
scenario = DeterministicScenario(config=talker_alias_config(), bridges=bridges)
|
|
base = PacketSpec(dst_id=91, rf_src=3120001, stream_id=0x91919191)
|
|
|
|
scenario.inject_hbp("MASTER-A", DeterministicScenario.voice_head_spec(base))
|
|
scenario.inject_hbp(
|
|
"MASTER-A",
|
|
DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1),
|
|
)
|
|
|
|
assert len(scenario.dmra_capture) == 1
|
|
|
|
|
|
def test_talker_alias_both_obp_source_injects_immediately() -> None:
|
|
"""both + OBP source (can never carry TA): inject template at VHEAD without waiting."""
|
|
config = talker_alias_config()
|
|
config["GLOBAL"]["TALKER_ALIAS_MODE"] = "both"
|
|
add_openbridge_system(config, "OBP-CL")
|
|
bridges = active_bridge(52090, (("OBP-CL", 2), ("MASTER-A", 2)))
|
|
scenario = DeterministicScenario(config=config, bridges=bridges)
|
|
base = PacketSpec(dst_id=52090, rf_src=3120001, stream_id=0x93939393, slot=2)
|
|
|
|
scenario.inject_obp("OBP-CL", DeterministicScenario.voice_head_spec(base))
|
|
|
|
assert len(scenario.dmra_capture) == 1
|
|
dmra = scenario.dmra_capture[0]
|
|
assert dmra.target_system == "MASTER-A"
|
|
payload = b"".join(p[8:15] for p in dmra.packets)
|
|
assert b"CE5RPY" in payload
|
|
|
|
|
|
def test_both_mode_without_ta_still_rewrites_group_embedded_lc() -> None:
|
|
"""Regression: with no injected TA (TX_TA_EMB None), bursts must still carry the
|
|
destination group LC (legacy bridge.py parity).
|
|
|
|
The earlier 'preserve source embedded LC' shortcut left a stale LC on the wire, which
|
|
the receiving MMDVM rejected on TG-rewriting legs -> 'watchdog expired, packet loss'.
|
|
"""
|
|
from bitarray import bitarray
|
|
|
|
config = talker_alias_config()
|
|
config["GLOBAL"]["TALKER_ALIAS_MODE"] = "both"
|
|
bridges = active_bridge(91, (("MASTER-A", 2), ("MASTER-B", 2)))
|
|
scenario = DeterministicScenario(config=config, bridges=bridges)
|
|
base = PacketSpec(dst_id=91, rf_src=3120001, stream_id=0x95959595, slot=2)
|
|
|
|
scenario.inject_hbp("MASTER-A", DeterministicScenario.voice_head_spec(base))
|
|
scenario.inject_hbp(
|
|
"MASTER-A", DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1)
|
|
)
|
|
|
|
ts_st = scenario.protocols["MASTER-B"].STATUS[2]
|
|
assert ts_st.get("TX_TA_EMB") is not None # both mode injects template at VHEAD
|
|
bursts = [p for p in scenario.capture.for_system("MASTER-B") if p.fields["dtype_vseq"] == 1]
|
|
assert bursts, "voice burst B should be forwarded to MASTER-B"
|
|
bits = bitarray(endian="big")
|
|
bits.frombytes(bursts[-1].fields["dmr_payload"])
|
|
emb = bits[116:148]
|
|
assert emb == ts_st["TX_EMB_LC"][1] # rewritten to destination group LC...
|
|
assert emb.any() # ...not the preserved all-zero source embedded LC
|
|
|
|
|
|
def test_talker_alias_local_repeat_excludes_source_peer() -> None:
|
|
scenario = DeterministicScenario(
|
|
config=talker_alias_config(),
|
|
bridges=active_bridge(91, (("MASTER-A", 2),)),
|
|
)
|
|
stream_id = bytes_4(0x92929292)
|
|
peer = bytes_4(1001)
|
|
rf_src = bytes_3(3120001)
|
|
|
|
scenario.bridge.send_talker_alias_local_repeat("MASTER-A", peer, rf_src, stream_id)
|
|
|
|
assert len(scenario.dmra_capture) == 1
|
|
assert scenario.dmra_capture[0].exclude_peer == peer
|
|
|
|
|
|
def test_repeat_embed_prepares_slot_state_for_mmdvm() -> None:
|
|
"""REPEAT must set TX_TA_EMB on the source MASTER slot for embedded downlink DMRD."""
|
|
from bitarray import bitarray
|
|
|
|
scenario = DeterministicScenario(
|
|
config=talker_alias_config(),
|
|
bridges=active_bridge(7304, (("MASTER-A", 2),)),
|
|
)
|
|
stream_id = bytes_4(0xA1A2A3A4)
|
|
peer = bytes_4(1001)
|
|
rf_src = bytes_3(7300392)
|
|
dst_id = bytes_3(7304)
|
|
|
|
scenario.bridge.prepare_talker_alias_local_repeat(
|
|
"MASTER-A", peer, rf_src, dst_id, 2, stream_id,
|
|
)
|
|
slot_st = scenario.protocols["MASTER-A"].STATUS[2]
|
|
assert slot_st.get("REP_STREAM_ID") == stream_id
|
|
assert "REP_EMB_LC" in slot_st
|
|
assert slot_st.get("TX_TA_EMB") is not None
|
|
|
|
raw = b"\x00" * 33
|
|
out = scenario.bridge.rewrite_repeat_voice_burst("MASTER-A", 2, stream_id, 1, raw)
|
|
bits = bitarray(endian="big")
|
|
bits.frombytes(out)
|
|
assert bits[116:148] == slot_st["REP_EMB_LC"][1]
|
|
|
|
scenario.bridge.clear_talker_alias_stream("MASTER-A", stream_id)
|
|
assert "REP_STREAM_ID" not in slot_st
|
|
assert "TX_TA_EMB" not in slot_st
|