"""
The operator can close uploading to everyone but themselves.

The point of these tests is the difference between a hidden button and a closed
door. The interface stops offering the control, which is a courtesy to the
people who are not trying; **the node refuses the upload**, which is the part
that holds against someone who is. A member who kept an old tab open, or who
speaks MNP directly, gets the same answer as everyone else.

Two further things are worth holding:

* the setting is changed by a **signed** operator instruction. A node that took
  it from an unsigned message would let any member turn it back on, and the
  control would be a suggestion;
* it is stored on the **node**, not the hub. A hub that could decide who may
  write to the operator's disk is a hub with authority over the node, which is
  the thing this whole design is arranged to avoid.
"""

import base64
from pathlib import Path

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey

from meshbay_common.adminop import OP_MEMBER_UPLOAD
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roster import Roster
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root

pytestmark = pytest.mark.asyncio


def _session(tmp_path: Path, user_id: str, *, member_upload: bool,
             operator: str | None = None) -> WebRTCPeerSession:
    shared_root = tmp_path / "shared"
    shared_root.mkdir(exist_ok=True)
    index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate())
    ctx = {
        "roots": one_root(shared_root),
        "index": index,
        "sk_node": index.sk_node,
        "member_upload": member_upload,
        "node_user_id": operator,
    }
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = None
    session._user_id = user_id
    session._pk_user = ""
    session._uploads = {}
    session.sent = []
    session._send = session.sent.append
    session._audit = lambda *a, **k: None
    return session


def _upload(session, filename="clip.mp4", body=b"bytes"):
    session._do_file_upload({
        "filename": filename, "chunk_index": 0, "total_chunks": 1,
        "data": base64.b64encode(body).decode(),
    })


def _uploads_dir(session) -> Path:
    return session._ctx["roots"].upload_root.path / "uploads"


# ── The door, not the button ────────────────────────────────────────────────

async def test_a_member_cannot_upload_when_it_is_turned_off(tmp_path):
    session = _session(tmp_path, "member-1", member_upload=False,
                       operator="the-operator")
    _upload(session)

    assert not (_uploads_dir(session) / "clip.mp4").exists(), (
        "the file was written even though uploading is off — the setting is "
        "decorative and the hidden button was the whole control")
    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "member_upload_off"


async def test_the_operator_can_still_upload(tmp_path):
    """Otherwise turning it off locks the operator out of their own node, and
    the only way back is a config file and a restart."""
    session = _session(tmp_path, "the-operator", member_upload=False,
                       operator="the-operator")
    _upload(session)

    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"


async def test_members_upload_normally_when_it_is_on(tmp_path):
    session = _session(tmp_path, "member-1", member_upload=True,
                       operator="the-operator")
    _upload(session)

    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"


async def test_a_node_that_never_heard_of_the_setting_still_accepts_uploads(tmp_path):
    """An existing node's context has no such key. The absence must read as
    "allowed", or upgrading the node silently closes every group."""
    session = _session(tmp_path, "member-1", member_upload=True,
                       operator="the-operator")
    del session._ctx["member_upload"]
    _upload(session)

    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"


# ── Who may change it ───────────────────────────────────────────────────────

async def test_changing_it_needs_a_signature(tmp_path):
    """
    The request only ever produces a challenge. Nothing is applied until a
    signature over the transcript verifies — the same path as removing a member.
    """
    session = _session(tmp_path, "member-1", member_upload=True,
                       operator="the-operator")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_member_upload({"allowed": False})

    assert issued == [(OP_MEMBER_UPLOAD, "off")]
    assert session._ctx["member_upload"] is True, "applied before it was signed"


async def test_the_subject_names_the_outcome_not_the_operation(tmp_path):
    """The operator is shown the subject before signing. "member_upload" tells
    them nothing; "off" tells them what they are about to do."""
    session = _session(tmp_path, "op", member_upload=False, operator="op")
    session._has_admin_authority = lambda: True
    issued = []
    session._issue_admin_challenge = lambda op, subject: issued.append((op, subject))

    session._do_member_upload({"allowed": True})

    assert issued == [(OP_MEMBER_UPLOAD, "on")]


async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
    session = _session(tmp_path, "member-1", member_upload=True,
                       operator="the-operator")
    session._has_admin_authority = lambda: False

    session._do_member_upload({"allowed": False})

    assert [m for m in session.sent if m.get("type") == "error"]


# ── Where it is stored ──────────────────────────────────────────────────────

async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path):
    roster = Roster(db_path=tmp_path / "roster.db")
    await roster.open()
    try:
        assert await roster.member_upload_allowed("g1") is True, (
            "absent must mean allowed, or an upgrade closes every group")
        await roster.set_member_upload("g1", False, set_by="op")
        assert await roster.member_upload_allowed("g1") is False
    finally:
        await roster.close()

    reopened = Roster(db_path=tmp_path / "roster.db")
    await reopened.open()
    try:
        assert await reopened.member_upload_allowed("g1") is False
        assert await reopened.member_upload_allowed("g2") is True, (
            "one group's setting must not answer for another")
    finally:
        await reopened.close()