"""
One member's typing must not spend what the whole group depends on.

`tmdb_search_req` takes a member's free text and calls TMDB with the
**operator's** credential. That credential is rated by TMDB and shared: the
automatic matching every other member sees runs on it too. So a member holding
down a search box — or a script doing it — degrades the library for everyone and
costs the operator their quota, and the node had no ceiling of any kind on it.
§6.5's standing rule is a bound and a named adversary in the same commit; this
handler shipped with neither.

Two members in every test here, which is the point: a ceiling that one person
can exhaust for another is not a ceiling, it is a queue. The per-member window
is what keeps them apart, and the node-wide one is what keeps them together
from emptying the operator's quota — they answer different questions and both
are checked.

The refusal is an error rather than an empty result. An empty list is what "no
such film" looks like, and telling somebody their film is unknown when the node
simply declined to ask is a worse answer than the truth.
"""

import pytest
from meshbay_node.transport import webrtc_server
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

GROUP = "g" * 32


class _FakeTmdb:
    """Counts what would have been spent."""

    def __init__(self):
        self.calls = 0

    async def search_movie_results(self, query):
        self.calls += 1
        return [{"id": 1, "title": "Some Saga", "release_date": "1999-01-01",
                 "poster_path": None}]

    async def search_tv_results(self, query):
        self.calls += 1
        return []


class _FakeMediaCache:
    async def get_thumb_hash_by_file_id(self, _file_id):
        return None


@pytest.fixture
def group():
    """One group's context, shared by every session in it, as a node has."""
    return {
        "gek": b"k" * 32,
        "tmdb_enabled": True,
    }


@pytest.fixture
def node(group):
    tmdb = _FakeTmdb()
    ctx = {
        "groups": {GROUP: group},
        "media_cache": _FakeMediaCache(),
        "tmdb_client": tmdb,
    }
    return ctx, tmdb


def _member(ctx, user_id: str) -> WebRTCPeerSession:
    s = WebRTCPeerSession.__new__(WebRTCPeerSession)
    s._ctx = ctx
    s._group_id = GROUP
    s._user_id = user_id
    s._peer_id = user_id
    s.sent = []
    s._send = s.sent.append
    s._audit = lambda *a, **k: None
    return s


async def _search(session, query="a film"):
    await session._do_tmdb_search_request(
        {"query": query, "media_type": "movie"})


def _refusals(session):
    return [m for m in session.sent
            if m.get("code") == "tmdb_search_rate_limited"]


async def test_a_member_at_the_ceiling_does_not_stop_another_one(node, monkeypatch):
    """
    The property a one-member test cannot state.

    Alice exhausts her own window; Bob, who has typed nothing, must be served
    exactly as if she had not been there.
    """
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 3)
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_NODE", 100)
    ctx, tmdb = node

    alice = _member(ctx, "alice")
    for i in range(4):
        await _search(alice, f"film {i}")
    assert tmdb.calls == 3, "the ceiling did not stop the fourth search"
    assert len(_refusals(alice)) == 1

    bob = _member(ctx, "bob")
    await _search(bob, "something else")
    assert tmdb.calls == 4
    assert _refusals(bob) == []


async def test_one_member_cannot_spend_the_whole_node_quota(node, monkeypatch):
    """
    And the other half: two members together still meet a node-wide ceiling,
    because the operator's credential is one credential however many people
    hold the search box down.
    """
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 100)
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_NODE", 2)
    ctx, tmdb = node

    alice, bob = _member(ctx, "alice"), _member(ctx, "bob")
    await _search(alice)
    await _search(bob)
    await _search(bob)

    assert tmdb.calls == 2
    assert len(_refusals(bob)) == 1


async def test_a_members_count_survives_their_reconnection(node, monkeypatch):
    """
    Kept in the group context, not on the session: otherwise the ceiling is one
    reconnect wide, and a client that drops its DataChannel between searches has
    no ceiling at all.
    """
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 2)
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_NODE", 100)
    ctx, tmdb = node

    first = _member(ctx, "alice")
    await _search(first, "one")
    await _search(first, "two")

    reconnected = _member(ctx, "alice")     # same person, new connection
    await _search(reconnected, "three")

    assert tmdb.calls == 2, "a reconnect reset the member's window"
    assert len(_refusals(reconnected)) == 1


async def test_a_refusal_is_said_out_loud_and_not_drawn_as_no_matches(node, monkeypatch):
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_PER_MEMBER", 0)
    ctx, _ = node

    alice = _member(ctx, "alice")
    await _search(alice)

    (msg,) = alice.sent
    assert msg["type"] == "error"
    assert msg["code"] == "tmdb_search_rate_limited"
    assert msg.get("results") is None, (
        "a refusal that carries an empty result list reads as 'no such film'")


async def test_the_windows_do_not_grow_without_bound(node, monkeypatch):
    """
    The lists are trimmed on every call, so the thing that bounds a member also
    bounds what remembering them costs.
    """
    monkeypatch.setattr(webrtc_server, "_TMDB_SEARCH_WINDOW", 0.0)
    ctx, tmdb = node

    alice = _member(ctx, "alice")
    for i in range(12):
        await _search(alice, f"film {i}")

    # Every entry ages out before the next call, so nothing is refused, and what
    # is kept is the one just recorded rather than one per search ever made.
    assert tmdb.calls == 12
    assert len(ctx["groups"][GROUP]["tmdb_search_hits"]["alice"]) == 1
    assert len(ctx["tmdb_search_hits_node"]) == 1