"""The operator's one lever on the picture, and the two locks in front of it.

`/client-stats` is one-way BY CONSTRUCTION — nothing is read back and the server cannot answer a
beacon, so it can never become a route for telling a RENDERER what to do. That property is about the
renderer channel and is asserted here to stay true. This is a different thing sitting beside it: the
producer being told, by a person, that a pause has turned out to be long. It writes no gun, no stop
and no result, and it goes through the same gate that governs cutting an OBS scene.
"""
from __future__ import annotations

import json
import sys
import threading
import urllib.error
import urllib.request
from pathlib import Path

import pytest

_LIVE = Path(__file__).resolve().parents[2] / "live"
if str(_LIVE) not in sys.path:
    sys.path.insert(0, str(_LIVE))

from live_feed_server import LiveFeedServer  # noqa: E402


class FakeWaiting:
    """The waiting picture, reduced to the two calls the route makes."""

    def __init__(self):
        self.asked: list[bool] = []

    def excursion_state(self) -> dict:
        return {"armed": True, "flying": False, "available": True, "requested": None}

    def request_excursion(self, *, fly: bool = True) -> dict:
        self.asked.append(fly)
        return {"status": "queued", "flying": False, "armed": True}


class FakeOnAirRace:
    def __init__(self):
        self.requests = 0

    def course_guide_state(self) -> dict:
        return {"status": "ready", "instanceId": "race-1"}

    def request_course_guide(self) -> dict:
        self.requests += 1
        return {"status": "queued", "instanceId": "race-1"}


@pytest.fixture
def srv():
    server = LiveFeedServer(port=0)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    try:
        yield server
    finally:
        server.shutdown()
        server.server_close()


@pytest.fixture(autouse=True)
def _allowed(monkeypatch):
    """The normal case: this IS the broadcasting box, and a token is configured."""
    monkeypatch.setenv("BROADCAST_ALLOW_CONTROL", "1")
    monkeypatch.setenv("VR_BROADCAST_CONTROL_TOKEN", "s3cret")
    monkeypatch.delenv("VR_FEED_TOKEN", raising=False)


def _call(srv, *, send=None, auth="s3cret", path="/control/excursion"):
    url = f"http://127.0.0.1:{srv.server_address[1]}{path}"
    data = json.dumps(send).encode() if send is not None else None
    req = urllib.request.Request(url, data=data, method="POST" if data is not None else "GET")
    if auth:
        req.add_header("Authorization", "Bearer " + auth)
    try:
        with urllib.request.urlopen(req, timeout=5) as r:
            return r.status, r.read()
    except urllib.error.HTTPError as e:
        return e.code, e.read() or b""


def _json(srv, **kw):
    code, body = _call(srv, **kw)
    return code, (json.loads(body) if body else {})


def test_an_operator_can_fly_the_excursion_and_call_it_home(srv):
    waiting = FakeWaiting()
    srv.excursion = waiting
    code, doc = _json(srv)
    assert code == 200 and doc["armed"] is True                    # the read
    assert _json(srv, send={"fly": True})[0] == 200
    assert _json(srv, send={"fly": False})[0] == 200
    assert waiting.asked == [True, False]


def test_a_post_with_no_body_means_fly(srv):
    """The common case is a button, and a button that has to send a payload to mean the obvious thing
    is a button somebody will wire up wrong."""
    waiting = FakeWaiting()
    srv.excursion = waiting
    assert _json(srv, send={})[0] == 200
    assert waiting.asked == [True]


def test_the_lever_is_shut_unless_this_is_the_broadcasting_box(srv, monkeypatch):
    """`BROADCAST_ALLOW_CONTROL` is default OFF and an explicit act per deployment — it is what stops
    a dev checkout from flying the camera on a live show."""
    monkeypatch.delenv("BROADCAST_ALLOW_CONTROL", raising=False)
    srv.excursion = FakeWaiting()
    code, doc = _json(srv)
    assert code == 503 and doc["hint"] == "BROADCAST_ALLOW_CONTROL"


def test_an_unconfigured_control_is_a_closed_door(srv, monkeypatch):
    """The OPPOSITE default to the feed's, deliberately. An unset `VR_FEED_TOKEN` means the feed is
    open, which is what a local run relies on; an unset control token must never mean the same."""
    monkeypatch.delenv("VR_BROADCAST_CONTROL_TOKEN", raising=False)
    srv.excursion = FakeWaiting()
    code, doc = _json(srv)
    assert code == 503 and doc["hint"] == "VR_BROADCAST_CONTROL_TOKEN"


def test_the_feed_token_is_not_the_control_token(srv, monkeypatch):
    """A partner rendering the picture for themselves holds the feed token. Being allowed to READ the
    feed must never amount to permission to STEER it."""
    monkeypatch.setenv("VR_FEED_TOKEN", "the-feed-token")
    srv.excursion = FakeWaiting()
    assert _json(srv, auth="the-feed-token")[0] == 401


