"""Taking a replay on air — and, above all, giving the programme back.

The cut itself is `test_day_cutter.py`. What is tested here is the sequence around it, because the
sequence is where a bad minute becomes a bad hour: OBS mutes an INPUT rather than a scene item, so a
replay can only carry its own commentary if the programme is silenced under it — and anything that
silences the programme has to restore it on EVERY path out, including the one where the process dies.
"""
from __future__ import annotations

import json
import sys
import threading
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))

import replay_take  # noqa: E402
from replay_take import Cue, CueLog, CueReader, ReplayTake, ReplayTakeError, _refusal  # noqa: E402

T0 = 1_755_600_000_000


class FakeObs:
    """Records what was done to OBS, in order — the order is half the contract."""

    def __init__(self, *, fail_on_scene: str | None = None):
        self.muted = False
        self.scene = replay_take.PROGRAMME_SCENE
        self.log: list[str] = []
        self._fail_on_scene = fail_on_scene

    def connect(self):
        return self

    def input_muted(self, name: str) -> bool:
        return self.muted

    def set_input_muted(self, name: str, muted: bool) -> None:
        self.muted = bool(muted)
        self.log.append(f"{'mute' if muted else 'unmute'} {name}")

    def set_scene(self, name: str) -> None:
        self.log.append(f"scene {name}")
        if self._fail_on_scene and name == self._fail_on_scene:
            raise RuntimeError("obs refused")
        self.scene = name


def _taker(tmp_path, obs, *, cutter=None, slept=None):
    return ReplayTake(
        anchor=tmp_path / "anchor.jsonl", replay_dir=tmp_path,
        obs=obs, now_ms=lambda: T0,
        sleep=(slept.append if slept is not None else (lambda _s: None)),
        cutter=cutter or (lambda *a, **k: {"durationS": 40.0, "pieces": ["d:/rec/a.mp4"]}),
        on_log=lambda _m: None,
    )


# --- the sequence ---------------------------------------------------------------------------------

def test_a_take_silences_the_programme_rolls_and_hands_it_back(tmp_path):
    """The clip carries the commentary from the moment it is a replay OF. Two voices over one picture
    is what the mute prevents; a broadcast left silent is what the unmute prevents."""
    obs = FakeObs()
    slept: list[float] = []
    state = _taker(tmp_path, obs, slept=slept).take(from_utc_ms=T0 - 40_000, to_utc_ms=T0)

    assert obs.log == [f"mute {replay_take.PROGRAMME_INPUT}",
                       f"scene {replay_take.REPLAY_SCENE}",
                       f"scene {replay_take.PROGRAMME_SCENE}",
                       f"unmute {replay_take.PROGRAMME_INPUT}"]
    assert obs.muted is False
    assert obs.scene == replay_take.PROGRAMME_SCENE
    assert slept == [pytest.approx(40.0 + replay_take.RETURN_PAD_S)]
    assert state.status == "idle"


def test_the_picture_comes_back_before_the_voice(tmp_path):
    """Unmuting while the clip is still up would put the live commentary over the replayed picture —
    briefly, but that is exactly the pairing this module exists to prevent."""
    obs = FakeObs()
    _taker(tmp_path, obs).take(from_utc_ms=T0 - 40_000, to_utc_ms=T0)
    assert obs.log.index(f"scene {replay_take.PROGRAMME_SCENE}") < obs.log.index(
        f"unmute {replay_take.PROGRAMME_INPUT}")


def test_obs_refusing_the_cut_back_still_restores_the_audio(tmp_path):
    """The worst available outcome is a silent broadcast, so the unmute must not be behind anything
    that can throw. If OBS will not change scene, the picture is wrong for a moment; if the audio is
    never restored, the championship is silent until somebody notices."""
    obs = FakeObs(fail_on_scene=replay_take.PROGRAMME_SCENE)
    _taker(tmp_path, obs).take(from_utc_ms=T0 - 40_000, to_utc_ms=T0)
    assert obs.muted is False
    assert obs.log[-1] == f"unmute {replay_take.PROGRAMME_INPUT}"


def test_a_failed_cut_never_touches_the_picture(tmp_path):
    """Nothing goes on air until there is a verified clip to put there."""
    from day_cutter import DayCutterError

    obs = FakeObs()

    def refuse(*_a, **_k):
        raise DayCutterError("the recording only reaches …")

    taker = _taker(tmp_path, obs, cutter=refuse)
    with pytest.raises(ReplayTakeError, match="only reaches"):
        taker.take(from_utc_ms=T0 - 40_000, to_utc_ms=T0)
    assert obs.log == []
    assert taker.state.status == "failed"


def test_a_second_take_while_one_is_rolling_is_refused(tmp_path):
    """Two replays at once would cross their own mute/unmute pairs and could leave the programme
    muted by the one that finished first."""
    obs = FakeObs()
    started, release = threading.Event(), threading.Event()
    taker = _taker(tmp_path, obs, slept=None)
    taker._sleep = lambda _s: (started.set(), release.wait(5))     # noqa: SLF001 — hold it mid-take

    thread = threading.Thread(target=taker.take,
                              kwargs={"from_utc_ms": T0 - 40_000, "to_utc_ms": T0}, daemon=True)
    thread.start()
    assert started.wait(5)
    with pytest.raises(ReplayTakeError, match="already rolling"):
        taker.take(from_utc_ms=T0 - 10_000, to_utc_ms=T0)
    release.set()
    thread.join(5)
    assert obs.muted is False


