potato-mesh/tests/test_reticulum_diagnostics_unit.py

# Copyright © 2025-26 l5yth & contributors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Unit tests for :mod:`data.mesh_ingestor.protocols.reticulum_diagnostics`.

The stack role, the privacy-filtered stack facts the connect line prints, the
thread-safe announce tally, the hourly summary with its warning and hints, its
rate limit, and the provider's wiring of all four (SPEC RG1-RG4).  The module
is reached through the provider module (the :func:`diag` fixture), as the
end-to-end file reaches it.
"""

from __future__ import annotations

import sys
import threading
import time
import types
from pathlib import Path

import pytest

REPO_ROOT = Path(__file__).resolve().parents[1]
if str(REPO_ROOT) not in sys.path:  # pragma: no cover - conftest adds it
    sys.path.insert(0, str(REPO_ROOT))

from RNS.Interfaces.AutoInterface import (  # noqa: E402 - path setup
    AutoInterface,
    AutoInterfacePeer,
)
from RNS.Interfaces.LocalInterface import (  # noqa: E402 - path setup
    LocalClientInterface,
    LocalServerInterface,
)
from RNS.Interfaces.RNodeInterface import RNodeInterface  # noqa: E402

from data.mesh_ingestor import config  # noqa: E402 - path setup
from data.mesh_ingestor.handlers import _state as handler_state  # noqa: E402
from data.mesh_ingestor.protocols import reticulum as mod  # noqa: E402
from test_reticulum_diagnostics import (  # noqa: E402 - shared stack doubles
    PEER,
    PEER_DEST,
    REJECTED,
    RNODE_DEAF,
    RNODE_OK,
    SHARED,
    _STATE_LINE as _STATE,
    _SUMMARY_LINE as _SUMMARY,
    _Stack,
    _client_socket,
)
from test_reticulum_scope_unit import _bare  # noqa: E402 - shared RNS doubles
from test_reticulum_unit import (  # noqa: E402 - shared RNS doubles
    _fake_rns,
    _no_ingestor_node_id,
)

_FAILED = "Reticulum diagnostics failed"
"""Message of a diagnostics read that raised (SPEC RG4)."""

_LAN = "AutoInterfacePeer[eth0/fe80::1]"
"""A LAN peer's printed name."""


@pytest.fixture
def diag():
    """Return the diagnostics module, through the provider module that imports it."""
    return mod.reticulum_diagnostics


@pytest.fixture
def logged(monkeypatch):
    """Capture every ``config._debug_log`` call as ``(severity, message, fields)``.

    ``fields`` keeps ``context`` and drops ``severity``.
    """
    calls: list = []

    def _log(message, **fields):
        severity = fields.pop("severity", "debug")
        calls.append((severity, message, fields))

    monkeypatch.setattr(config, "_debug_log", _log)
    return calls


def _messages(logged) -> list[str]:
    """Return the message of every captured line, in order."""
    return [message for _severity, message, _fields in logged]


def _only(logged, message: str) -> tuple[str, dict]:
    """Return the severity and fields of the one captured *message* line."""
    found = [(sev, fields) for sev, msg, fields in logged if msg == message]
    assert len(found) == 1, f"{len(found)} {message!r} lines"
    return found[0]


class _Clock:
    """A settable monotonic clock."""

    def __init__(self) -> None:
        """Start at zero."""
        self.now = 0.0

    def __call__(self) -> float:
        """Return the current reading."""
        return self.now


def _hour(diag, stack, local, record=None):
    """Connect, let *record* change the stack and tally announces, then summarise.

    Parameters:
        diag: The diagnostics module.
        stack: The instance both reads see.
        local: What ``RNS.Transport.interfaces`` holds.
        record: Called with the diagnostics object inside the hour.

    Returns:
        The :class:`StackDiagnostics` after its first summary.
    """
    clock = _Clock()
    diagnostics = diag.StackDiagnostics(lambda: local, clock=clock)
    diagnostics.connected(stack)
    if record is not None:
        record(diagnostics)
    clock.now = diag.SUMMARY_INTERVAL_SECS
    diagnostics.tick(stack)
    return diagnostics


# ---------------------------------------------------------------------------
# The stack role and what reached this process (SPEC RG1)
# ---------------------------------------------------------------------------