def test_no_token_at_all_is_refused(srv):
    srv.excursion = FakeWaiting()
    assert _json(srv, auth="")[0] == 401


def test_with_a_race_on_air_there_is_nothing_to_steer(srv):
    """409, not 404: the route exists, the state does not. The producer clears the hook when a race
    takes the air, so the lever cannot steer a picture nobody is looking at."""
    srv.excursion = None
    code, doc = _json(srv)
    assert code == 409 and doc["code"] == "not-waiting"


def test_an_operator_can_request_a_course_guide_over_the_on_air_race(srv):
    race = FakeOnAirRace()
    srv.course_guide = race
    code, state = _json(srv, path="/control/course-guide")
    assert code == 200 and state["status"] == "ready"
    code, state = _json(srv, send={}, path="/control/course-guide")
    assert code == 200 and state["status"] == "queued"
    assert race.requests == 1


def test_course_guide_control_refuses_when_no_race_is_on_air(srv):
    code, state = _json(srv, send={}, path="/control/course-guide")
    assert code == 409 and state["code"] == "no-on-air-race"


def test_the_renderer_beacon_is_still_one_way(srv):
    """The property this route sits beside and must not erode: a beacon is answered with a status and
    NO BODY, so nothing can be read back through it — it can never become a route for telling a
    renderer what to do."""
    code, body = _call(srv, send={"fps": 30}, auth="", path="/client-stats")
    assert code in (204, 400) and body == b""


def test_the_control_path_is_not_reachable_as_a_static_file(srv):
    """The server also serves the broadcast tree. A control path that fell through to the static
    handler would be a 404 that looks like a missing feature rather than a shut door."""
    code, _body = _call(srv, auth="", path="/control/excursion")
    assert code == 401, "the gate answered, not the file server"


def _publish_camera_race(srv):
    teams = [{"boats": [
        {"deviceId": "boat-one", "sailId": "1"},
        {"deviceId": "boat-two", "sailId": "2"},
        {"deviceId": "boat-three", "sailId": "3"},
    ]}]
    srv.publish_init({"instanceId": "heat-1", "gunUtcMs": 1000, "trackStartUtcMs": 0},
                     {}, teams)


def test_an_operator_can_assign_a_sail_number_to_a_camera_slot(srv):
    _publish_camera_race(srv)
    code, doc = _json(srv, send={"slot": "B", "boat": "3"}, path="/control/camera-slot")
    assert code == 200
    assert doc["deviceId"] == "boat-three"
    assert doc["cameraSlots"] == ["boat-one", "boat-three", "boat-two"]
    code, state = _json(srv, path="/control/camera-slot")
    assert code == 200 and state["assignments"]["B"] == "boat-three"


def test_camera_assignment_rejects_unknown_boats_and_slots(srv):
    _publish_camera_race(srv)
    code, doc = _json(srv, send={"slot": "Z", "boat": "2"}, path="/control/camera-slot")
    assert code == 409 and doc["code"] == "invalid-camera-assignment"
    code, doc = _json(srv, send={"slot": "B", "boat": "99"}, path="/control/camera-slot")
    assert code == 409 and "not found" in doc["error"]


def test_a_boat_without_current_media_can_be_put_in_a_slot_while_it_wakes(srv):
    teams = [{"boats": [{"deviceId": "awake", "sailId": "1"},
                         {"deviceId": "sleeping", "sailId": "2"}]}]
    srv.publish_init({"instanceId": "heat-1", "availableOnboardCameraBoatIds": ["awake"]},
                     {}, teams)
    code, doc = _json(srv, send={"slot": "A", "boat": "2"}, path="/control/camera-slot")
    assert code == 200 and doc["cameraSlots"] == ["sleeping"]


def test_camera_assignment_survives_header_refresh_but_not_a_new_heat(srv):
    _publish_camera_race(srv)
    _json(srv, send={"slot": "B", "boat": "3"}, path="/control/camera-slot")
    _publish_camera_race(srv)
    assert srv.camera_slot_state()["cameraSlots"] == ["boat-one", "boat-three", "boat-two"]
    teams = [{"boats": [{"deviceId": "new-one", "sailId": "1"},
                         {"deviceId": "new-two", "sailId": "2"}]}]
    srv.publish_init({"instanceId": "heat-2", "gunUtcMs": 2000, "trackStartUtcMs": 1000}, {}, teams)
    assert srv.camera_slot_state()["cameraSlots"] == ["new-one", "new-two"]


def test_camera_assignment_uses_the_same_control_locks(srv, monkeypatch):
    _publish_camera_race(srv)
    monkeypatch.delenv("BROADCAST_ALLOW_CONTROL", raising=False)
    code, doc = _json(srv, send={"slot": "B", "boat": "2"}, path="/control/camera-slot")
    assert code == 503 and doc["hint"] == "BROADCAST_ALLOW_CONTROL"