def test_start_up_restores_audio_a_killed_process_could_not(tmp_path):
    """A `finally` does not run when a process is killed, and the state it would have restored lives
    in OBS rather than here."""
    obs = FakeObs()
    obs.muted = True                                              # as a crash mid-take would leave it
    assert _taker(tmp_path, obs).restore_audio() is True
    assert obs.muted is False


def test_start_up_survives_obs_not_being_up_yet(tmp_path):
    class Dead:
        def connect(self):
            raise OSError("connection refused")

    assert _taker(tmp_path, Dead()).restore_audio() is False      # reported, not raised


# --- the window -----------------------------------------------------------------------------------

def test_a_cue_is_cut_around_its_own_instant(tmp_path):
    """The cue already carries PRESENTATION time — the producer adds the broadcast delay before
    publishing, because the recording holds the picture as it went out rather than the water as it
    happened. So the window is simply lead/trail around it."""
    taker = _taker(tmp_path, FakeObs())
    cue = Cue(id="r/1", kind="mark_rounding", label="Mark 3", utcMs=T0)
    start, end = taker.window_for_cue(cue, lead_s=25.0, trail_s=15.0)
    assert (start, end) == (T0 - 25_000, T0 + 15_000)


def test_the_last_n_seconds_stops_short_of_the_write_frontier(tmp_path):
    """The encoder holds a little, so the literal current instant is not on disk yet."""
    taker = _taker(tmp_path, FakeObs())
    start, end = taker.window_for_lookback(40.0)
    assert end == T0 - replay_take.WRITE_MARGIN_MS
    assert end - start == 40_000


# --- the locks ------------------------------------------------------------------------------------

def test_the_two_locks_refuse_different_mistakes(monkeypatch):
    """"Not enabled on this box" and "wrong key" need opposite actions, and one 403 for both sends
    somebody to check the wrong one. The FEED token is deliberately not accepted: being allowed to
    read the broadcast is not permission to change the picture."""
    monkeypatch.delenv("BROADCAST_ALLOW_CONTROL", raising=False)
    monkeypatch.delenv("VR_BROADCAST_CONTROL_TOKEN", raising=False)
    assert _refusal()["hint"] == "BROADCAST_ALLOW_CONTROL"

    monkeypatch.setenv("BROADCAST_ALLOW_CONTROL", "1")
    assert _refusal()["hint"] == "VR_BROADCAST_CONTROL_TOKEN"

    monkeypatch.setenv("VR_BROADCAST_CONTROL_TOKEN", "s3cret")
    assert _refusal() is None
    assert replay_take._authorised({"Authorization": "Bearer s3cret"}) is True
    assert replay_take._authorised({"Authorization": "Bearer wrong"}) is False
    assert replay_take._authorised({}) is False


# --- the cues, which are suggestions ---------------------------------------------------------------

def test_cues_are_filed_deduplicated_and_bounded():
    log = CueLog(limit=3)
    for i in range(5):
        log.add(Cue(id=f"r/{i}", kind="penalty", label=f"p{i}", utcMs=T0 + i))
    log.add(Cue(id="r/4", kind="penalty", label="again", utcMs=T0))          # the same moment twice
    assert [c.id for c in log.recent()] == ["r/4", "r/3", "r/2"]              # newest first


def test_the_reader_files_a_cue_off_the_feed_and_ignores_everything_else():
    log = CueLog()
    payload = (b"event: delta\ndata: {}\n\n"
               b"event: replay_cue\n"
               b'data: {"id":"r/900/mark_rounding","kind":"mark_rounding","label":"Mark 3",'
               b'"utcMs":1755600000000,"race":"Race 1009"}\n\n'
               b"event: replay_cue\ndata: not json\n\n"
               b"event: replay_cue\ndata: {\"kind\":\"penalty\"}\n\n")            # no id, no instant

    class Stream:
        def __enter__(self):
            return payload.splitlines(keepends=True)

        def __exit__(self, *a):
            return False

    reader = CueReader("http://producer", log, open_url=lambda _u: Stream())
    reader._consume()                                             # noqa: SLF001 — one pass, no thread
    assert [c.id for c in log.recent()] == ["r/900/mark_rounding"]
    assert log.recent()[0].label == "Mark 3"


def test_a_producer_that_is_not_there_costs_the_list_and_not_the_button():
    """The take is a local file operation and a loopback websocket. The office link being down must
    never be able to stop an operator putting a replay on air."""
    log = CueLog()
    logged: list[str] = []

    def dead(_url):
        raise OSError("connection refused")

    reader = CueReader("http://producer", log, open_url=dead, on_log=logged.append, retry_s=0.01)
    reader.start()
    try:
        assert any("no cues from the producer" in m for m in _wait_for(logged))
    finally:
        reader.stop()
    assert log.recent() == []


def _wait_for(rows: list, timeout_s: float = 5.0) -> list:
    end = threading.Event()
    for _ in range(int(timeout_s * 100)):
        if rows:
            return rows
        end.wait(0.01)
    return rows


def test_the_cue_the_producer_publishes_is_the_shape_the_reader_expects():
    """The two ends are in different processes on different machines; this is the only place the
    shape is written down twice."""
    published = json.dumps({"id": "inst/905/mark_rounding", "kind": "mark_rounding",
                            "label": "Ireland 6 is round first", "utcMs": T0, "race": "Race 1009"})
    log = CueLog()
    CueReader("http://p", log)._file(published)                   # noqa: SLF001
    cue = log.recent()[0]
    assert (cue.kind, cue.race) == ("mark_rounding", "Race 1009")
    assert cue.utcMs == T0