@pytest.mark.parametrize("role", ["client", "shared", "standalone"])
def test_stack_role_reads_the_three_rns_flags(diag, role):
    """A client of rnsd, the shared instance itself, or a standalone stack."""
    assert diag.stack_role(_Stack(role=role)) == role


def test_stack_role_is_unknown_without_a_set_flag(diag):
    """A half-built instance sets no flag, a fake lacks them, and only ``True`` counts."""
    assert diag.stack_role(None) == "unknown"
    assert diag.stack_role(object()) == "unknown"
    assert diag.stack_role(_Stack(role="none")) == "unknown"
    truthy = types.SimpleNamespace(is_connected_to_shared_instance=1)
    assert diag.stack_role(truthy) == "unknown"


def test_received_is_what_rnsd_forwarded_to_a_client(diag):
    """A client's one interface is its socket to rnsd; its ``arxc`` is the count."""
    socket = _client_socket()
    socket.arxc = 7
    assert diag.received_announces([socket]) == 7


def test_received_counts_an_own_stack_once_per_announce(diag):
    """A parent is left out: some count their spawned interfaces' announces again."""
    auto = _bare(AutoInterface, arxc=9)
    peers = [
        _bare(AutoInterfacePeer, arxc=4, parent_interface=auto),
        _bare(AutoInterfacePeer, arxc=5, parent_interface=auto),
    ]
    server = _bare(LocalServerInterface, arxc=0)
    app = _bare(LocalClientInterface, arxc=2, parent_interface=server)
    rnode = _bare(RNodeInterface, arxc=3)
    assert diag.received_announces([auto, *peers, server, app, rnode]) == 14


def test_received_is_unknown_without_an_interface_list(diag):
    """No list is unknown; a missing or non-integer ``arxc`` counts nothing."""
    assert diag.received_announces(None) is None
    assert diag.received_announces("LocalInterface[rns/default]") is None
    odd = [
        _bare(LocalClientInterface),
        types.SimpleNamespace(arxc="7"),
        types.SimpleNamespace(arxc=True),
        types.SimpleNamespace(arxc=-3),
    ]
    assert diag.received_announces(odd) == 0


# ---------------------------------------------------------------------------
# The connect line: classes and counts (SPEC RG1)
# ---------------------------------------------------------------------------


def test_the_connect_line_summarises_classes_and_counts(diag):
    """Classes with counts, the RNode's counters and IFAC flag, the shared clients."""
    socket = _client_socket()
    socket.arxc = 3
    lan = {"name": _LAN, "type": "AutoInterfacePeer"}
    facts = diag.read_facts(_Stack(interfaces=[SHARED, RNODE_DEAF, lan]), [socket])
    assert diag.connect_fields(facts) == {
        "role": "client",
        "stats": "ok",
        "received": 3,
        "interfaces": {
            "AutoInterfacePeer": 1,
            "LocalServerInterface": 1,
            "RNodeInterface": 1,
        },
        "shared_clients": 2,
        "rnode_online": "1/1",
        "rnode_ifac": "off",
        "rnode_rxb": 4_950_000,
        "rnode_txb": 0,
        "rnode_arxc": 0,
        "rnode_protocol_violations": 2760,
        "rnode_ifac_violations": 12135,
        "rnode_filter_hits": 2012,
    }
    assert facts.classes[_LAN] == "AutoInterfacePeer"  # kept for the tally, not logged


def test_rnode_figures_count_each_radio_once(diag):
    """A multi-radio RNode counts its sub-interfaces; the device is not counted again."""
    multi = {
        "name": "RNodeMultiInterface[Multi]",
        "type": "RNodeMultiInterface",
        "status": True,
        "txb": 300,
    }
    low = dict(RNODE_OK, name="Multi[868]", type="RNodeSubInterface", txb=100)
    high = dict(low, name="Multi[433]", status=False, txb=200, ifac_size=8)
    fields = diag.connect_fields(
        diag.read_facts(_Stack(interfaces=[multi, low, high]), None)
    )
    assert fields["interfaces"] == {"RNodeMultiInterface": 1, "RNodeSubInterface": 2}
    assert (fields["rnode_online"], fields["rnode_ifac"]) == ("1/2", "mixed")
    assert (fields["rnode_txb"], fields["rnode_arxc"]) == (300, 80)
    assert "shared_clients" not in fields  # no shared instance listed
    assert fields["received"] is None  # no interface list to read


