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.
371 lines
13 KiB
371 lines
13 KiB
# ADN DMR Peer Server - tests domain mesh session
|
|
#
|
|
# 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
|
|
###############################################################################
|
|
|
|
"""Live OpenBridge session state: peer address, keepalive and source quench."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from adn_server.domain import bytes_3, bytes_4
|
|
from adn_server.domain.mesh_session import (
|
|
MeshSessionStore,
|
|
ObpBridgeSession,
|
|
mesh_sessions,
|
|
obp_session,
|
|
)
|
|
|
|
_CONFIGURED = ("82.65.127.86", 62201)
|
|
_ELSEWHERE = ("85.241.222.7", 62268)
|
|
_NOW = 1_800_000_000.0
|
|
|
|
|
|
def _session(configured: tuple = _CONFIGURED) -> ObpBridgeSession:
|
|
return ObpBridgeSession(system_name="OBP-FR", configured_peer=configured)
|
|
|
|
|
|
def _config(**overrides) -> dict:
|
|
sys_cfg = {
|
|
"MODE": "OPENBRIDGE",
|
|
"ENABLED": True,
|
|
"TARGET_IP": _CONFIGURED[0],
|
|
"TARGET_PORT": _CONFIGURED[1],
|
|
"TARGET_SOCK": _CONFIGURED,
|
|
}
|
|
sys_cfg.update(overrides)
|
|
return {"SYSTEMS": {"OBP-FR": sys_cfg}}
|
|
|
|
|
|
# --- peer anchored to DNS -----------------------------------------------------
|
|
|
|
|
|
def _dns_session() -> ObpBridgeSession:
|
|
return ObpBridgeSession(
|
|
system_name="OBP-FR", configured_peer=_CONFIGURED, dns_host="peer.example.net"
|
|
)
|
|
|
|
|
|
def test_a_dns_anchored_peer_ignores_what_the_wire_says() -> None:
|
|
"""The name is the identity: a host that holds the passphrase but is not what
|
|
the name resolves to cannot take the link over."""
|
|
session = _dns_session()
|
|
assert session.dns_anchored
|
|
assert session.learn_peer(_ELSEWHERE, at=_NOW) is False
|
|
assert session.peer == _CONFIGURED
|
|
|
|
|
|
def test_a_name_that_never_resolved_anchors_nothing() -> None:
|
|
"""normalize_obp_config leaves TARGET_SOCK as (None, port) when the name does not
|
|
resolve at startup. Anchoring on that would take the link off the air for good."""
|
|
session = ObpBridgeSession(
|
|
system_name="OBP-FR", configured_peer=(None, 62201), dns_host="peer.example.net"
|
|
)
|
|
assert session.dns_anchored is False
|
|
assert session.learn_peer(_ELSEWHERE, at=_NOW) is True
|
|
|
|
|
|
def test_a_dns_anchored_peer_takes_a_new_port_on_its_own_host() -> None:
|
|
"""The name pins the host. A peer answers from whatever socket it bound, which
|
|
NAT may rewrite and which needs not be the port we send to."""
|
|
session = _dns_session()
|
|
other_port = (_CONFIGURED[0], 57933)
|
|
assert session.learn_peer(other_port, at=_NOW) is True
|
|
assert session.peer == other_port
|
|
|
|
|
|
def test_resolving_elsewhere_drops_a_port_learned_on_the_old_host() -> None:
|
|
session = _dns_session()
|
|
session.learn_peer((_CONFIGURED[0], 57933), at=_NOW)
|
|
assert session.adopt_resolved(_ELSEWHERE, at=_NOW) is True
|
|
assert session.peer == _ELSEWHERE
|
|
assert session.learned_peer is None
|
|
|
|
|
|
def test_a_dns_anchored_peer_moves_when_the_name_resolves_elsewhere() -> None:
|
|
session = _dns_session()
|
|
assert session.adopt_resolved(_ELSEWHERE, at=_NOW) is True
|
|
assert session.peer == _ELSEWHERE
|
|
assert session.dns_checked_at == _NOW
|
|
|
|
|
|
def test_resolving_to_the_same_address_is_not_a_move() -> None:
|
|
session = _dns_session()
|
|
assert session.adopt_resolved(_CONFIGURED, at=_NOW) is False
|
|
assert session.peer == _CONFIGURED
|
|
assert session.dns_checked_at == _NOW
|
|
|
|
|
|
def test_a_target_written_as_an_address_is_not_dns_anchored() -> None:
|
|
"""Only a name can be re-resolved; a literal address keeps the old behaviour."""
|
|
store = MeshSessionStore()
|
|
session = store.session("OBP-FR", {"_TARGET_IP": _CONFIGURED[0], "TARGET_SOCK": _CONFIGURED})
|
|
assert session.dns_anchored is False
|
|
assert session.learn_peer(_ELSEWHERE, at=_NOW) is True
|
|
|
|
|
|
def test_a_target_written_as_a_name_is_dns_anchored() -> None:
|
|
store = MeshSessionStore()
|
|
session = store.session(
|
|
"OBP-FR", {"_TARGET_IP": "peer.example.net", "TARGET_SOCK": _CONFIGURED}
|
|
)
|
|
assert session.dns_host == "peer.example.net"
|
|
assert session.peer == _CONFIGURED
|
|
|
|
|
|
# --- peer address ------------------------------------------------------------
|
|
|
|
|
|
def test_a_fresh_session_sends_to_the_configured_peer() -> None:
|
|
session = _session()
|
|
assert session.peer == _CONFIGURED
|
|
assert session.peer_known
|
|
|
|
|
|
def test_learning_a_peer_moves_egress_but_not_the_configuration() -> None:
|
|
session = _session()
|
|
assert session.learn_peer(_ELSEWHERE, at=_NOW) is True
|
|
assert session.peer == _ELSEWHERE
|
|
assert session.configured_peer == _CONFIGURED
|
|
assert session.learned_at == _NOW
|
|
|
|
|
|
def test_learning_the_same_address_twice_reports_no_move() -> None:
|
|
session = _session()
|
|
assert session.learn_peer(_CONFIGURED, at=_NOW) is False
|
|
session.learn_peer(_ELSEWHERE, at=_NOW)
|
|
assert session.learn_peer(_ELSEWHERE, at=_NOW + 1) is False
|
|
assert session.learned_at == _NOW
|
|
|
|
|
|
def test_an_empty_address_is_not_learned() -> None:
|
|
session = _session()
|
|
assert session.learn_peer(("", 0), at=_NOW) is False
|
|
assert session.learned_peer is None
|
|
|
|
|
|
def test_forgetting_falls_back_to_the_configured_peer() -> None:
|
|
session = _session()
|
|
session.learn_peer(_ELSEWHERE, at=_NOW)
|
|
session.forget_learned_peer()
|
|
assert session.peer == _CONFIGURED
|
|
|
|
|
|
def test_a_peer_that_never_resolved_is_not_known() -> None:
|
|
assert _session((None, 62201)).peer_known is False
|
|
|
|
|
|
# --- keepalive ---------------------------------------------------------------
|
|
|
|
|
|
def test_keepalive_starts_unseen() -> None:
|
|
session = _session()
|
|
assert session.keepalive_seen is False
|
|
assert session.keepalive_age(_NOW) is None
|
|
assert session.keepalive_stale(_NOW) is False # never seen is not the same as stale
|
|
assert session.keepalive_ok(_NOW) is False
|
|
|
|
|
|
def test_a_fresh_keepalive_is_ok_and_an_old_one_is_stale() -> None:
|
|
session = _session()
|
|
session.note_keepalive(_NOW - 10)
|
|
assert session.keepalive_ok(_NOW) is True
|
|
assert session.keepalive_stale(_NOW) is False
|
|
assert session.keepalive_age(_NOW) == 10
|
|
session.note_keepalive(_NOW - 61)
|
|
assert session.keepalive_stale(_NOW) is True
|
|
assert session.keepalive_ok(_NOW) is False
|
|
|
|
|
|
def test_the_keepalive_timeout_can_be_narrowed() -> None:
|
|
session = _session()
|
|
session.note_keepalive(_NOW - 30)
|
|
assert session.keepalive_stale(_NOW, timeout=10) is True
|
|
|
|
|
|
# --- source quench -----------------------------------------------------------
|
|
|
|
|
|
def test_a_quenched_stream_is_reported_for_its_talkgroup() -> None:
|
|
session = _session()
|
|
session.quench(bytes_3(214), b"strm")
|
|
assert session.quenches(bytes_3(214), b"strm") is True
|
|
assert session.quenches(bytes_3(214), b"other") is False
|
|
assert session.quenches(bytes_3(215), b"strm") is False
|
|
|
|
|
|
def test_a_talkgroup_matches_whatever_width_it_arrives_in() -> None:
|
|
"""TGs travel as 3 bytes here and 4 bytes elsewhere; the number is the same."""
|
|
session = _session()
|
|
session.quench(bytes_4(214), b"strm")
|
|
assert session.quenches(bytes_3(214), b"strm") is True
|
|
|
|
|
|
def test_nothing_is_quenched_on_a_fresh_session() -> None:
|
|
assert _session().quenches(bytes_3(214), b"strm") is False
|
|
|
|
|
|
def test_a_finished_stream_releases_its_quench() -> None:
|
|
session = _session()
|
|
session.quench(bytes_3(214), b"strm")
|
|
session.quench(bytes_3(215), b"other")
|
|
session.release_stream(b"strm")
|
|
assert session.quenches(bytes_3(214), b"strm") is False
|
|
assert session.quenches(bytes_3(215), b"other") is True
|
|
|
|
|
|
def test_a_bridge_stunned_by_its_peer_stays_stunned() -> None:
|
|
session = _session()
|
|
assert session.stunned is False
|
|
session.stun()
|
|
assert session.stunned is True
|
|
|
|
|
|
# --- the store ---------------------------------------------------------------
|
|
|
|
|
|
def test_a_session_is_created_from_the_system_configuration() -> None:
|
|
store = MeshSessionStore()
|
|
session = store.session("OBP-FR", _config()["SYSTEMS"]["OBP-FR"])
|
|
assert session.configured_peer == _CONFIGURED
|
|
assert store.session("OBP-FR") is session
|
|
|
|
|
|
def test_a_session_falls_back_to_target_ip_and_port() -> None:
|
|
store = MeshSessionStore()
|
|
cfg = {"TARGET_IP": "10.0.0.1", "TARGET_PORT": "62044"}
|
|
assert store.session("OBP-XX", cfg).configured_peer == ("10.0.0.1", 62044)
|
|
|
|
|
|
def test_an_unusable_target_port_falls_back_to_the_default() -> None:
|
|
store = MeshSessionStore()
|
|
assert store.session("OBP-XX", {"TARGET_IP": "10.0.0.1", "TARGET_PORT": "?"}).configured_peer == (
|
|
"10.0.0.1",
|
|
62044,
|
|
)
|
|
|
|
|
|
def test_a_quench_key_that_is_not_a_talkgroup_is_ignored() -> None:
|
|
session = _session()
|
|
session.quenched[b"xx"] = b"strm" # short key, as a bad peer could send
|
|
assert session.quenches(bytes_3(214), b"strm") is False
|
|
|
|
|
|
def test_sync_registers_enabled_openbridges_only() -> None:
|
|
config = _config()
|
|
config["SYSTEMS"]["HOTSPOT"] = {"MODE": "MASTER", "ENABLED": True}
|
|
config["SYSTEMS"]["OBP-OFF"] = {"MODE": "OPENBRIDGE", "ENABLED": False}
|
|
store = MeshSessionStore()
|
|
store.sync(config)
|
|
assert "OBP-FR" in store
|
|
assert "HOTSPOT" not in store
|
|
assert "OBP-OFF" not in store
|
|
assert len(store) == 1
|
|
|
|
|
|
def test_sync_anchors_a_session_it_creates() -> None:
|
|
"""Missing it here loses DNS anchoring silently: nothing else derives it."""
|
|
config = _config(_TARGET_IP="peer.example.net")
|
|
store = MeshSessionStore()
|
|
store.sync(config)
|
|
assert store.session("OBP-FR").dns_host == "peer.example.net"
|
|
|
|
|
|
def test_sync_anchors_a_session_that_already_existed() -> None:
|
|
config = _config()
|
|
store = MeshSessionStore()
|
|
store.sync(config)
|
|
assert store.session("OBP-FR").dns_host is None
|
|
|
|
config["SYSTEMS"]["OBP-FR"]["_TARGET_IP"] = "peer.example.net"
|
|
store.sync(config)
|
|
assert store.session("OBP-FR").dns_host == "peer.example.net"
|
|
|
|
|
|
def test_a_reload_that_moves_the_peer_wins_over_what_was_learned() -> None:
|
|
"""Editing TARGET_IP in the YAML is the way out of a bad learned address."""
|
|
config = _config()
|
|
store = MeshSessionStore()
|
|
store.sync(config)
|
|
store.session("OBP-FR").learn_peer(_ELSEWHERE, at=_NOW)
|
|
|
|
config["SYSTEMS"]["OBP-FR"]["TARGET_SOCK"] = ("44.31.61.66", 62032)
|
|
store.sync(config)
|
|
session = store.session("OBP-FR")
|
|
assert session.configured_peer == ("44.31.61.66", 62032)
|
|
assert session.learned_peer is None
|
|
assert session.peer == ("44.31.61.66", 62032)
|
|
|
|
|
|
def test_a_reload_that_changes_nothing_keeps_what_was_learned() -> None:
|
|
config = _config()
|
|
store = MeshSessionStore()
|
|
store.sync(config)
|
|
store.session("OBP-FR").learn_peer(_ELSEWHERE, at=_NOW)
|
|
store.session("OBP-FR").note_keepalive(_NOW)
|
|
|
|
store.sync(config)
|
|
session = store.session("OBP-FR")
|
|
assert session.peer == _ELSEWHERE
|
|
assert session.last_keepalive == _NOW
|
|
|
|
|
|
def test_a_bridge_removed_from_the_config_loses_its_session() -> None:
|
|
config = _config()
|
|
store = MeshSessionStore()
|
|
store.sync(config)
|
|
config["SYSTEMS"].pop("OBP-FR")
|
|
store.sync(config)
|
|
assert "OBP-FR" not in store
|
|
|
|
|
|
def test_dropping_a_session_by_hand() -> None:
|
|
store = MeshSessionStore()
|
|
store.session("OBP-FR")
|
|
store.drop("OBP-FR")
|
|
assert store.get("OBP-FR") is None
|
|
|
|
|
|
def test_the_store_lives_beside_the_other_runtime_tables() -> None:
|
|
config = _config()
|
|
store = mesh_sessions(config)
|
|
assert config["_MESH_SESSIONS"] is store
|
|
assert mesh_sessions(config) is store
|
|
assert store.get("OBP-FR") is not None
|
|
|
|
|
|
def test_one_session_per_system_from_the_server_config() -> None:
|
|
config = _config()
|
|
session = obp_session(config, "OBP-FR")
|
|
assert session.configured_peer == _CONFIGURED
|
|
assert obp_session(config, "OBP-FR") is session
|
|
|
|
|
|
def test_asking_for_an_unknown_system_creates_an_empty_session() -> None:
|
|
"""Callers ask by name; a system with no OBP config simply has no peer."""
|
|
session = obp_session(_config(), "NOPE")
|
|
assert session.peer_known is False
|
|
assert session.keepalive_seen is False
|
|
|
|
|
|
def test_refused_frames_are_tallied_by_reason() -> None:
|
|
session = _session()
|
|
session.count_drop("tg-filter")
|
|
session.count_drop("tg-filter")
|
|
session.count_drop("max-hops")
|
|
assert session.drops == {"tg-filter": 2, "max-hops": 1}
|