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.
105 lines
3.6 KiB
105 lines
3.6 KiB
# ADN DMR Peer Server - tests application store sync
|
|
#
|
|
# 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
|
|
###############################################################################
|
|
|
|
"""SubscriptionStore sync from BRIDGES."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from adn_server.application.subscription.routing_table_export import export_routing_table
|
|
from adn_server.application.subscription.store_sync import replace_store_from_routing_table
|
|
from adn_server.domain import bytes_3, int_id
|
|
from adn_server.infrastructure.bootstrap.peer_server import _seed_echo_routing_table
|
|
from adn_server.infrastructure.subscription_store import InMemorySubscriptionStore
|
|
|
|
|
|
def _row_fingerprint(row: dict[str, Any]) -> tuple[Any, ...]:
|
|
return (
|
|
row.get("SYSTEM"),
|
|
int(row.get("TS") or 1),
|
|
int_id(row.get("TGID") or b"\x00\x00\x00"),
|
|
bool(row.get("ACTIVE")),
|
|
str(row.get("TO_TYPE", "ON")),
|
|
row.get("TIMEOUT"),
|
|
)
|
|
|
|
|
|
def _bridges_fingerprints(bridges: dict[str, list[dict[str, Any]]]) -> set[tuple[Any, ...]]:
|
|
fps: set[tuple[Any, ...]] = set()
|
|
for rows in bridges.values():
|
|
for row in rows:
|
|
if isinstance(row, dict):
|
|
fps.add(_row_fingerprint(row))
|
|
return fps
|
|
|
|
|
|
def test_replace_store_from_echo_bridges_round_trip():
|
|
config = {
|
|
"SYSTEMS": {
|
|
"ECHO": {"MODE": "PEER"},
|
|
"MASTER-A": {"MODE": "MASTER", "DEFAULT_UA_TIMER": 10},
|
|
"MASTER-B": {"MODE": "MASTER", "DEFAULT_UA_TIMER": 15},
|
|
}
|
|
}
|
|
bridges = _seed_echo_routing_table(config)
|
|
store = InMemorySubscriptionStore()
|
|
replace_store_from_routing_table(store, bridges)
|
|
|
|
assert len(store.snapshot()) == len(_bridges_fingerprints(bridges))
|
|
|
|
now = 1_700_000_000.0
|
|
exported = export_routing_table(store, now=now)
|
|
assert _bridges_fingerprints(bridges) == _bridges_fingerprints(exported)
|
|
|
|
|
|
def test_replace_store_clears_stale_entries():
|
|
store = InMemorySubscriptionStore()
|
|
bridges_a = {
|
|
"100": [
|
|
{
|
|
"SYSTEM": "SYS-A",
|
|
"TS": 1,
|
|
"TGID": bytes_3(100),
|
|
"ACTIVE": False,
|
|
"TIMEOUT": 600.0,
|
|
"TO_TYPE": "ON",
|
|
}
|
|
]
|
|
}
|
|
bridges_b = {
|
|
"200": [
|
|
{
|
|
"SYSTEM": "SYS-B",
|
|
"TS": 2,
|
|
"TGID": bytes_3(200),
|
|
"ACTIVE": True,
|
|
"TIMEOUT": 600.0,
|
|
"TO_TYPE": "ON",
|
|
}
|
|
]
|
|
}
|
|
replace_store_from_routing_table(store, bridges_a)
|
|
assert len(store.snapshot()) == 1
|
|
replace_store_from_routing_table(store, bridges_b)
|
|
assert len(store.snapshot()) == 1
|
|
(sub,) = store.snapshot()
|
|
assert int(sub.channel.tgid) == 200
|