def test_ifac_is_on_when_every_radio_sets_a_size(diag):
    """``ifac_size`` alone says it; the network name is never read."""
    rnode = dict(RNODE_OK, ifac_size=8, ifac_netname="never-logged")
    fields = diag.connect_fields(diag.read_facts(_Stack(interfaces=[rnode]), []))
    assert fields["rnode_ifac"] == "on"
    assert "never-logged" not in repr(fields)


def test_a_stack_without_radios_has_no_rnode_figures(diag):
    """No RNode listed: the classes say so, and no RNode figure is invented."""
    tcp = {"name": "TCPInterface[Hub]", "type": "TCPClientInterface", "rxb": 9}
    fields = diag.connect_fields(diag.read_facts(_Stack(interfaces=[tcp]), []))
    assert fields == {
        "role": "client",
        "stats": "ok",
        "received": 0,
        "interfaces": {"TCPClientInterface": 1},
    }


def test_a_refused_stats_rpc_reports_its_class_only(diag):
    """The class names the failure; the message and the interfaces are not logged."""
    facts = diag.read_facts(_Stack(stats_error=REJECTED), [_client_socket()])
    fields = diag.connect_fields(facts)
    assert fields == {"role": "client", "stats": "AuthenticationError", "received": 0}
    assert facts.classes is None


@pytest.mark.parametrize(
    "instance",
    [None, object(), types.SimpleNamespace(get_interface_stats=None)],
    ids=["no_instance", "no_members", "not_callable"],
)
def test_a_stack_without_stats_is_unavailable(diag, instance):
    """Fakes and half-built instances lack the RNS members; ``getattr`` defaults apply."""
    fields = diag.connect_fields(diag.read_facts(instance, None))
    assert fields == {"role": "unknown", "stats": "unavailable", "received": None}


@pytest.mark.parametrize("reply", [None, [], {}, {"interfaces": "eth0"}])
def test_a_reply_without_an_interface_list_lists_none(diag, reply):
    """A read that answers with no list still succeeded: ``ok``, no classes."""
    instance = types.SimpleNamespace(get_interface_stats=lambda: reply)
    fields = diag.connect_fields(diag.read_facts(instance, []))
    assert (fields["stats"], fields["interfaces"]) == ("ok", {})


def test_malformed_entries_count_as_unknown_and_bad_counters_as_zero(diag):
    """Entries that are no mapping are skipped; a mapping without a class is unknown."""
    rnode = dict(RNODE_DEAF, rxb="4950000", arxc=None, txb=True, status="yes")
    reply = {"interfaces": [None, "eth0", {"name": "y"}, dict(rnode, ifac_size="8")]}
    instance = types.SimpleNamespace(get_interface_stats=lambda: reply)
    fields = diag.connect_fields(diag.read_facts(instance, []))
    assert fields["interfaces"] == {"RNodeInterface": 1, "unknown": 1}
    assert (fields["rnode_online"], fields["rnode_ifac"]) == ("0/1", "off")
    counters = (fields["rnode_rxb"], fields["rnode_txb"], fields["rnode_arxc"])
    assert counters == (0, 0, 0)


# ---------------------------------------------------------------------------
# The announce tally (SPEC RG2)
# ---------------------------------------------------------------------------


def test_the_tally_counts_each_outcome_by_aspect_and_reason(diag):
    """Delivered and admitted per aspect, drops per reason, scope drops per class."""
    tally = diag.AnnounceTally()
    tally.use_classes({_LAN: "AutoInterfacePeer"})
    tally.admitted("lxmf.delivery")
    tally.admitted("nomadnetwork.node")
    tally.out_of_scope("lxmf.delivery", _LAN)
    tally.out_of_scope("lxmf.delivery", "TCPInterface[Hub]")  # not in the map
    tally.out_of_scope("lxmf.propagation", None)
    tally.out_of_scope("lxmf.propagation", "")
    tally.dropped("lxmf.delivery", "unusable_hash")
    tally.dropped("nomadnetwork.node", "error")
    counts = tally.drain()
    assert counts.delivered == {
        "lxmf.delivery": 4,
        "lxmf.propagation": 2,
        "nomadnetwork.node": 2,
    }
    assert counts.admitted == {"lxmf.delivery": 1, "nomadnetwork.node": 1}
    assert counts.dropped == {
        "error": 1,
        "no_interface": 2,
        "scope": 2,
        "unusable_hash": 1,
    }
    assert counts.scope_classes == {"AutoInterfacePeer": 1, "unknown": 1}
    assert tally.drain() == diag.TallyCounts({}, {}, {}, {})  # drained


