"""
Availability, on the node: what the people you invited can cost you.

The message budget for an unauthenticated peer (`PRE_HANDSHAKE_MAX_MSG`, H6)
bounds what *one* of them spends. Nothing bounded how many there could be, or
how long one could stay without ever proving anything — and the hub's cap is
three offers in flight **per account**, which is a limit on each caller rather
than on this machine, so an operator's exposure grew with the number of people
in their groups. A member who never meant any harm — a tab left open through a
sleep, a client retrying a connection it cannot complete — arrives here the
same way as one who does.
"""

import asyncio
from unittest.mock import MagicMock

import pytest
from meshbay_node.transport import webrtc_server as ws_mod
from meshbay_node.transport.webrtc_server import (
    MAX_PEER_SESSIONS,
    UNAUTHENTICATED_SESSION_TIMEOUT,
    WebRTCTransport,
)


def _transport() -> WebRTCTransport:
    return WebRTCTransport(sk_node=MagicMock(), hub_pk_pem=b"", gek=None,
                           roots=None, index=None)


class _FakeSession:
    """Stands in for a peer that connected and then said nothing."""

    def __init__(self, user_id=None):
        self._user_id = user_id
        self.closed = False

    async def close(self):
        self.closed = True


@pytest.mark.asyncio
async def test_the_node_refuses_more_peers_than_it_will_hold():
    """The refusal comes before an RTCPeerConnection is allocated, or the cap
    would be counting the thing it is meant to prevent."""
    tp = _transport()
    for i in range(MAX_PEER_SESSIONS):
        tp._sessions[f"peer-{i}"] = _FakeSession(user_id="someone")

    with pytest.raises(RuntimeError, match="peer-connection limit"):
        await tp.handle_offer("v=0", "one-too-many")

    assert "one-too-many" not in tp._sessions


@pytest.mark.asyncio
async def test_a_peer_that_never_handshakes_is_closed(monkeypatch):
    """`connectionstatechange` reaps a connection that *fails*. One that
    succeeds and stays silent was held for the life of the daemon."""
    monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
    tp = _transport()
    session = _FakeSession(user_id=None)
    tp._sessions["quiet"] = session

    tp._reap_if_unauthenticated("quiet")
    await asyncio.sleep(0.2)

    assert session.closed, "a peer that never proved anything was kept"
    assert "quiet" not in tp._sessions


@pytest.mark.asyncio
async def test_a_peer_that_handshaked_is_left_alone(monkeypatch):
    """`_user_id` is set by the GEK proof, and is the one honest test of
    whether this peer ever became anybody."""
    monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05)
    tp = _transport()
    session = _FakeSession(user_id=None)
    tp._sessions["real"] = session

    tp._reap_if_unauthenticated("real")
    session._user_id = "a-real-member"        # the handshake completes
    await asyncio.sleep(0.2)

    assert not session.closed, "a member who completed the handshake was cut off"
    assert tp._sessions["real"] is session


@pytest.mark.asyncio
async def test_handle_offer_arms_the_reaper(monkeypatch):
    """
    The seam, driven rather than described. The three tests above call
    `_reap_if_unauthenticated` themselves, so every one of them would still
    pass with the call removed from `handle_offer` and no peer reaped at all —
    which is the defect, not the helper. aiortc is stubbed because a real
    RTCPeerConnection wants a real SDP; everything else here is the shipped
    code path.
    """
    class _FakePC:
        def __init__(self, *a, **kw):
            self.localDescription = MagicMock(sdp="v=0\r\n")
            self.remoteDescription = None

        def on(self, _event):
            return lambda fn: fn

        async def setRemoteDescription(self, _d): pass
        async def createAnswer(self): return MagicMock()
        async def setLocalDescription(self, _d): pass
        async def close(self): pass

    monkeypatch.setattr(ws_mod, "RTCPeerConnection", _FakePC)
    monkeypatch.setattr(ws_mod, "RTCSessionDescription",
                        lambda **kw: MagicMock(**kw))

    tp = _transport()
    await tp.handle_offer("v=0\r\n", "fresh-peer")
    try:
        assert tp._reapers, (
            "handle_offer allocated a peer session and armed nothing to "
            "close it if the handshake never comes")
    finally:
        await tp.close_all()


class _StatefulPC:
    """An RTCPeerConnection that keeps the handlers the transport registers, so
    a test can drive the connection through its states."""

    def __init__(self, *a, **kw):
        self.localDescription = MagicMock(sdp="v=0\r\n")
        self.remoteDescription = None
        self.connectionState = "new"
        self.handlers = {}

    def on(self, event):
        def register(fn):
            self.handlers[event] = fn
            return fn
        return register

    async def setRemoteDescription(self, _d): pass
    async def createAnswer(self): return MagicMock()
    async def setLocalDescription(self, _d): pass
    async def close(self): pass


@pytest.mark.asyncio
@pytest.mark.parametrize("state", ["closed", "failed"])
async def test_a_connection_that_ends_leaves_its_groups_peer_set(monkeypatch, state):
    """
    A closed tab, a phone gone to sleep and a dead network all arrive at
    `connectionstatechange`, which dropped the session from the transport and
    stopped its tasks but left it in its group's peer set: only `close()` took
    it out, and nothing on this path called it. Every broadcast to the group
    was then written to a closed channel ("WebRTC send skipped", thirteen times
    per settings change on a node up for a day), and every reconnect left one
    more dead session held until the node restarted.
    """
    pcs = []

    def make_pc(*a, **kw):
        pcs.append(_StatefulPC())
        return pcs[-1]

    monkeypatch.setattr(ws_mod, "RTCPeerConnection", make_pc)
    monkeypatch.setattr(ws_mod, "RTCSessionDescription",
                        lambda **kw: MagicMock(**kw))

    tp = _transport()
    tp._ctx["groups"] = {"g1": {"name": "g1"}}
    try:
        await tp.handle_offer("v=0\r\n", "peer-1")
        session = tp._sessions["peer-1"]
        session._user_id, session._group_id = "a-member", "g1"   # handshake done
        session._register_peer()
        peers = tp._ctx["groups"]["g1"]["_peers"]
        assert list(peers.values()) == [session]

        pcs[0].connectionState = state
        await pcs[0].handlers["connectionstatechange"]()

        assert "peer-1" not in tp._sessions
        assert not peers, "the group still broadcasts to a connection that is gone"
    finally:
        await tp.close_all()


@pytest.mark.asyncio
async def test_the_reaper_is_held_and_cancelled_with_the_transport():
    """asyncio keeps only a weak reference to a task, and a reaper collected
    mid-sleep reaps nothing — the same trap as every other task in this file."""
    tp = _transport()
    tp._sessions["held"] = _FakeSession()
    tp._reap_if_unauthenticated("held")
    assert tp._reapers, "the reaper was fired and forgotten"

    await tp.close_all()
    assert not tp._reapers
    # The timeout is a real duration, not something a test has to wait out.
    assert UNAUTHENTICATED_SESSION_TIMEOUT >= 30