"""
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()


@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