def test_the_tally_counts_every_announce_from_many_threads(diag):
    """RNS runs each handler on a thread of its own; no count is lost."""
    tally = diag.AnnounceTally()

    def work() -> None:
        for _ in range(2000):
            tally.admitted("lxmf.delivery")
            tally.out_of_scope("nomadnetwork.node", None)

    threads = [threading.Thread(target=work) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    counts = tally.drain()
    assert counts.delivered == {"lxmf.delivery": 16000, "nomadnetwork.node": 16000}
    assert counts.dropped == {"no_interface": 16000}


# ---------------------------------------------------------------------------
# The hourly summary, its warning and hints (SPEC RG2, RG3)
# ---------------------------------------------------------------------------


def test_an_hour_with_admitted_announces_is_info(diag, logged):
    """Counts per aspect and reason, the received delta and the RNode deltas."""
    socket = _client_socket()
    stack = _Stack(interfaces=[SHARED, RNODE_OK])

    def record(diagnostics) -> None:
        socket.arxc += 3
        stack.interfaces[1].update(rxb=RNODE_OK["rxb"] + 600, arxc=43, txb=12)
        diagnostics.tally.admitted("lxmf.delivery")
        diagnostics.tally.admitted("lxmf.delivery")
        diagnostics.tally.out_of_scope("lxmf.delivery", "AutoInterfacePeer[x]")

    _hour(diag, stack, [socket], record)
    assert _messages(logged) == [_STATE, _SUMMARY]
    assert _only(logged, _SUMMARY) == (
        "info",
        {
            "context": "reticulum.summary",
            "delivered": {"lxmf.delivery": 3},
            "admitted": {"lxmf.delivery": 2},
            "dropped": {"scope": 1},
            "scope_classes": {"unknown": 1},
            "received": 3,
            "stats": "ok",
            "rnode_rxb": 600,
            "rnode_txb": 12,
            "rnode_arxc": 3,
            "rnode_protocol_violations": 0,
            "rnode_ifac_violations": 0,
            "rnode_filter_hits": 0,
        },
    )
    assert _only(logged, _STATE)[0] == "info"


def test_an_hour_with_nothing_received_warns_with_hints(diag, logged, monkeypatch):
    """``received`` 0: a warning, with the hints the connect facts call for."""
    monkeypatch.setattr(handler_state, "_host_node_id", None)
    _hour(diag, _Stack(interfaces=[SHARED, RNODE_DEAF]), [_client_socket()])
    severity, fields = _only(logged, _SUMMARY)
    assert (severity, fields["received"], fields["delivered"]) == ("warn", 0, {})
    assert fields["hints"] == [diag.HINT_RNODE, diag.HINT_PATHS, diag.HINT_HOST_ID]


def test_an_hour_the_scope_dropped_entirely_warns(diag, logged, monkeypatch):
    """Scope and no-interface drops together: the scope dropped all it was given."""
    monkeypatch.setattr(handler_state, "_host_node_id", "!27716218")
    socket = _client_socket()

    def record(diagnostics) -> None:
        socket.arxc += 3
        diagnostics.tally.out_of_scope("lxmf.delivery", _LAN)
        diagnostics.tally.out_of_scope("lxmf.delivery", None)
        diagnostics.tally.out_of_scope("nomadnetwork.node", _LAN)

    lan = {"name": _LAN, "type": "AutoInterfacePeer"}
    _hour(diag, _Stack(interfaces=[SHARED, RNODE_OK, lan]), [socket], record)
    severity, fields = _only(logged, _SUMMARY)
    assert (severity, fields["received"]) == ("warn", 3)
    assert fields["dropped"] == {"no_interface": 1, "scope": 2}
    assert fields["scope_classes"] == {"AutoInterfacePeer": 2}
    assert fields["hints"] == [diag.HINT_PATHS]  # the host id is registered


def test_an_hour_with_other_drops_stays_info(diag, logged):
    """An unusable hash or an error is no scope drop and warns on its own line."""
    socket = _client_socket()

    def record(diagnostics) -> None:
        socket.arxc += 2
        diagnostics.tally.out_of_scope("lxmf.delivery", _LAN)
        diagnostics.tally.dropped("lxmf.delivery", "unusable_hash")

    _hour(diag, _Stack(interfaces=[SHARED, RNODE_OK]), [socket], record)
    severity, fields = _only(logged, _SUMMARY)
    assert severity == "info"
    assert "hints" not in fields


def test_an_unreadable_stack_never_warns_by_itself(diag, logged):
    """No stats, no interface list: the lines say so, at info."""
    _hour(diag, object(), None)
    assert _only(logged, _STATE) == (
        "info",
        {
            "context": "reticulum.connect",
            "role": "unknown",
            "stats": "unavailable",
            "received": None,
        },
    )
    severity, fields = _only(logged, _SUMMARY)
    assert (severity, fields["received"], fields["stats"]) == (
        "info",
        None,
        "unavailable",
    )
    assert "rnode_rxb" not in fields


def test_rnode_deltas_count_a_reset_counter_from_zero(diag, logged):
    """rnsd restarted inside the hour: a counter below its last reading counts anew."""
    socket = _client_socket()
    stack = _Stack(interfaces=[SHARED, dict(RNODE_OK, rxb=1000, arxc=40)])

    def record(_diagnostics) -> None:
        socket.arxc += 1
        stack.interfaces[1].update(rxb=1600, arxc=4)

    _hour(diag, stack, [socket], record)
    _severity, fields = _only(logged, _SUMMARY)
    assert (fields["rnode_rxb"], fields["rnode_arxc"]) == (600, 4)


def test_rnode_deltas_need_radios_in_both_reads(diag, logged, monkeypatch):
    """A refused read at connect leaves the first hour without RNode deltas."""
    monkeypatch.setattr(handler_state, "_host_node_id", "!27716218")
    stack = _Stack(interfaces=[SHARED, RNODE_OK], stats_error=REJECTED)

    def record(_diagnostics) -> None:
        stack.stats_error = None

    _hour(diag, stack, [_client_socket()], record)
    severity, fields = _only(logged, _SUMMARY)
    assert (severity, fields["stats"]) == ("warn", "ok")
    assert not any(key.startswith("rnode_") for key in fields)
    # The hints follow the connect facts, where the RPC was refused.
    assert fields["hints"] == [diag.HINT_STATS, diag.HINT_PATHS]


@pytest.mark.parametrize(
    ("stack", "pending", "expected"),
    [
        (_Stack(interfaces=[SHARED, RNODE_OK]), False, ["PATHS"]),
        (_Stack(interfaces=[SHARED, RNODE_DEAF]), True, ["RNODE", "PATHS", "HOST_ID"]),
        (
            _Stack(role="shared", interfaces=[SHARED, RNODE_OK]),
            False,
            ["ROLE", "PATHS"],
        ),
        (_Stack(stats_error=REJECTED), False, ["STATS", "PATHS"]),
        (_Stack(interfaces=[dict(RNODE_DEAF, rxb=0)]), False, ["PATHS"]),
        (object(), False, ["ROLE", "PATHS"]),
    ],
    ids=["working", "deaf_rnode", "own_stack", "refused", "silent_rnode", "unknown"],
)
def test_hints_follow_the_connect_facts(diag, stack, pending, expected):
    """Each hint answers one connect fact; ``rnpath -t`` is always worth a look."""
    facts = diag.read_facts(stack, [])
    chosen = diag.hints(facts, host_pending=pending)
    assert chosen == [getattr(diag, f"HINT_{name}") for name in expected]


def test_hints_are_short_operator_lines(diag):
    """Short imperative lines, no em-dash or en-dash (SPEC RG3)."""
    for name in ("ROLE", "STATS", "RNODE", "PATHS", "HOST_ID"):
        hint = getattr(diag, f"HINT_{name}")
        assert len(hint) <= 120, name
        assert "—" not in hint and "–" not in hint, name
        assert hint[0].isupper(), name


# ---------------------------------------------------------------------------
# The rate limit, the window and failures (SPEC RG4)
# ---------------------------------------------------------------------------


def test_the_summary_runs_once_per_interval(diag, logged):
    """Every loop asks; one summary per interval, one stats read each, no path table."""
    stack = _Stack(interfaces=[SHARED, RNODE_OK])
    clock = _Clock()
    diagnostics = diag.StackDiagnostics(lambda: [_client_socket()], clock=clock)
    diagnostics.tick(stack)  # nothing connected yet
    assert logged == []
    diagnostics.connected(stack)
    hour = diag.SUMMARY_INTERVAL_SECS
    for now in (60.0, hour - 1, hour, hour + 60, 2 * hour - 1, 2 * hour):
        clock.now = now
        diagnostics.tick(stack)
    assert _messages(logged) == [_STATE, _SUMMARY, _SUMMARY]
    assert (stack.stats_reads, stack.path_table_reads) == (3, 0)
    assert hour == 3600.0


def test_a_reconnect_refreshes_the_facts_and_keeps_the_window(
    diag, logged, monkeypatch
):
    """The summary covers the hour since the first connect, hinted from the latest."""
    monkeypatch.setattr(handler_state, "_host_node_id", "!27716218")
    stack = _Stack(interfaces=[SHARED, RNODE_OK], stats_error=REJECTED)
    clock = _Clock()
    diagnostics = diag.StackDiagnostics(lambda: [_client_socket()], clock=clock)
    diagnostics.connected(stack)
    clock.now, stack.stats_error = 1800.0, None
    diagnostics.connected(stack)  # the reconnect reads the stats this time
    clock.now = diag.SUMMARY_INTERVAL_SECS - 1
    diagnostics.tick(stack)
    clock.now = diag.SUMMARY_INTERVAL_SECS
    diagnostics.tick(stack)
    assert _messages(logged) == [_STATE, _STATE, _SUMMARY]
    severity, fields = _only(logged, _SUMMARY)
    assert (severity, fields["hints"]) == ("warn", [diag.HINT_PATHS])
    assert "rnode_rxb" not in fields  # the window's first read was refused


def test_a_failing_connect_read_logs_its_class_and_never_raises(diag, logged):
    """A diagnostics bug must not fail the connect it runs in (SPEC RG4, RG5)."""

    def broken():
        raise RuntimeError("RNodeInterface[Garden Shed]")

    clock = _Clock()
    diagnostics = diag.StackDiagnostics(broken, clock=clock)
    diagnostics.connected(_Stack())
    assert logged == [
        (
            "warn",
            _FAILED,
            {"context": "reticulum.connect", "error_class": "RuntimeError"},
        )
    ]
    clock.now = diag.SUMMARY_INTERVAL_SECS
    diagnostics.tick(_Stack())  # no window started, so no summary
    assert len(logged) == 1


def test_a_failing_summary_is_logged_once_per_interval(diag, logged):
    """The window advances, so a broken read repeats hourly, not every loop."""
    reads: list = []

    def interfaces():
        reads.append(1)
        if len(reads) > 1:
            raise RuntimeError("Garden Shed")
        return []

    clock = _Clock()
    diagnostics = diag.StackDiagnostics(interfaces, clock=clock)
    diagnostics.connected(_Stack())
    for now in (3600.0, 3601.0, 7200.0):
        clock.now = now
        diagnostics.tick(_Stack())
    failures = [fields for _sev, msg, fields in logged if msg == _FAILED]
    assert (
        failures
        == [{"context": "reticulum.summary", "error_class": "RuntimeError"}] * 2
    )
    assert "Garden Shed" not in repr(logged)


def test_the_module_reads_only_what_it_is_given(diag):
    """No RNS import: it cannot transmit or read the path table (RN5, MA7, RE-A14)."""
    assert not hasattr(diag, "RNS")


# ---------------------------------------------------------------------------
# The provider's wiring (SPEC RG2, RG4)
# ---------------------------------------------------------------------------


def test_reconnects_share_one_tally(diag, monkeypatch, tmp_path, logged):
    """The tally and window belong to the provider, so they outlive a connection."""
    fake, _state = _fake_rns()
    monkeypatch.setattr(mod, "RNS", fake)
    _no_ingestor_node_id(monkeypatch, tmp_path)
    provider = mod.ReticulumProvider()
    first, _target, _next = provider.connect(active_candidate=None)
    second, _target, _next = provider.connect(active_candidate=None)
    assert first.tally is second.tally
    assert isinstance(first.tally, diag.AnnounceTally)
    assert mod._ReticulumInterface(target=None).tally is not first.tally
    # The fakes lack the RNS members: the connect line says so and never warns.
    state = [fields for _sev, msg, fields in logged if msg == _STATE]
    assert (
        state
        == [
            {
                "context": "reticulum.connect",
                "role": "unknown",
                "stats": "unavailable",
                "received": None,
            }
        ]
        * 2
    )
    assert [sev for sev, _msg, _fields in logged if sev == "warn"] == []


def test_the_hook_summarises_the_connected_stack_once_due(
    diag, monkeypatch, tmp_path, logged
):
    """``log_diagnostics`` reads the interface's stack and keeps to the interval."""
    stack = _Stack(interfaces=[SHARED, RNODE_OK])
    fake, _state = _fake_rns(existing_instance=stack)
    fake.Transport.interfaces = [_client_socket()]
    monkeypatch.setattr(mod, "RNS", fake)
    _no_ingestor_node_id(monkeypatch, tmp_path)
    now = [0.0]
    monkeypatch.setattr(time, "monotonic", lambda: now[0])
    provider = mod.ReticulumProvider()
    provider.log_diagnostics(None)  # before any connect: nothing
    iface, _target, _next = provider.connect(active_candidate=None)
    provider.log_diagnostics(iface)
    now[0] = diag.SUMMARY_INTERVAL_SECS
    provider.log_diagnostics(iface)
    provider.log_diagnostics(iface)
    assert _messages(logged).count(_SUMMARY) == 1
    assert _only(logged, _STATE)[1]["role"] == "client"


def test_each_announce_outcome_reaches_the_tally(diag, monkeypatch):
    """Every handler outcome is tallied; only admitted announces count for MA1."""
    fake, _state = _fake_rns(hops=1, interface="TCPInterface[Hub]")
    monkeypatch.setattr(mod, "RNS", fake)
    monkeypatch.setattr(config, "RETICULUM_INTERFACES", ("rnode",))
    monkeypatch.setattr(config, "_debug_log", lambda *_a, **_k: None)
    seen: list = []
    monkeypatch.setattr(mod.handlers, "_mark_packet_seen", lambda: seen.append(1))
    monkeypatch.setattr(mod.handlers, "upsert_node", lambda *_a, **_k: None)
    iface = mod._ReticulumInterface(target=None)
    iface.tally.use_classes({"TCPInterface[Hub]": "TCPClientInterface"})
    handler = mod._ReticulumAnnounceHandler("lxmf.delivery", iface)

    handler.received_announce(PEER_DEST, PEER, b"Peer")  # TCP: out of scope
    fake.Transport.next_hop_interface = lambda _dh: None
    handler.received_announce(PEER_DEST, PEER, b"Peer")  # no interface
    fake.Transport.next_hop_interface = lambda _dh: "RNodeInterface[RNode LoRa]"
    handler.received_announce(PEER_DEST, PEER, b"Peer")  # admitted
    handler.received_announce(b"\x01\x02", None, b"Peer")  # unusable hash

    def _boom(*_args, **_kwargs):
        raise RuntimeError("queue down")

    monkeypatch.setattr(mod.handlers, "upsert_node", _boom)
    handler.received_announce(PEER_DEST, PEER, b"Peer")  # error

    counts = iface.tally.drain()
    assert counts.delivered == {"lxmf.delivery": 5}
    assert counts.admitted == {"lxmf.delivery": 1}
    assert counts.dropped == {
        "error": 1,
        "no_interface": 1,
        "scope": 1,
        "unusable_hash": 1,
    }
    assert counts.scope_classes == {"TCPClientInterface": 1}
    assert len(seen) == 3  # admitted, unusable hash, error; never a scope drop


def test_radio_facts_skip_an_interface_type_that_is_not_a_string(diag):
    """A stats entry whose ``type`` is unhashable is no radio and raises nothing."""
    radio = diag._radio_facts([{"type": ["RNodeInterface"]}, RNODE_OK])
    assert radio["rnode_online"] == "1/1"