"""
`stream`: a file served on this machine to a media player — byte ranges from
the node's chunks, read without a lease, only to a request that names this
address and carries the token; the connection opened again when it dies, and
the stream stopped by the node's refusal, by Ctrl+C or by closing the player.
"""

import asyncio
import json
import os
import secrets
import stat
import sys
from urllib.parse import urlsplit

import pytest
from fake_hub import PASSPHRASE, USERNAME, FakeHub
from fake_node import FakeNode
from meshbay_cli import __main__ as cli
from meshbay_cli.client import relay
from meshbay_cli.client.relay import byte_range, content_type
from meshbay_cli.commands import context, stream
from meshbay_cli.errors import NODE, NOT_FOUND, USAGE, CliError

MiB = 1024 * 1024
FILM = secrets.token_bytes(3 * MiB + 12345)


@pytest.fixture
def world(monkeypatch, capsys):
    hub, node = FakeHub(), FakeNode()
    monkeypatch.setattr(context, "Hub", lambda url: hub)
    sessions = []

    async def open_group(account, group, **_):
        sessions.append(node.session())
        return sessions[-1]
    monkeypatch.setattr(stream, "open_group", open_group)
    monkeypatch.setattr(relay, "PING_EVERY", 0.05)
    monkeypatch.setattr(relay, "RECONNECT_WAITS", (0.01,))
    node.add("share/films", "Holiday.mkv", FILM)["type"] = "video"
    node.add("share/docs", "notes.txt", b"hello")
    assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
                     "login"]) == 0
    capsys.readouterr()

    def start(*argv):
        args = cli.build_parser().parse_args(list(argv))
        return asyncio.ensure_future(stream.run(context.Context(args)))
    return node, sessions, start


def node_film(node) -> dict:
    return next(e for e in node.files.values() if e["name"] == "Holiday.mkv")


async def _until(cond, timeout=5.0):
    deadline = asyncio.get_running_loop().time() + timeout
    while not cond():
        assert asyncio.get_running_loop().time() < deadline, "timed out"
        await asyncio.sleep(0.02)


async def _url(task, capsys, said: list | None = None) -> str:
    """The address printed on stdout; what was said on stderr meanwhile into `said`."""
    out = []

    def printed():
        assert not task.done(), task.exception()
        got = capsys.readouterr()
        out.extend(got.out.split())
        if said is not None:
            said.append(got.err)
        return out
    await _until(printed)
    return out[0]


async def http(url: str, *, method="GET", headers=None, host=None, path=None,
               keep=False, reader_writer=None):
    """One request; (status, headers, body). With `keep`, the connection is returned
    for another request."""
    u = urlsplit(url)
    reader, writer = reader_writer or await asyncio.open_connection(u.hostname, u.port)
    lines = [f"{method} {path or u.path} HTTP/1.1", f"Host: {host or u.netloc}",
             *(f"{k}: {v}" for k, v in (headers or {}).items())]
    if not keep:
        lines.append("Connection: close")
    writer.write(("\r\n".join(lines) + "\r\n\r\n").encode())
    await writer.drain()
    head = (await reader.readuntil(b"\r\n\r\n")).decode("latin-1").split("\r\n")
    status = int(head[0].split(" ")[1])
    fields = {k.strip().lower(): v.strip() for k, _, v in
              (line.partition(":") for line in head[1:] if line)}
    body = b""
    if method != "HEAD":
        body = await reader.readexactly(int(fields.get("content-length", 0)))
    if keep:
        return status, fields, body, (reader, writer)
    writer.close()
    return status, fields, body


# ── Ranges and types ─────────────────────────────────────────────────────────

def test_a_range_is_read_as_players_send_it():
    assert byte_range(None, 1000) is None
    assert byte_range("bytes=0-", 1000) == (0, 999)
    assert byte_range("bytes=10-19", 1000) == (10, 19)
    assert byte_range("bytes=990-5000", 1000) == (990, 999)
    assert byte_range("bytes=-100", 1000) == (900, 999)
    assert byte_range("bytes=-5000", 1000) == (0, 999)
    assert byte_range("bytes=0-1,5-9", 1000) is None        # several: the whole file
    assert byte_range("items=0-9", 1000) is None
    for bad in ("bytes=1000-", "bytes=20-10", "bytes=x-", "bytes=-0"):
        with pytest.raises(ValueError):
            byte_range(bad, 1000)


def test_a_file_is_served_as_audio_or_video_and_never_as_a_page():
    assert content_type("Film.MKV", "video") == "video/x-matroska"
    assert content_type("a.flac", "audio") == "audio/flac"
    assert content_type("trap.html", "video") == "video/octet-stream"
    assert content_type("trap.svg", "audio") == "audio/octet-stream"


# ── Serving ──────────────────────────────────────────────────────────────────

