"""Downloads and uploads against a node in memory (`fake_node.py`)."""

import io
import json
import secrets

import pytest
from fake_node import CHUNK, FakeNode
from meshbay_cli.client.names import UnsafeName, local_path, upload_name_problem
from meshbay_cli.client.transfers import (
    PART_SUFFIX,
    UPLOAD_CHUNK,
    TransferError,
    create_dir,
    download,
    upload,
)


def _reader(data: bytes):
    buf = io.BytesIO(data)

    def read(offset, length):
        buf.seek(offset)
        return buf.read(length)
    return read


# ── Download ─────────────────────────────────────────────────────────────────

async def test_a_file_of_several_chunks_arrives_whole(tmp_path):
    node = FakeNode()
    data = secrets.token_bytes(2 * CHUNK + 12345)
    entry = node.add("share", "big.bin", data)
    target = tmp_path / "big.bin"
    seen = []
    assert await download(node.session(), entry, target, progress=seen.append) == len(data)
    assert target.read_bytes() == data
    assert sum(seen) == len(data)
    assert not list(tmp_path.glob(f"*{PART_SUFFIX}*"))
    assert node.leases == {}, "the lease is closed at the end"


async def test_an_interrupted_download_resumes_from_its_last_whole_chunk(tmp_path):
    node = FakeNode()
    data = secrets.token_bytes(3 * CHUNK + 7)
    entry = node.add("share", "big.bin", data)
    target = tmp_path / "big.bin"
    part = tmp_path / f"big.bin{PART_SUFFIX}"
    # A chunk and a half on disk: only the whole chunk counts.
    part.write_bytes(data[:CHUNK + CHUNK // 2])
    (tmp_path / f"big.bin{PART_SUFFIX}.json").write_text(
        json.dumps({"id": entry["id"], "size": entry["size"]}))
    await download(node.session(), entry, target)
    assert target.read_bytes() == data
    assert node.chunk_requests == 3


async def test_a_part_of_another_file_is_started_over(tmp_path):
    node = FakeNode()
    data = secrets.token_bytes(2 * CHUNK)
    entry = node.add("share", "big.bin", data)
    (tmp_path / f"big.bin{PART_SUFFIX}").write_bytes(b"x" * CHUNK)
    (tmp_path / f"big.bin{PART_SUFFIX}.json").write_text(
        json.dumps({"id": "0" * 64, "size": entry["size"]}))
    await download(node.session(), entry, tmp_path / "big.bin")
    assert (tmp_path / "big.bin").read_bytes() == data
    assert node.chunk_requests == 2


async def test_a_refused_chunk_fails_the_file_and_keeps_the_part(tmp_path):
    node = FakeNode()
    entry = node.add("share", "big.bin", secrets.token_bytes(3 * CHUNK))
    node.fail_chunk = (entry["id"], 2)
    with pytest.raises(TransferError) as e:
        await download(node.session(), entry, tmp_path / "big.bin")
    assert "chunk 2" in e.value.message and "stopped answering" in e.value.message
    assert not (tmp_path / "big.bin").exists()
    assert (tmp_path / f"big.bin{PART_SUFFIX}").stat().st_size == 2 * CHUNK
    assert node.leases == {}


async def test_a_file_that_appeared_meanwhile_is_not_replaced(tmp_path):
    node = FakeNode()
    entry = node.add("share", "a.txt", b"from the group")
    (tmp_path / "a.txt").write_bytes(b"mine")
    with pytest.raises(TransferError):
        await download(node.session(), entry, tmp_path / "a.txt")
    assert (tmp_path / "a.txt").read_bytes() == b"mine"
    await download(node.session(), entry, tmp_path / "a.txt", overwrite=True)
    assert (tmp_path / "a.txt").read_bytes() == b"from the group"


async def test_a_queued_lease_waits_for_its_grant(tmp_path):
    node = FakeNode()
    node.queue_first = 1
    entry = node.add("share", "a.txt", b"hello")
    queued = []
    await download(node.session(), entry, tmp_path / "a.txt", on_queued=queued.append)
    assert queued == [1] and (tmp_path / "a.txt").read_bytes() == b"hello"


async def test_an_interrupted_download_keeps_its_part_and_closes_its_lease(tmp_path):
    import asyncio
    node = FakeNode()
    node.chunk_delay = 0.05
    data = secrets.token_bytes(20 * CHUNK)
    entry = node.add("share", "big.bin", data)
    task = asyncio.ensure_future(download(node.session(), entry, tmp_path / "big.bin"))
    while node.chunk_requests < 12:
        await asyncio.sleep(0.01)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    assert node.leases == {}
    assert not (tmp_path / "big.bin").exists()
    kept = (tmp_path / f"big.bin{PART_SUFFIX}").stat().st_size
    assert kept > 0 and kept % CHUNK == 0
    node.chunk_delay = 0
    await download(node.session(), entry, tmp_path / "big.bin")
    assert (tmp_path / "big.bin").read_bytes() == data


# ── Upload ───────────────────────────────────────────────────────────────────

async def test_an_upload_of_several_chunks_is_stored_whole():
    node = FakeNode()
    data = secrets.token_bytes(5 * UPLOAD_CHUNK + 99)
    stored = await upload(node.session(), _reader(data), len(data), "scan.pdf", "share")
    assert stored == "scan.pdf"
    assert node.stored("share/scan.pdf") == data
    assert node.leases == {}


async def test_an_interrupted_upload_resumes_where_the_node_stopped():
    node = FakeNode()
    data = secrets.token_bytes(4 * UPLOAD_CHUNK)
    node.uploads[("share", "scan.pdf")] = {"next": 2, "data": data[:2 * UPLOAD_CHUNK]}
    sent = []
    await upload(node.session(), _reader(data), len(data), "scan.pdf", "share",
                 progress=sent.append)
    assert node.stored("share/scan.pdf") == data
    assert sum(sent) == len(data)


async def test_a_name_taken_meanwhile_is_reported():
    node = FakeNode()
    node.add("share", "scan.pdf", b"theirs")
    stored = await upload(node.session(), _reader(b"mine"), 4, "scan.pdf", "share")
    assert stored == "scan (1).pdf"
    assert node.stored("share/scan.pdf") == b"theirs"


async def test_a_full_disk_is_fatal():
    node = FakeNode()
    node.refuse_upload_code = "disk_full"
    with pytest.raises(TransferError) as e:
        await upload(node.session(), _reader(b"x"), 1, "a.txt", "share")
    assert e.value.fatal


async def test_folders_are_created_where_asked():
    node = FakeNode()
    assert await create_dir(node.session(), "share", "Scans") == "share/Scans"
    assert "share/Scans" in node.dirs


# ── Names ────────────────────────────────────────────────────────────────────

NAMES = [
    "photo.jpg", "Été 2024", "東京", "track (1).flac", "[live] set", "a_b-c",
    "rock & roll", "l’été", "x+y#z@w", "9lives",
    ".DS_Store", "_private", "-dash", " space", "trailing ", "trailing.",
    "a/b", "a\\b", "semi;colon", "x" * 128, "x" * 129, "", "tab\tname",
]


def test_the_cli_refuses_exactly_the_names_the_node_refuses():
    roots = pytest.importorskip("meshbay_node.roots")
    differ = [n for n in NAMES
              if bool(roots.SAFE_UPLOAD_NAME.match(n)) != (upload_name_problem(n, is_dir=True)
                                                           is None)]
    assert not differ, f"the CLI and the node disagree on: {differ}"
    for n in ["desktop.ini", "x.LNK", "a.url", "b.scf", "c.library-ms", "d.searchConnector-ms"]:
        assert roots.shell_active(n) and upload_name_problem(n), n


def test_a_downloaded_name_is_made_writable_and_stays_inside(tmp_path):
    path, renamed = local_path(tmp_path, ["evil\x1b[31mred.txt"])
    assert renamed and path.parent == tmp_path and "\x1b" not in path.name
    path, renamed = local_path(tmp_path, ["CON.txt"])
    assert renamed and path.name != "CON.txt"
    path, renamed = local_path(tmp_path, ["été", "plage de l'ouest.jpg"])
    assert not renamed and path == tmp_path / "été" / "plage de l'ouest.jpg"
    path, _ = local_path(tmp_path, [".."])
    assert path.parent == tmp_path


def test_a_link_out_of_the_destination_is_refused(tmp_path):
    outside = tmp_path / "outside"
    outside.mkdir()
    dest = tmp_path / "dest"
    dest.mkdir()
    (dest / "sub").symlink_to(outside)
    with pytest.raises(UnsafeName):
        local_path(dest, ["sub", "x.txt"])