Add real-stack REPEAT and proxy fan-in E2E tests so regressions outside DeterministicScenario are caught. Remove duplicate or fragile tests, consolidate monitor remap cases, and document harness vs integration policy.pull/1/head
Rodrigo Pérez
4 months ago
parent
614c882bcb
commit
46024c4da4
@ -1,19 +0,0 @@
|
||||
"""Regression: configs that must keep loading after proxy integration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from adn_server.infrastructure import YamlConfigLoader
|
||||
|
||||
|
||||
def test_adn_parrot_yaml_loads_without_proxy_block() -> None:
|
||||
loader = YamlConfigLoader("/opt/new-adn-server")
|
||||
config = loader.load("/opt/new-adn-server/adn-parrot.yaml")
|
||||
assert "PROXY" not in config or config.get("PROXY") == {}
|
||||
assert config["SYSTEMS"]["PARROT"]["MODE"] == "PEER"
|
||||
|
||||
|
||||
def test_adn_server_yaml_loads_with_proxy() -> None:
|
||||
loader = YamlConfigLoader("/opt/new-adn-server")
|
||||
config = loader.load("/opt/new-adn-server/adn-server.yaml")
|
||||
assert config["PROXY"]["TARGET_SYSTEM"] == "SYSTEM"
|
||||
assert config["PROXY"]["LISTEN_PORT"] == 62031
|
||||
@ -0,0 +1,88 @@
|
||||
"""Proxy fan-in → MASTER inject → REPEAT to second hotspot (DroidStar path)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
from bitarray import bitarray
|
||||
|
||||
from adn_server.application.proxy import ProxyUseCases
|
||||
from adn_server.domain import bytes_4
|
||||
from adn_server.infrastructure.config_normalizer import ensure_system_runtime_config
|
||||
from adn_server.infrastructure.hbp_constants import DMRD
|
||||
from adn_server.infrastructure.proxy import (
|
||||
InMemoryPendingRptoQueue,
|
||||
InMemoryProxySlotStore,
|
||||
InProcessHbpSink,
|
||||
ProxyFanInProtocol,
|
||||
ProxyReplyTransport,
|
||||
)
|
||||
from tests.harness.deterministic import DeterministicScenario, PacketSpec
|
||||
from tests.support.hbp_repeat_stack import HbpRepeatStack, build_hbp_repeat_stack
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
_PEER_TX = bytes_4(730039210)
|
||||
_PEER_RX = bytes_4(730039101)
|
||||
_ADDR_TX = ("192.168.50.10", 62031)
|
||||
_ADDR_RX = ("192.168.50.20", 62032)
|
||||
_EMB_SLICE = slice(116, 148)
|
||||
|
||||
|
||||
def _proxy_fanin_stack() -> tuple[ProxyFanInProtocol, HbpRepeatStack]:
|
||||
stack = build_hbp_repeat_stack(talker_alias=True, system_name="MASTER-A")
|
||||
stack.register_peer(_PEER_TX, _ADDR_TX)
|
||||
stack.register_peer(_PEER_RX, _ADDR_RX)
|
||||
|
||||
proxy = ProxyUseCases(
|
||||
InMemoryProxySlotStore(),
|
||||
InMemoryPendingRptoQueue(),
|
||||
max_peers=8,
|
||||
)
|
||||
sink = InProcessHbpSink(stack.hbp)
|
||||
fanin = ProxyFanInProtocol(proxy, sink)
|
||||
fanin.transport = stack.transport # type: ignore[assignment]
|
||||
stack.hbp.transport = ProxyReplyTransport(stack.transport)
|
||||
return fanin, stack
|
||||
|
||||
|
||||
def test_proxy_inject_repeat_reaches_listener_with_embed_lc() -> None:
|
||||
"""TX via proxy (DroidStar) must REPEAT to other peer with TA embed, not raw copy."""
|
||||
fanin, stack = _proxy_fanin_stack()
|
||||
base = PacketSpec(
|
||||
peer_id=730039210,
|
||||
rf_src=7300392,
|
||||
dst_id=7304,
|
||||
slot=2,
|
||||
stream_id=0x11223344,
|
||||
payload=b"\x00" * 33,
|
||||
)
|
||||
vhead = DeterministicScenario.voice_head_spec(base)
|
||||
burst = DeterministicScenario.voice_burst_spec(base, seq=1, dtype_vseq=1)
|
||||
|
||||
fanin.datagramReceived(vhead.data(), _ADDR_TX)
|
||||
stack.transport.clear()
|
||||
fanin.datagramReceived(burst.data(), _ADDR_TX)
|
||||
|
||||
listener_pkts = [p for p in stack.transport.for_addr(_ADDR_RX) if p[:4] == DMRD]
|
||||
assert len(listener_pkts) == 1
|
||||
bits = bitarray(endian="big")
|
||||
bits.frombytes(listener_pkts[0][20:53])
|
||||
slot_st = stack.hbp.STATUS[2]
|
||||
assert slot_st.get("TX_TA_EMB") is not None
|
||||
assert bits[_EMB_SLICE] == slot_st["REP_EMB_LC"][1]
|
||||
assert listener_pkts[0][20:53] != burst.payload
|
||||
|
||||
|
||||
def test_proxy_attach_binds_sockaddr_used_for_master_ingress() -> None:
|
||||
"""Peer must be registered at proxy client addr so DMRD is accepted and REPEAT fans out."""
|
||||
fanin, stack = _proxy_fanin_stack()
|
||||
ensure_system_runtime_config(stack.config)
|
||||
packet = DeterministicScenario.voice_head_spec(
|
||||
PacketSpec(peer_id=730039210, stream_id=0x55667788)
|
||||
).data()
|
||||
|
||||
fanin.datagramReceived(packet, _ADDR_TX)
|
||||
assert stack.hbp._peers[_PEER_TX]["SOCKADDR"] == _ADDR_TX
|
||||
rx_traffic = stack.transport.for_addr(_ADDR_RX)
|
||||
assert rx_traffic, "REPEAT must reach the other logged-in hotspot"
|
||||
assert any(p[:4] == DMRD for p in rx_traffic)
|
||||
@ -1,91 +0,0 @@
|
||||
"""Live UDP smoke test for integrated proxy (isolated port; no production restart)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import socket
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
from twisted.internet import reactor
|
||||
|
||||
from adn_server.application.proxy import ProxyUseCases
|
||||
from adn_server.domain.value_objects import bytes_4
|
||||
from adn_server.infrastructure.config_normalizer import ensure_system_runtime_config
|
||||
from adn_server.infrastructure.hbp_constants import RPTACK, RPTL
|
||||
from adn_server.infrastructure.proxy import (
|
||||
InMemoryPendingRptoQueue,
|
||||
InMemoryProxySlotStore,
|
||||
InProcessHbpSink,
|
||||
ProxyFanInProtocol,
|
||||
ProxyReplyTransport,
|
||||
)
|
||||
from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol
|
||||
|
||||
_SMOKE_PORT = 62032
|
||||
_PEER = bytes_4(1234567)
|
||||
|
||||
|
||||
class _AclRouter:
|
||||
def acl_check(self, peer_id: bytes, acl: object) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
def _build_fanin() -> ProxyFanInProtocol:
|
||||
config = {
|
||||
"GLOBAL": {"PING_TIME": 10, "MAX_MISSED": 3, "USE_ACL": False},
|
||||
"SYSTEMS": {
|
||||
"HOTSPOT": {
|
||||
"MODE": "MASTER",
|
||||
"ENABLED": True,
|
||||
"MAX_PEERS": 8,
|
||||
"OPTIONS": "TS2=9990;",
|
||||
}
|
||||
},
|
||||
}
|
||||
ensure_system_runtime_config(config)
|
||||
hbp = HBPProtocol("HOTSPOT", config)
|
||||
hbp._router = _AclRouter() # type: ignore[assignment]
|
||||
proxy = ProxyUseCases(
|
||||
InMemoryProxySlotStore(),
|
||||
InMemoryPendingRptoQueue(),
|
||||
max_peers=8,
|
||||
)
|
||||
sink = InProcessHbpSink(hbp)
|
||||
fanin = ProxyFanInProtocol(proxy, sink)
|
||||
return fanin, hbp, sink
|
||||
|
||||
|
||||
@pytest.mark.smoke
|
||||
def test_live_udp_rptl_rptack_on_isolated_port() -> None:
|
||||
"""RPTL in → inject → RPTACK out on 127.0.0.1:62032 (does not use production 62031)."""
|
||||
fanin, hbp, _ = _build_fanin()
|
||||
result: dict[str, bytes | None] = {"reply": None, "error": None}
|
||||
|
||||
def _run_client() -> None:
|
||||
try:
|
||||
client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
client.settimeout(2.0)
|
||||
client.bind(("127.0.0.1", 0))
|
||||
client.sendto(RPTL + _PEER, ("127.0.0.1", _SMOKE_PORT))
|
||||
data, _ = client.recvfrom(4096)
|
||||
result["reply"] = data
|
||||
client.close()
|
||||
except OSError as exc:
|
||||
result["error"] = str(exc).encode()
|
||||
finally:
|
||||
reactor.callFromThread(reactor.stop)
|
||||
|
||||
listener = reactor.listenUDP(_SMOKE_PORT, fanin, interface="127.0.0.1")
|
||||
hbp.transport = ProxyReplyTransport(fanin.transport)
|
||||
reactor.callWhenRunning(
|
||||
lambda: threading.Thread(target=_run_client, daemon=True).start()
|
||||
)
|
||||
reactor.callLater(5.0, reactor.stop)
|
||||
reactor.run()
|
||||
|
||||
listener.stopListening()
|
||||
if result["error"]:
|
||||
pytest.fail(result["error"].decode())
|
||||
reply = result["reply"]
|
||||
assert reply is not None, "no UDP reply (timeout?)"
|
||||
assert reply.startswith(RPTACK), f"expected RPTACK, got {reply[:8]!r}"
|
||||
@ -1,10 +0,0 @@
|
||||
"""Report wire encoder factory."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from adn_server.infrastructure.twisted_adapters.report import ReportWire, create_report_wire
|
||||
|
||||
|
||||
def test_create_report_wire_returns_encoder() -> None:
|
||||
wire = create_report_wire({"REPORTS": {}})
|
||||
assert isinstance(wire, ReportWire)
|
||||
@ -1,174 +0,0 @@
|
||||
"""Tests for scripts/freedmr_cfg_to_yaml.py."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import textwrap
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(_ROOT / "scripts"))
|
||||
|
||||
from freedmr_cfg_to_yaml import dump_yaml, parse_freedmr_cfg # noqa: E402
|
||||
|
||||
|
||||
def _write_cfg(tmp_path: Path, body: str) -> Path:
|
||||
path = tmp_path / "test.cfg"
|
||||
path.write_text(textwrap.dedent(body).strip() + "\n", encoding="utf-8")
|
||||
return path
|
||||
|
||||
|
||||
def test_converts_global_echo_and_obp(tmp_path: Path) -> None:
|
||||
cfg = _write_cfg(
|
||||
tmp_path,
|
||||
"""
|
||||
[GLOBAL]
|
||||
SERVER_ID: 73010
|
||||
USE_ACL: True
|
||||
TGID_TS2_ACL: PERMIT:ALL
|
||||
|
||||
[REPORTS]
|
||||
REPORT: True
|
||||
REPORT_CLIENTS: 127.0.0.1
|
||||
|
||||
[LOGGER]
|
||||
LOG_FILE: /var/log/FreeDMR/FreeDMR.log
|
||||
LOG_NAME: FreeDMR
|
||||
|
||||
[ALIASES]
|
||||
TRY_DOWNLOAD: True
|
||||
STALE_DAYS: 1
|
||||
|
||||
[ECHO]
|
||||
MODE: MASTER
|
||||
ENABLED: True
|
||||
PORT: 54917
|
||||
TS2_STATIC: 9990
|
||||
TGID_TS2_ACL: PERMIT:9990
|
||||
GENERATOR: 0
|
||||
|
||||
[OBP-CR]
|
||||
MODE: OPENBRIDGE
|
||||
ENABLED: False
|
||||
PORT: 62052
|
||||
NETWORK_ID: 71210
|
||||
PASSPHRASE: passw0rd
|
||||
TARGET_IP: freedmrcr.net
|
||||
TARGET_PORT: 62026
|
||||
TGID_ACL: DENY :0-82,9990-9999
|
||||
TGID_TS1_ACL: DENY :0-89
|
||||
PROTO_VER: 5
|
||||
""",
|
||||
)
|
||||
out = parse_freedmr_cfg(cfg)
|
||||
|
||||
assert out["GLOBAL"]["SERVER_ID"] == 73010
|
||||
assert out["GLOBAL"]["TALKER_ALIAS"] is False
|
||||
assert out["REPORTS"]["REPORT_CLIENTS"] == "127.0.0.1"
|
||||
assert out["LOGGER"]["LOG_FILE"] == "/var/log/adn-server/adn-server.log"
|
||||
assert out["LOGGER"]["LOG_NAME"] == "adn-server"
|
||||
assert out["ALIASES"]["KEYS_FILE"] == "keys.json"
|
||||
|
||||
echo = out["SYSTEMS"]["ECHO"]
|
||||
assert echo["MODE"] == "MASTER"
|
||||
assert echo["TS2_STATIC"] == "9990"
|
||||
assert echo["TGID_TS2_ACL"] == "PERMIT:9990"
|
||||
|
||||
obp = out["SYSTEMS"]["OBP-CR"]
|
||||
assert obp["ENABLED"] is False
|
||||
assert obp["TGID_ACL"] == "DENY:0-82,9990-9999"
|
||||
assert obp["TGID_TS1_ACL"] == "DENY:0-89"
|
||||
assert obp["PROTO_VER"] == 5
|
||||
|
||||
|
||||
def test_includes_disabled_systems(tmp_path: Path) -> None:
|
||||
cfg = _write_cfg(
|
||||
tmp_path,
|
||||
"""
|
||||
[OBP-OFF]
|
||||
MODE: OPENBRIDGE
|
||||
ENABLED: False
|
||||
PORT: 62099
|
||||
NETWORK_ID: 1
|
||||
PASSPHRASE: x
|
||||
TARGET_IP: 1.2.3.4
|
||||
TARGET_PORT: 62099
|
||||
""",
|
||||
)
|
||||
out = parse_freedmr_cfg(cfg)
|
||||
assert "OBP-OFF" in out["SYSTEMS"]
|
||||
assert out["SYSTEMS"]["OBP-OFF"]["ENABLED"] is False
|
||||
|
||||
|
||||
def test_preserves_section_and_key_order(tmp_path: Path) -> None:
|
||||
cfg = _write_cfg(
|
||||
tmp_path,
|
||||
"""
|
||||
[GLOBAL]
|
||||
PATH: ./
|
||||
PING_TIME: 10
|
||||
SERVER_ID: 73010
|
||||
|
||||
[REPORTS]
|
||||
REPORT: True
|
||||
|
||||
[ALLSTAR]
|
||||
ENABLED: False
|
||||
|
||||
[ECHO]
|
||||
MODE: MASTER
|
||||
ENABLED: True
|
||||
PORT: 54917
|
||||
TS2_STATIC: 9990
|
||||
|
||||
[OBP-CR]
|
||||
MODE: OPENBRIDGE
|
||||
ENABLED: False
|
||||
PORT: 62052
|
||||
NETWORK_ID: 71210
|
||||
PASSPHRASE: passw0rd
|
||||
TARGET_IP: freedmrcr.net
|
||||
TARGET_PORT: 62026
|
||||
PROTO_VER: 5
|
||||
""",
|
||||
)
|
||||
out = parse_freedmr_cfg(cfg)
|
||||
|
||||
assert list(out.keys()) == ["GLOBAL", "REPORTS", "ALLSTAR", "SYSTEMS"]
|
||||
assert list(out["GLOBAL"].keys())[:3] == ["PATH", "PING_TIME", "SERVER_ID"]
|
||||
assert list(out["SYSTEMS"].keys()) == ["ECHO", "OBP-CR"]
|
||||
assert list(out["SYSTEMS"]["ECHO"].keys())[:4] == ["MODE", "ENABLED", "PORT", "TS2_STATIC"]
|
||||
|
||||
text = dump_yaml(out)
|
||||
yaml_keys = [line.rstrip(":") for line in text.splitlines() if line.endswith(":") and not line.startswith("#")]
|
||||
assert yaml_keys[:4] == ["GLOBAL", "REPORTS", "ALLSTAR", "SYSTEMS"]
|
||||
|
||||
|
||||
def test_fixture_matches_cfg_section_order() -> None:
|
||||
fixture = Path(__file__).resolve().parents[1] / "fixtures" / "sample_freedmr.cfg"
|
||||
out = parse_freedmr_cfg(fixture)
|
||||
expected_top = ["GLOBAL", "REPORTS", "LOGGER", "ALIASES", "ALLSTAR", "SYSTEMS"]
|
||||
assert list(out.keys()) == expected_top
|
||||
assert list(out["SYSTEMS"].keys()) == ["SYSTEM", "ECHO", "OBP-ES", "OBP-UY"]
|
||||
assert list(out["GLOBAL"].keys())[:5] == ["PATH", "PING_TIME", "MAX_MISSED", "USE_ACL", "REG_ACL"]
|
||||
|
||||
|
||||
def test_dump_yaml_is_parseable(tmp_path: Path) -> None:
|
||||
cfg = _write_cfg(
|
||||
tmp_path,
|
||||
"""
|
||||
[GLOBAL]
|
||||
SERVER_ID: 1
|
||||
[SYSTEM]
|
||||
MODE: MASTER
|
||||
ENABLED: True
|
||||
PORT: 56400
|
||||
PASSPHRASE: secret
|
||||
GENERATOR: 1
|
||||
""",
|
||||
)
|
||||
text = dump_yaml(parse_freedmr_cfg(cfg))
|
||||
loaded = yaml.safe_load(text.split("\n", 2)[2])
|
||||
assert loaded["SYSTEMS"]["SYSTEM"]["PORT"] == 56400
|
||||
@ -1,20 +0,0 @@
|
||||
"""Smoke tests for PacketSpec and DMR header parsing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from tests.harness.deterministic import PacketSpec, parse_dmr_fields
|
||||
|
||||
from adn_server.domain import int_id
|
||||
|
||||
|
||||
def test_packet_spec_builds_valid_dmr_header() -> None:
|
||||
spec = PacketSpec(dst_id=91, rf_src=3120001, peer_id=1001, seq=7)
|
||||
packet = spec.data()
|
||||
fields = parse_dmr_fields(packet)
|
||||
|
||||
assert fields["opcode"] == b"DMRD"
|
||||
assert fields["seq"] == 7
|
||||
assert int_id(fields["dst_id"]) == 91
|
||||
assert int_id(fields["rf_src"]) == 3120001
|
||||
assert fields["slot"] == 2
|
||||
assert fields["call_type"] == "group"
|
||||
@ -0,0 +1,128 @@
|
||||
"""HBP MASTER + BridgeUseCases stack for REPEAT / talker-alias integration tests."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import copy
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
from adn_server.application.bridge_use_cases import BridgeUseCases
|
||||
from adn_server.application.reporting_use_cases import ReportingUseCases
|
||||
from adn_server.domain.dmr.bptc import encode_emblc
|
||||
from adn_server.infrastructure.bridge_router_impl import InMemoryBridgeRouter
|
||||
from adn_server.infrastructure.config_normalizer import (
|
||||
apply_talker_alias_defaults,
|
||||
ensure_system_runtime_config,
|
||||
)
|
||||
from adn_server.infrastructure.talker_alias_emblc import default_ta_emblc_encoder
|
||||
from adn_server.infrastructure.twisted_adapters.udp_hbp import HBPProtocol
|
||||
|
||||
from tests.harness.deterministic import FakeReportFactory, FakeReportSender, PacketSpec
|
||||
from tests.harness.scenarios import talker_alias_config
|
||||
|
||||
|
||||
class RecordingTransport:
|
||||
"""Capture MASTER downlink UDP writes (REPEAT and DMRA)."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.sent: list[tuple[bytes, tuple[str, int]]] = []
|
||||
|
||||
def write(self, data: bytes, addr: tuple[str, int]) -> None:
|
||||
self.sent.append((data, addr))
|
||||
|
||||
def for_addr(self, addr: tuple[str, int]) -> list[bytes]:
|
||||
return [pkt for pkt, target in self.sent if target == addr]
|
||||
|
||||
def clear(self) -> None:
|
||||
self.sent.clear()
|
||||
|
||||
|
||||
class _AclRouter:
|
||||
def acl_check(self, _value: bytes, _acl: object) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
@dataclass
|
||||
class HbpRepeatStack:
|
||||
system_name: str
|
||||
config: dict[str, Any]
|
||||
hbp: HBPProtocol
|
||||
bridge: BridgeUseCases
|
||||
transport: RecordingTransport
|
||||
dmra_capture: list[tuple[list[bytes], bytes | None]] = field(default_factory=list)
|
||||
|
||||
def register_peer(self, peer_id: bytes, sockaddr: tuple[str, int]) -> None:
|
||||
self.hbp._peers[peer_id] = {
|
||||
"CONNECTION": "YES",
|
||||
"CONNECTED": 1_700_000_000.0,
|
||||
"LAST_PING": 1_700_000_000.0,
|
||||
"SOCKADDR": sockaddr,
|
||||
"CALLSIGN": b"CE5RPY ",
|
||||
"RADIO_ID": str(int.from_bytes(peer_id, "big")),
|
||||
}
|
||||
self.config["SYSTEMS"][self.system_name].setdefault("PEERS", {})[peer_id] = (
|
||||
self.hbp._peers[peer_id]
|
||||
)
|
||||
|
||||
def inject(self, packet: bytes, sockaddr: tuple[str, int]) -> None:
|
||||
self.hbp.datagramReceived(packet, sockaddr)
|
||||
|
||||
def inject_spec(self, spec: PacketSpec, sockaddr: tuple[str, int]) -> None:
|
||||
self.inject(spec.data(), sockaddr)
|
||||
|
||||
|
||||
def build_hbp_repeat_stack(
|
||||
*,
|
||||
talker_alias: bool = True,
|
||||
system_name: str = "MASTER-A",
|
||||
) -> HbpRepeatStack:
|
||||
"""Real HBPProtocol with BridgeUseCases TA callbacks (not FakeHbpProtocol)."""
|
||||
config = copy.deepcopy(talker_alias_config())
|
||||
if not talker_alias:
|
||||
config["GLOBAL"]["TALKER_ALIAS"] = False
|
||||
apply_talker_alias_defaults(config)
|
||||
ensure_system_runtime_config(config)
|
||||
sys_cfg = config["SYSTEMS"][system_name]
|
||||
sys_cfg["REPEAT"] = True
|
||||
sys_cfg["MAX_PEERS"] = 8
|
||||
sys_cfg["USE_ACL"] = False
|
||||
|
||||
transport = RecordingTransport()
|
||||
hbp = HBPProtocol(system_name, config, router=_AclRouter())
|
||||
hbp.transport = transport # type: ignore[assignment]
|
||||
|
||||
protocols: dict[str, Any] = {system_name: hbp}
|
||||
dmra_capture: list[tuple[list[bytes], bytes | None]] = []
|
||||
|
||||
def _send_dmra(target_system: str, packets: list[bytes], exclude_peer: bytes | None = None) -> int:
|
||||
proto = protocols[target_system]
|
||||
dmra_capture.append((list(packets), exclude_peer))
|
||||
return proto.send_dmra_to_peers(packets, exclude_peer=exclude_peer)
|
||||
|
||||
def _get_dmra_blocks(_system: str, stream_id: bytes) -> dict[int, bytes] | None:
|
||||
return hbp.get_dmra_blocks(stream_id)
|
||||
|
||||
report_factory = FakeReportFactory()
|
||||
bridge = BridgeUseCases(
|
||||
InMemoryBridgeRouter(),
|
||||
config,
|
||||
send_to_system=lambda *_a, **_k: None,
|
||||
get_protocols=lambda: protocols,
|
||||
reporting=ReportingUseCases(FakeReportSender(report_factory), config),
|
||||
send_dmra_to_system=_send_dmra,
|
||||
get_dmra_blocks=_get_dmra_blocks,
|
||||
encode_emblc=encode_emblc,
|
||||
ta_emblc_encoder=default_ta_emblc_encoder,
|
||||
)
|
||||
hbp._on_talker_alias_repeat_prepare = bridge.prepare_talker_alias_local_repeat
|
||||
hbp._on_talker_alias_repeat_burst = bridge.rewrite_repeat_voice_burst
|
||||
hbp._on_talker_alias_stream_end = bridge.clear_talker_alias_stream
|
||||
|
||||
return HbpRepeatStack(
|
||||
system_name=system_name,
|
||||
config=config,
|
||||
hbp=hbp,
|
||||
bridge=bridge,
|
||||
transport=transport,
|
||||
dmra_capture=dmra_capture,
|
||||
)
|
||||
@ -1,39 +0,0 @@
|
||||
"""Talker Alias embedded LC alternation on voice bursts."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from tests.harness.deterministic import DeterministicScenario, PacketSpec, active_bridge
|
||||
from tests.harness.scenarios import talker_alias_config
|
||||
|
||||
|
||||
def test_talker_alias_embed_state_prepared_on_bridge_vhead() -> 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=0xABABABAB)
|
||||
|
||||
scenario.inject_hbp("MASTER-A", DeterministicScenario.voice_head_spec(base))
|
||||
|
||||
ts_st = scenario.protocols["MASTER-B"].STATUS.get(2, {})
|
||||
assert "TX_TA_EMB" in ts_st
|
||||
assert ts_st.get("TX_TA_ON") is False
|
||||
assert ts_st.get("TX_TA_PHASE") == 0
|
||||
|
||||
|
||||
def test_talker_alias_embed_cleared_on_vterm() -> 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=0xCDCDCDCD)
|
||||
|
||||
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),
|
||||
)
|
||||
scenario.inject_hbp(
|
||||
"MASTER-A",
|
||||
DeterministicScenario.voice_term_spec(base, seq=99),
|
||||
)
|
||||
|
||||
ts_st = scenario.protocols["MASTER-B"].STATUS.get(2, {})
|
||||
assert "TX_TA_EMB" not in ts_st
|
||||
assert "TX_TA_ON" not in ts_st
|
||||
Loading…
Reference in new issue