"""
The two scopes a transfer cap has, and the rule that they are not the same kind
of setting.

The **pools** are the machine's: how many transfers this node runs at once,
across every group, from `[node]` in node.toml with a roster override — the
docs/MESHBAY_DESIGN.md §6.8 pattern, changed from the Node page or the CLI,
applied live.

The **per-member cap** is a group's: how many one member may run at once here.
It lives on the node like every other group setting (not the hub, which would
have authority over someone else's disk; not node.toml, which is hand-written
and needs a restart), and changing it is a signed operator instruction, because
an unsigned cap is one any member can raise for themselves.

What is checked here is the seam between the stored value and the pool that
enforces it — a setting that is written, acknowledged and never read is the
shape of the bug this whole branch started from (`webrtc._stream_sem`).
"""

import pytest

from meshbay_node.roster import Roster
from meshbay_node.transfers import (
    DEFAULT_MAX_PER_MEMBER, DOWNLOAD, UPLOAD, TransferSlots,
)


@pytest.fixture
async def roster(tmp_path):
    r = Roster(db_path=tmp_path / "roster.db")
    await r.open()
    yield r
    await r.close()


# ── the group's own cap ─────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_absent_means_the_default_not_unlimited(roster):
    """A group that predates the setting must not come back unlimited: the
    node-wide pool would then be the only control, which is the situation slots
    exist to end."""
    assert await roster.transfer_limits("g1") == {}
    slots = TransferSlots()
    assert slots.member_cap(DOWNLOAD, ("g1", "alice")) == DEFAULT_MAX_PER_MEMBER


@pytest.mark.asyncio
async def test_the_cap_survives_a_restart(roster, tmp_path):
    await roster.set_transfer_limits("g1", {"download": 4, "upload": 1},
                                     set_by="op")
    await roster.close()
    reopened = Roster(db_path=tmp_path / "roster.db")
    await reopened.open()
    try:
        assert await reopened.transfer_limits("g1") == {"download": 4,
                                                        "upload": 1}
    finally:
        await reopened.close()


@pytest.mark.asyncio
async def test_one_group_does_not_set_anothers(roster):
    await roster.set_transfer_limits("g1", {"download": 5}, set_by="op")
    assert await roster.transfer_limits("g2") == {}


@pytest.mark.asyncio
async def test_a_stored_zero_never_becomes_a_cap_of_zero(roster):
    """Zero is not "unlimited" and must not be "nobody may transfer" either.
    Whatever reaches storage, the floor is one."""
    await roster.set_transfer_limits("g1", {"download": 0}, set_by="op")
    assert (await roster.transfer_limits("g1"))["download"] == 1


@pytest.mark.asyncio
async def test_rubbish_in_the_row_reads_as_unset(roster):
    """A row this code did not write must not take a group's transfers down —
    the same "a payload that does not open ends nothing silently" discipline
    the sealed messages follow."""
    await roster.set_setting("g1", Roster.SETTING_TRANSFER_LIMITS,
                             "not json", "op")
    assert await roster.transfer_limits("g1") == {}


# ── the pool that enforces it ───────────────────────────────────────────────

def test_a_group_cap_overrides_the_node_default():
    slots = TransferSlots()
    slots.set_group_limits("strict", {DOWNLOAD: 1})
    assert slots.member_cap(DOWNLOAD, ("strict", "alice")) == 1
    assert slots.member_cap(DOWNLOAD, ("other", "alice")) == DEFAULT_MAX_PER_MEMBER
    assert slots.member_cap(UPLOAD, ("strict", "alice")) == DEFAULT_MAX_PER_MEMBER, (
        "setting the download cap must not silently change the upload one")


def test_the_group_cap_is_what_queues_a_member():
    slots = TransferSlots()
    slots.set_group_limits("strict", {DOWNLOAD: 1})
    args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice",
                group_id="strict")
    assert slots.open(tr="t1", **args)[0].state == "granted"
    assert slots.open(tr="t2", **args)[0].state == "queued"


def test_raising_a_group_cap_starts_what_was_waiting():
    slots = TransferSlots()
    slots.set_group_limits("g1", {DOWNLOAD: 1})
    args = dict(kind=DOWNLOAD, session_key="s1", user_id="alice", group_id="g1")
    slots.open(tr="t1", **args)
    slots.open(tr="t2", **args)
    granted = slots.set_group_limits("g1", {DOWNLOAD: 3})
    assert [x.tr for x in granted] == ["t2"], (
        "the cap was raised and the waiting transfer was left waiting")


def test_one_groups_cap_does_not_move_anothers_queue():
    slots = TransferSlots()
    slots.set_group_limits("g1", {DOWNLOAD: 1})
    slots.set_group_limits("g2", {DOWNLOAD: 1})
    for g in ("g1", "g2"):
        args = dict(kind=DOWNLOAD, session_key=f"s-{g}", user_id="alice",
                    group_id=g)
        slots.open(tr=f"{g}-1", **args)
        slots.open(tr=f"{g}-2", **args)
    granted = slots.set_group_limits("g1", {DOWNLOAD: 2})
    assert [x.tr for x in granted] == ["g1-2"]
    assert slots.leases["g2-2"].state == "queued"


# ── the node-wide pools ─────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_the_node_wide_caps_round_trip_through_the_roster(roster):
    defaults = {"max_concurrent_downloads": 8, "max_concurrent_uploads": 8}
    assert await roster.node_settings(defaults) == {
        **{k: v for k, v in defaults.items()},
        **{k: None for k in ("invite_ttl_hours", "pair_ttl_hours",
                             "device_request_ttl_minutes",
                             "max_concurrent_streams",
                             "transcode_incompatible_video")},
        "stun_servers": [], "ice_interfaces": [],
    }
    await roster.set_node_setting(roster.SETTING_MAX_DOWNLOADS, "3", "op")
    assert (await roster.node_settings(defaults))["max_concurrent_downloads"] == 3


def test_node_toml_carries_both_keys():
    """The template is what an operator reads before they read any document."""
    from meshbay_node.config import EXAMPLE_CONFIG as tpl
    assert "max_concurrent_downloads" in tpl
    assert "max_concurrent_uploads" in tpl