"""
`max_concurrent_streams` must take effect without a restart.

`ops.set_node_settings` did this by assigning `webrtc._stream_sem` — an
attribute that has never existed. The pool is `ctx["_transcode_sem"]`, so
`hasattr(webrtc, "_stream_sem")` was always False, the branch never ran, and the
setting only ever applied on a restart. Draft-v6 §2.11 says it applies live, the
Node page offers it as a live setting, and it did nothing: an operator lowering
the cap on a struggling machine, or raising it after "Server busy", saw no
change and had no way to know why.

Nothing here mocks the pool. `set_capacity` is called on a real
`WebRTCTransport` and the assertions read what a stream request would actually
find.
"""

import asyncio

import pytest
from meshbay_node.transport.webrtc.apps.streaming import MAX_CONCURRENT_TRANSCODES
from meshbay_node.transport.webrtc_server import WebRTCPeerSession, WebRTCTransport


def _pool(transport) -> asyncio.Semaphore:
    """The pool a stream request would acquire, built the way one builds it."""
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = transport._ctx
    return session._transcode_semaphore()


@pytest.fixture
def transport(tmp_path):
    """A real WebRTCTransport. Its keys and index are genuine but incidental —
    nothing below the capacity code reads them."""
    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
    from meshbay_common.crypto import generate_gek
    from meshbay_node.indexer.group_index import GroupIndex

    from conftest import one_root

    sk_node = Ed25519PrivateKey.generate()
    gek = generate_gek()
    shared = tmp_path / "shared"
    shared.mkdir()
    return WebRTCTransport(
        sk_node=sk_node, hub_pk_pem=b"", gek=gek,
        roots=one_root(shared),
        index=GroupIndex(group_id="g", sk_node=sk_node, gek=gek),
        stun_servers=[])


def test_raising_the_cap_is_visible_to_the_next_stream(transport):
    """The bug, at its simplest: the number changes and nothing happens."""
    pool = _pool(transport)
    assert pool._value == MAX_CONCURRENT_TRANSCODES
    transport.set_capacity(max_concurrent_streams=16)
    assert _pool(transport)._value == 16, (
        "the setting was accepted and the pool never changed — this is the "
        "no-op that shipped")


def test_lowering_the_cap_does_not_interrupt_what_is_running(transport):
    """
    A slot is held for the length of a film, so lowering the cap cannot take a
    viewer's film away. It stops the next one starting, and the replacement pool
    carries only the permits that remain.
    """
    _pool(transport)
    transport._ctx["_streams_in_flight"] = 3
    transport.set_capacity(max_concurrent_streams=4)
    assert _pool(transport)._value == 1, (
        "a full set of permits would let more viewers in than either the old "
        "cap or the new one, on top of the three still watching")


def test_lowering_below_what_is_running_refuses_the_next_one(transport):
    _pool(transport)
    transport._ctx["_streams_in_flight"] = 6
    transport.set_capacity(max_concurrent_streams=2)
    assert _pool(transport)._value == 0, "the pool must not go negative"


def test_the_value_is_kept_for_a_pool_not_yet_built(transport):
    """Nothing has streamed, so there is nothing to resize — but the number has
    to be there when the first request builds the pool."""
    transport.set_capacity(max_concurrent_streams=3)
    assert transport._ctx.get("_transcode_sem") is None
    assert _pool(transport)._value == 3


def test_a_cap_below_one_is_refused(transport):
    for bad in (0, -1):
        with pytest.raises(ValueError):
            transport.set_capacity(max_concurrent_streams=bad)


def test_nothing_changes_when_nothing_is_passed(transport):
    _pool(transport)
    before = transport._ctx["_transcode_sem"]
    assert transport.set_capacity() == {}
    assert transport._ctx["_transcode_sem"] is before


@pytest.mark.asyncio
async def test_in_flight_is_counted_by_the_streaming_path_itself(transport):
    """
    `set_capacity` resizes against `_streams_in_flight`, so that counter has to
    be maintained where slots are actually taken — not set by a test. Drives the
    real `_stream_video`, with the work under it stubbed: what is being checked
    is the accounting around the slot, which is where flow control in this repo
    has gone wrong before.
    """
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = transport._ctx
    session._send = lambda msg: None

    seen = []
    release = asyncio.Event()

    async def _inner(_msg):
        seen.append(transport._ctx.get("_streams_in_flight"))
        await release.wait()

    session._stream_video_inner = _inner
    task = asyncio.create_task(session._stream_video({"file_id": "x"}))
    await asyncio.sleep(0)
    await asyncio.sleep(0)
    assert seen == [1], "the slot was taken without being counted"

    release.set()
    await task
    assert transport._ctx["_streams_in_flight"] == 0, (
        "a slot that is not given back is a viewer nobody can replace — the "
        "class of bug _replace_stream and shutdown_tasks exist for")


def test_ops_calls_the_real_mechanism():
    """
    The dead branch, pinned. `hasattr(webrtc, '_stream_sem')` is False for every
    WebRTCTransport that has ever existed, so a test that only checked
    "set_node_settings does not raise" passed throughout.
    """
    import inspect

    from meshbay_node import ops

    src = inspect.getsource(ops.set_node_settings)
    # Comments stripped: this function now *explains* the dead attribute, and a
    # test that matched the prose would fail on its own documentation.
    code = "\n".join(line.split("#", 1)[0] for line in src.splitlines())
    assert "_stream_sem" not in code, "the attribute that never existed is back"
    assert "set_capacity" in code, "the setting must reach the pool that exists"
    assert not hasattr(WebRTCTransport, "_stream_sem")