"""The embedded console server: a real socket, and the deadman it feeds.

These start the server for real rather than faking it. The thing worth checking
is the boundary between the two threads — a browser's message reaching the tick,
and a browser going away reaching the deadman — and a fake on either side of
that boundary would check nothing.
"""

import asyncio
import json
import socket
import time

import aiohttp
import pytest

from dora_teleop import main as loop
from dora_teleop.arbiter import WEB
from dora_teleop.config import Config, ConsoleConfig
from dora_teleop.server import Console
from tests.conftest import RecordingNode


def free_port():
    """Ask the OS for a port nobody is using."""
    with socket.socket() as held:
        held.bind(("127.0.0.1", 0))
        return held.getsockname()[1]


@pytest.fixture
def console():
    """Run a real server on a real port, and stop it however the test ends."""
    running = Console("127.0.0.1", free_port())
    running.start()
    try:
        yield running
    finally:
        running.stop()


def visit(console, exchange):
    """Open one browser's socket against the server and run `exchange` on it."""

    async def session():
        async with aiohttp.ClientSession() as client:
            url = f"http://{console.host}:{console.port}"
            async with client.ws_connect(f"{url}/ws") as socket_:
                return await exchange(client, url, socket_)

    return asyncio.run(session())


def wait_for(predicate):
    """Block until the server thread catches up, rather than guessing at a sleep."""
    for _ in range(200):
        if predicate():
            return
        time.sleep(0.01)
    raise AssertionError("the server never got there")


def test_the_page_is_served_at_the_root(console):
    async def exchange(client, url, socket_):
        async with client.get(url) as answer:
            return answer.status, await answer.text()

    status, body = visit(console, exchange)

    assert status == 200
    assert "teleop console" in body


def test_a_message_from_a_browser_reaches_the_tick(console):
    # The console's half of the boundary: what the page sends has to be waiting
    # on the queue the next tick drains, in the shape the loop reads.
    async def exchange(client, url, socket_):
        await socket_.send_str(json.dumps({"type": "jog", "axes": [1.0, 0, 0, 0, 0, 0]}))
        await asyncio.sleep(0.05)

    visit(console, exchange)

    assert console.take() == [{"type": "jog", "axes": [1.0, 0, 0, 0, 0, 0]}]
    assert console.take() == []


def test_a_snapshot_reaches_the_browser(console):
    # The other half: a telemetry object handed over from the dora thread comes
    # out of the socket as the page's own JSON, with no envelope around it.
    async def exchange(client, url, socket_):
        console.broadcast({"type": "telemetry", "arm_state": "idle"})
        return json.loads(await socket_.receive_str())

    assert visit(console, exchange) == {"type": "telemetry", "arm_state": "idle"}


def test_a_browser_that_is_not_talking_json_is_ignored(console):
    # A page from a newer build saying something we cannot parse must not take
    # the arm down, and must not stop the messages after it being read.
    async def exchange(client, url, socket_):
        await socket_.send_str("{not json")
        await socket_.send_str(json.dumps({"type": "estop"}))
        await asyncio.sleep(0.05)

    visit(console, exchange)

    assert console.take() == [{"type": "estop"}]


def test_the_server_counts_the_browsers_on_it(console):
    assert console.clients == 0
    seen = []

    async def exchange(client, url, socket_):
        await asyncio.sleep(0.05)
        seen.append(console.clients)

    visit(console, exchange)

    assert seen == [1]

    # And the socket closing is what the deadman below reads.
    wait_for(lambda: console.clients == 0)


def test_a_flood_from_one_browser_cannot_grow_without_bound(console):
    # An unbounded queue would answer a flood by growing until the node died,
    # and the node dying is the one thing that must not stop the arm from being
    # stopped.
    async def exchange(client, url, socket_):
        for index in range(600):
            await socket_.send_str(json.dumps({"type": "jog_stop", "n": index}))
        await asyncio.sleep(0.2)

    visit(console, exchange)

    assert 0 < len(console.take()) <= 256


# --- where the server meets the deadman ---


def build():
    config = Config.from_env()
    config.expo = 0.0
    return RecordingNode(), loop.Teleop(config)


def test_the_last_browser_leaving_takes_the_arm_back(console):
    # The web deadman used to arrive as a `clients` message from the bridge. It
    # is now read straight off the server, and it has to mean the same thing.
    node, state = build()
    state.console = console
    state.arbiter.take(WEB, 1000.0)

    loop._drain_console(node, state)

    assert state.clients == 0
    assert state.arbiter.owner is None
    assert state.web_axes == [0.0] * 6


def test_a_node_with_no_console_is_left_alone():
    # A pad-only dataflow opens no port, and the tick must not read a lease
    # release out of a server that was never started.
    node, state = build()
    state.clients = 1
    state.arbiter.take(WEB, 1000.0)

    loop._drain_console(node, state)

    assert state.arbiter.owner == WEB


# --- the port itself ---


def test_the_console_is_on_by_default(monkeypatch):
    monkeypatch.delenv("WEB_PORT", raising=False)
    settings = ConsoleConfig.from_env()

    assert settings.enabled is True
    # Pinned because it is the number in the README, the manifest and every
    # dataflow: a default that drifts from the docs sends an operator to a port
    # nothing is listening on.
    assert settings.port == 7000


def test_port_zero_means_no_port_is_opened(monkeypatch):
    # How `dataflow-gamepad.yml` asks for a pad and nothing else.
    monkeypatch.setenv("WEB_PORT", "0")
    assert ConsoleConfig.from_env().enabled is False


def test_a_port_that_is_not_a_port_stops_the_node(monkeypatch):
    monkeypatch.setenv("WEB_PORT", "99999")
    with pytest.raises(ValueError, match="WEB_PORT"):
        ConsoleConfig.from_env()


def test_a_port_that_is_not_a_number_stops_the_node(monkeypatch):
    monkeypatch.setenv("WEB_PORT", "eight-thousand")
    with pytest.raises(ValueError, match="WEB_PORT"):
        ConsoleConfig.from_env()


def test_a_port_already_in_use_stops_the_node(console):
    # Found the hard way: the bind failed on the server thread, `start` waited
    # out its timeout and then announced a console anyway, and the node ran on
    # serving nothing. An operator opening the URL got silence while the log
    # said it was up.
    second = Console(console.host, console.port)

    with pytest.raises(OSError):
        second.start()

    second.stop()


def test_stopping_a_console_that_never_started_is_harmless():
    # `main` stops the console in a `finally`, which runs whether or not the
    # server ever came up.
    Console("127.0.0.1", 0).stop()