async def test_the_player_gets_the_file_and_any_range_of_it(world, capsys):
    node, sessions, start = world
    task = start("stream", "Family/share/films/Holiday.mkv")
    try:
        said = []
        url = await _url(task, capsys, said)
        assert url.startswith("http://127.0.0.1:") and url.endswith("/Holiday.mkv")
        assert "Ctrl+C to stop" in "".join(said)

        status, fields, body = await http(url)
        assert status == 200 and body == FILM
        assert fields["content-type"] == "video/x-matroska"
        assert fields["accept-ranges"] == "bytes"
        assert fields["x-content-type-options"] == "nosniff"

        # Across a chunk boundary, as a seek lands.
        a, b = MiB - 100, 2 * MiB + 50
        status, fields, body = await http(url, headers={"Range": f"bytes={a}-{b}"})
        assert status == 206 and body == FILM[a:b + 1]
        assert fields["content-range"] == f"bytes {a}-{b}/{len(FILM)}"

        # The end of the file, as a player probes for an index there; then the
        # same connection again.
        status, _, body, conn = await http(url, headers={"Range": "bytes=-64"}, keep=True)
        assert status == 206 and body == FILM[-64:]
        status, _, body, conn = await http(url, headers={"Range": "bytes=5-9"}, keep=True,
                                           reader_writer=conn)
        assert status == 206 and body == FILM[5:10]
        conn[1].close()

        status, fields, body = await http(url, method="HEAD")
        assert status == 200 and fields["content-length"] == str(len(FILM)) and body == b""
        status, fields, _ = await http(url, headers={"Range": f"bytes={len(FILM)}-"})
        assert status == 416 and fields["content-range"] == f"bytes */{len(FILM)}"

        # Read as a player reads, without a lease; chunks kept are not asked twice.
        assert node.leases == {} and node.leaseless_reads
        assert sorted(node.leaseless_reads) == [(node_film(node)["id"], i) for i in range(4)]
        assert len(sessions) == 1
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


async def test_only_this_address_with_its_token_is_answered(world, capsys):
    node, _, start = world
    task = start("stream", "Family/share/films/Holiday.mkv")
    try:
        url = await _url(task, capsys)
        port = urlsplit(url).port
        # A page that rebinds its own name to 127.0.0.1 sends its own Host.
        status, _, body = await http(url, host=f"evil.example:{port}")
        assert status == 403 and body == b""
        status, _, body = await http(url, path="/not-the-token/Holiday.mkv")
        assert status == 404 and body == b""
        status, _, _ = await http(url, method="POST")
        assert status == 405
        assert node.leaseless_reads == [], "nothing refused reached the node"
        status, _, body = await http(url, host=f"localhost:{port}",
                                     headers={"Range": "bytes=0-3"})
        assert status == 206 and body == FILM[:4]
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


async def test_a_dead_connection_is_opened_again(world, capsys):
    node, sessions, start = world
    task = start("stream", "Family/share/films/Holiday.mkv")
    try:
        url = await _url(task, capsys)
        assert (await http(url, headers={"Range": "bytes=0-9"}))[2] == FILM[:10]

        sessions[-1].ch.closed.set()                    # the link goes down
        await _until(lambda: len(sessions) == 2)
        a = 3 * MiB
        assert (await http(url, headers={"Range": f"bytes={a}-"}))[2] == FILM[a:]

        node.unresponsive = True                        # it stops answering pings
        await _until(lambda: len(sessions) == 3)
        node.unresponsive = False
        assert "reconnecting" in capsys.readouterr().err
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


async def test_a_refusal_by_the_node_stops_the_stream(world, capsys):
    node, _, start = world
    node.fail_chunk = (node_film(node)["id"], 2)
    task = start("stream", "Family/share/films/Holiday.mkv")
    url = await _url(task, capsys)
    status, _, _ = await http(url, headers={"Range": f"bytes={2 * MiB}-"})
    assert status == 502
    with pytest.raises(CliError) as e:
        await asyncio.wait_for(task, 5)
    assert e.value.code == NODE and "drive stopped answering" in e.value.message


async def test_what_cannot_be_streamed_is_said(world, capsys):
    _, _, start = world
    for argv, code, says in [
        (["Family/share/docs/notes.txt"], USAGE, "not a video or a sound file"),
        (["Family/share/films"], USAGE, "is a folder"),
        (["Family/share/films/Nope.mkv"], NOT_FOUND, "Nothing"),
        (["Family"], USAGE, "Name the file"),
        (["--player", "no-such-player-here", "Family/share/films/Holiday.mkv"], USAGE,
         "was not found"),
    ]:
        with pytest.raises(CliError) as e:
            await start("stream", *argv)
        assert e.value.code == code and says in e.value.message, argv


# ── --player ─────────────────────────────────────────────────────────────────

def _player(tmp_path, body: str) -> str:
    script = tmp_path / "player"
    script.write_text(f"#!{sys.executable}\nimport os, sys, time, urllib.request\n{body}\n")
    script.chmod(script.stat().st_mode | stat.S_IXUSR)
    return str(script)


@pytest.mark.skipif(sys.platform == "win32", reason="a script as a program")
async def test_closing_the_player_stops_the_stream(world, capsys, tmp_path):
    out = tmp_path / "played"
    player = _player(tmp_path, (
        "opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))\n"
        f"open({str(out)!r}, 'wb').write(opener.open(sys.argv[1]).read())"))
    _, _, start = world
    code = await asyncio.wait_for(
        start("stream", "--player", player, "Family/share/films/Holiday.mkv"), 10)
    assert code == 0 and out.read_bytes() == FILM
    assert "player was closed" in capsys.readouterr().err


@pytest.mark.skipif(sys.platform == "win32", reason="a script as a program")
async def test_stopping_the_stream_closes_the_player_it_started(world, capsys, tmp_path):
    pid = tmp_path / "pid"
    player = _player(tmp_path, f"open({str(pid)!r}, 'w').write(str(os.getpid()))\n"
                               "time.sleep(60)")
    _, _, start = world
    task = start("--json", "stream", "--player", player, "Family/share/films/Holiday.mkv")
    await _until(lambda: pid.exists() and pid.read_text())
    shown = json.loads(capsys.readouterr().out)
    assert shown["name"] == "Holiday.mkv" and shown["size"] == len(FILM)
    task.cancel()                                       # Ctrl+C
    await asyncio.gather(task, return_exceptions=True)

    def gone():
        try:
            os.kill(int(pid.read_text()), 0)
        except ProcessLookupError:
            return True
        with open(f"/proc/{pid.read_text()}/stat") as f:
            return f.read().split()[2] == "Z"           # ended, not yet reaped
    await _until(gone)