"""
Who may write to the operator's disk, now that RO/RW on the root decides it.

This replaces `test_member_upload_policy.py`. The old model had two orthogonal
controls — one root designated as the upload target, and a group-wide
`member_upload` switch — and collapsed into one property per root: `writable`.
The properties worth keeping from the old file survive the change unaltered:

* the interface hiding a control is a courtesy to the people who are not
  trying; **the node refusing is the part that holds** against someone who is.
  A member with an old tab open, or one speaking MNP directly, gets the same
  answer. That half is pinned in `test_security_regressions.py`, next to the
  overwrite properties it belongs with;
* the setting is changed by a **signed** operator instruction, or it is a
  suggestion any member can undo;
* it is stored on the **node**, never the hub. A hub that could decide who
  writes to the operator's disk would have authority over the node.

And one that is new: the *old* message must no longer be able to change
anything. A deprecated instruction that still works is not deprecated, and this
one would reopen uploads group-wide.
"""

import base64
from pathlib import Path

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

from meshbay_common.adminop import OP_ROOT_UPDATE, OP_ROOT_EJECT, OP_ROOT_PLUG
from meshbay_common.protocol import MNP
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.roots import RootSet
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

pytestmark = pytest.mark.asyncio


def _session(tmp_path: Path, user_id: str, *,
             writable: bool = True,
             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": RootSet.build([{"path": str(shared_root), "writable": writable}]),
        "index": index,
        "sk_node": index.sk_node,
        "node_user_id": operator,
    }
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = "g" * 32
    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, "root": "shared",
        "chunk_index": 0, "total_chunks": 1,
        "data": base64.b64encode(body).decode(),
    })


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


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

async def test_a_member_cannot_upload_to_a_read_only_root(tmp_path):
    session = _session(tmp_path, "member-1", writable=False)
    _upload(session)

    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_read_only"
    assert not _uploads_dir(session).exists()


async def test_members_upload_normally_to_a_writable_root(tmp_path):
    session = _session(tmp_path, "member-1", writable=True)
    _upload(session)

    assert not [m for m in session.sent if m.get("type") == "error"]
    assert (_uploads_dir(session) / "clip.mp4").read_bytes() == b"bytes"


async def test_read_only_binds_the_operator_too(tmp_path):
    """
    The old model exempted the operator, because the switch was about *members*.
    RO is about the directory: a published library is read-only for everyone, and
    an exception for admin authority is how a rule turns into a default.
    """
    session = _session(tmp_path, "the-operator", writable=False,
                       operator="the-operator")
    session._is_node_admin = lambda: True
    _upload(session)

    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_read_only"


# ── Signed, or it is a suggestion ────────────────────────────────────────────

def _capture_challenges(session) -> list[tuple[str, str]]:
    issued: list[tuple[str, str]] = []

    def issue(op, subject, **kw):
        issued.append((op, subject))

    session._issue_admin_challenge = issue
    session._has_admin_authority = lambda: True
    return issued


async def test_changing_a_roots_flags_needs_a_signature(tmp_path):
    """The flags are not applied by the request — only by the signed response."""
    session = _session(tmp_path, "the-operator", operator="the-operator")
    issued = _capture_challenges(session)

    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
                             "writable": False})

    assert [op for op, _ in issued] == [OP_ROOT_UPDATE]
    assert session._ctx["roots"].roots[0].writable 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, so it has to say what will
    be true afterwards. "shared" alone would have them authorize a change they
    cannot see the direction of.
    """
    session = _session(tmp_path, "op", operator="op")
    issued = _capture_challenges(session)

    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
                             "writable": True, "removable": True})

    assert issued == [(OP_ROOT_UPDATE, "shared:rw=on,rem=on")]


async def test_eject_and_plug_are_signed_too(tmp_path):
    """
    Hiding a group's whole library from every member is not a lesser act than
    changing a flag. An unsigned one would let any member black out a group.
    """
    session = _session(tmp_path, "op", operator="op")
    issued = _capture_challenges(session)

    session._do_root_eject({"group_id": "g" * 32, "root_name": "shared"})
    session._do_root_plug({"group_id": "g" * 32, "root_name": "shared"})

    assert issued == [(OP_ROOT_EJECT, "shared"), (OP_ROOT_PLUG, "shared")]


async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path):
    """
    An unpaired node has no key to check a signature against, so the challenge
    is never issued rather than issued and then unverifiable.
    """
    session = _session(tmp_path, "member-1")
    issued = _capture_challenges(session)
    session._has_admin_authority = lambda: False

    session._do_root_update({"group_id": "g" * 32, "root_name": "shared",
                             "writable": True})

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


# ── The deprecated message must not still work ───────────────────────────────

async def test_the_old_member_upload_message_changes_nothing(tmp_path):
    """
    MNP still parses `member_upload` so an old client gets an answer instead of
    a dropped request. What it must not do is act: this instruction could
    reopen uploads for a whole group, and a client old enough to send it is
    exactly one that knows nothing about read-only roots.
    """
    session = _session(tmp_path, "member-1", writable=False)
    session._has_admin_authority = lambda: True
    issued = _capture_challenges(session)

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

    assert issued == [], "a deprecated instruction asked to be signed"
    assert session._ctx["roots"].roots[0].writable is False
    acks = [m for m in session.sent if m.get("type") == MNP.MEMBER_UPLOAD_ACK]
    assert acks and acks[0].get("deprecated") is True

    # And the door is still shut.
    _upload(session)
    refusal = [m for m in session.sent if m.get("type") == "error"]
    assert refusal and refusal[0].get("code") == "root_read_only"