"""
A full disk is a stated refusal, not an exception.

Nothing on the upload path used to know about ENOSPC: a write that found no
room raised out of the handler, the catch-all answered "Request failed", and
the `.part` stayed behind holding the space that had run out. A client could
not tell that from any other fault, so the phone's nightly photo backup would
retry the same file every day. Now the node refuses with `disk_full` — up
front when the announced size cannot fit beside the reserve, and at the chunk
where a write actually fails — and keeps nothing it cannot finish.
"""

import errno

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import generate_gek
from meshbay_node.indexer.group_index import GroupIndex
from meshbay_node.transport.webrtc import upload_handlers
from meshbay_node.transport.webrtc.upload_handlers import DISK_RESERVE_BYTES
from meshbay_node.transport.webrtc_server import WebRTCPeerSession

from conftest import one_root, sealed_upload

GROUP = "g" * 32


def _peer(ctx: dict) -> WebRTCPeerSession:
    session = WebRTCPeerSession.__new__(WebRTCPeerSession)
    session._ctx = ctx
    session._group_id = GROUP
    session._user_id = "user-1"
    session._pk_user = ""
    session.sent = []
    session._send = session.sent.append
    session.audited = []
    session._audit = lambda *a, **k: session.audited.append(a)
    return session


def _group_ctx(tmp_path) -> dict:
    shared = tmp_path / "shared"
    shared.mkdir(exist_ok=True)
    return {"roots": one_root(shared),
            "index": GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate()),
            "gek": generate_gek()}


def _codes(session):
    return [m.get("code") for m in session.sent if m.get("type") == "error"]


def _free(monkeypatch, nbytes: int) -> None:
    monkeypatch.setattr(upload_handlers, "_free_bytes", lambda _d: nbytes)


def _shared(ctx):
    return ctx["roots"].roots[0].path


async def test_a_file_that_cannot_fit_is_refused_before_a_byte_is_written(tmp_path, monkeypatch):
    ctx = _group_ctx(tmp_path)
    peer = _peer(ctx)
    _free(monkeypatch, DISK_RESERVE_BYTES + 10)
    await peer._do_file_upload(sealed_upload(peer, filename="IMG_0001.jpg",
                                             data=b"x" * 8, total_chunks=2))
    assert _codes(peer) == ["disk_full"]
    assert list(_shared(ctx).iterdir()) == [], "a refused upload left a file behind"
    assert ("upload_refused", "disk_full") in peer.audited


async def test_the_reserve_is_never_handed_out(tmp_path, monkeypatch):
    """Exactly enough for the file, nothing for the reserve, is still a refusal:
    a disk filled to the last byte breaks the node's own databases too."""
    ctx = _group_ctx(tmp_path)
    peer = _peer(ctx)
    _free(monkeypatch, DISK_RESERVE_BYTES + 15)
    await peer._do_file_upload(sealed_upload(peer, filename="a.jpg",
                                             data=b"x" * 8, total_chunks=2))
    assert _codes(peer) == ["disk_full"]


async def test_room_beside_the_reserve_is_accepted(tmp_path, monkeypatch):
    ctx = _group_ctx(tmp_path)
    peer = _peer(ctx)
    _free(monkeypatch, DISK_RESERVE_BYTES + 16)
    await peer._do_file_upload(sealed_upload(peer, filename="a.jpg",
                                             data=b"x" * 8, total_chunks=2))
    assert _codes(peer) == []


@pytest.mark.parametrize("err", [errno.ENOSPC, getattr(errno, "EDQUOT", errno.ENOSPC)])
async def test_a_write_that_finds_no_room_is_refused_and_cleaned_up(tmp_path, monkeypatch, err):
    """The check above can be passed by two uploads at once, or by a disk the
    operator fills meanwhile. The write is the last word."""
    ctx = _group_ctx(tmp_path)
    peer = _peer(ctx)
    _free(monkeypatch, 100 * DISK_RESERVE_BYTES)
    await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"first",
                                             chunk_index=0, total_chunks=2))
    assert _codes(peer) == []

    def no_room(*_a):
        raise OSError(err, "No space left on device")
    monkeypatch.setattr(upload_handlers, "_append_chunk", no_room)
    await peer._do_file_upload(sealed_upload(peer, filename="film.mkv", data=b"second",
                                             chunk_index=1, total_chunks=2))
    assert _codes(peer) == ["disk_full"]
    assert list(_shared(ctx).iterdir()) == [], (
        "the .part was kept, holding the very space that ran out")
    assert ctx["partial_uploads"].get("user-1", "shared", "film.mkv") is None


async def test_another_write_failure_is_not_called_a_full_disk(tmp_path, monkeypatch):
    ctx = _group_ctx(tmp_path)
    peer = _peer(ctx)
    _free(monkeypatch, 100 * DISK_RESERVE_BYTES)

    def denied(*_a):
        raise OSError(errno.EACCES, "Permission denied")
    monkeypatch.setattr(upload_handlers, "_append_chunk", denied)
    with pytest.raises(OSError):
        await peer._do_file_upload(sealed_upload(peer, filename="a.jpg", data=b"x"))


async def test_the_real_free_space_is_what_is_read(tmp_path):
    """The un-patched path: a small file into a temp directory goes through."""
    ctx = _group_ctx(tmp_path)
    peer = _peer(ctx)
    if upload_handlers._free_bytes(_shared(ctx)) < DISK_RESERVE_BYTES + 1024:
        pytest.skip("this machine's temp filesystem is itself nearly full")
    await peer._do_file_upload(sealed_upload(peer, filename="a.jpg", data=b"hello"))
    assert _codes(peer) == []
    assert (_shared(ctx) / "a.jpg").read_bytes() == b"hello"