"""
How many people may watch at once, and who decides.

Bounding the client's read-ahead to ninety seconds of film changed what a
transcode slot is. It used to be a burst — the browser took segments as fast as
it could append them, so a slot came back within the minute whatever the length
of the film. Now a slot is held for as long as someone is watching, so the cap
is a cap on simultaneous viewers, and the right number stopped being a property
of the code: it depends on the machine the node runs on.

So it belongs to the operator. `max_concurrent_streams` under [node] in
node.toml, `MESHBAY_MAX_CONCURRENT_STREAMS` in the environment, and the
constant in the source as the default when neither says anything.

The tests below follow the value along that whole path rather than checking
that the field parses, because every join in it has been wrong at least once:
the daemon reads `self._config`, not `self.cfg`, and nothing about the
attribute that does not exist fails until a video is played.
"""

import asyncio
import textwrap
from pathlib import Path

import pytest

from meshbay_node.config import load_config
from meshbay_node.transport.webrtc_server import (
    MAX_CONCURRENT_TRANSCODES,
    WebRTCPeerSession,
    WebRTCTransport,
)


def _cfg(tmp_path: Path, body: str):
    p = tmp_path / "node.toml"
    p.write_text(textwrap.dedent(body))
    return load_config(p)


# ── What the operator writes ──────────────────────────────────────────────────

def test_the_operator_sets_it(tmp_path):
    cfg = _cfg(tmp_path, """
        [node]
        max_concurrent_streams = 3
    """)
    assert cfg.node.max_concurrent_streams == 3


def test_saying_nothing_gets_the_default(tmp_path):
    cfg = _cfg(tmp_path, """
        [node]
        quic_port = 19010
    """)
    assert cfg.node.max_concurrent_streams == MAX_CONCURRENT_TRANSCODES, (
        "the config default and the source default disagree, so the number "
        "depends on whether a node.toml happens to mention it")


def test_the_environment_wins_over_the_file(tmp_path, monkeypatch):
    monkeypatch.setenv("MESHBAY_MAX_CONCURRENT_STREAMS", "5")
    cfg = _cfg(tmp_path, """
        [node]
        max_concurrent_streams = 3
    """)
    assert cfg.node.max_concurrent_streams == 5


@pytest.mark.parametrize("value", ["0", "-4", '"lots"', "true"])
def test_a_value_that_would_break_streaming_is_refused(tmp_path, value, caplog):
    """Zero is the dangerous one.

    `asyncio.Semaphore(0)` is not "no limit". It is a node where every video
    waits forever, with nothing in the log to say why — so the operator gets
    the default and a warning naming the setting instead.
    """
    cfg = _cfg(tmp_path, f"""
        [node]
        max_concurrent_streams = {value}
    """)
    assert cfg.node.max_concurrent_streams == MAX_CONCURRENT_TRANSCODES
    assert "max_concurrent_streams" in caplog.text, (
        "the value was silently discarded — the operator has no way to learn "
        "their setting is not in effect")


# ── That the number reaches the thing it limits ───────────────────────────────

class _FakePC:
    def on(self, *a, **k):
        return lambda f: f


def _semaphore_size(n):
    t = WebRTCTransport(
        sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
        shared_root=Path("/tmp"), index=None, max_concurrent_streams=n)
    s = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="p")
    return s._transcode_semaphore()._value


@pytest.mark.parametrize("n,expect", [(None, MAX_CONCURRENT_TRANSCODES), (3, 3), (20, 20)])
def test_the_configured_number_is_the_semaphore(n, expect):
    assert asyncio.run(_run(n)) == expect


async def _run(n):
    return _semaphore_size(n)


def test_the_budget_is_shared_between_peers():
    """One budget for the node, not one per browser.

    Building it per call would cap nothing: every viewer would arrive with a
    full allowance and the node would spawn ffmpeg without limit.
    """
    t = WebRTCTransport(
        sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
        shared_root=Path("/tmp"), index=None, max_concurrent_streams=2)

    async def go():
        a = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="a")
        b = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="b")
        sem_a, sem_b = a._transcode_semaphore(), b._transcode_semaphore()
        assert sem_a is sem_b, "each peer got its own budget, so there is no cap"
        await sem_a.acquire()
        assert sem_b._value == 1, "one peer's stream did not spend the node's budget"

    asyncio.run(go())


def test_the_daemon_passes_it(  ):
    """The join that syntax checking cannot see.

    `self.cfg` parses and imports perfectly well; it raises AttributeError the
    first time somebody plays a video, which is not where anyone would look.
    """
    daemon = (Path(__file__).resolve().parents[1] / "src" / "meshbay_node"
              / "daemon.py").read_text()
    i = daemon.index("WebRTCTransport(")
    call = daemon[i:daemon.index(")", daemon.index("denylist=denylist", i))]
    assert "max_concurrent_streams=" in call, (
        "the daemon builds the transport without the operator's setting, so "
        "node.toml is read and then ignored")
    assert "self._config.node.max_concurrent_streams" in call, (
        "the daemon holds its config in _config; any other attribute is an "
        "AttributeError deferred until someone plays a video")