"""The break roll — approved phone clips made playable on the machine that runs OBS.

No ffmpeg. The commands are built as data by pure functions and run by a thin wrapper, so what is
pinned here is the two things that decide whether a break is right: WHAT ffmpeg is asked to do, and
what the module does when something is missing, revoked or unreachable.

The load-bearing one is the last of those. A pass that could not ask must leave the folder exactly as
it was — "I could not read approved.json" is not "there is nothing approved", and the difference is a
break that plays and a break that is suddenly empty on air.
"""
from __future__ import annotations

import json
import sys
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 break_roll as BR  # noqa: E402


class FakeFfmpeg:
    """Records every command and produces the output file the real one would."""

    def __init__(self, *, encoders: str = "", audio: bool = True, fail: set[str] | None = None):
        self.encoders, self.audio, self.fail = encoders, audio, set(fail or ())
        self.cmds: list[list[str]] = []

    def __call__(self, cmd, timeout=None):                       # noqa: ARG002
        self.cmds.append(list(cmd))
        if cmd[0] == "ffprobe":
            return _R(0, "0\n" if self.audio else "")
        if "-encoders" in cmd:
            return _R(0, self.encoders)
        out = Path(cmd[-1])
        if any(f in str(out) for f in self.fail):
            return _R(1, "", "ffmpeg: moov atom not found")
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_bytes(b"\x00" * 64)
        return _R(0)

    def of(self, marker: str) -> list[list[str]]:
        return [c for c in self.cmds if any(marker in str(part) for part in c)]


class _R:
    def __init__(self, returncode: int, stdout: str = "", stderr: str = ""):
        self.returncode, self.stdout, self.stderr = returncode, stdout, stderr


@pytest.fixture
def have_ffmpeg(monkeypatch):
    monkeypatch.setattr(BR, "have_tool", lambda _n: True)


def _seed(directory: Path, *names: str) -> None:
    src = directory / "source"
    src.mkdir(parents=True, exist_ok=True)
    for n in names:
        (src / n).write_bytes(b"phone clip")


# --- what ffmpeg is asked to do ------------------------------------------------------------------

def test_one_filter_graph_covers_both_orientations():
    """Half the material is portrait — a phone held the way people hold phones. The sides get a
    blurred cover-scaled copy of the same clip, and for a landscape clip the foreground covers that
    backdrop entirely, so the identical command is correct for both. A branch here would have to read
    the rotation matrix rather than the encoded dimensions, since a portrait clip has LANDSCAPE
    dimensions and expresses the quarter-turn separately."""
    cmd = BR.normalize_cmd(Path("in.mov"), Path("out.mp4"))
    graph = cmd[cmd.index("-filter_complex") + 1]
    assert "force_original_aspect_ratio=increase" in graph      # the backdrop, cover-scaled
    assert "force_original_aspect_ratio=decrease" in graph      # the picture, whole
    assert "overlay=(W-w)/2:(H-h)/2" in graph                   # centred over it
    assert "fps=30" in graph and "format=yuv420p" in graph
    assert "1920:1080" in graph and cmd.count("-filter_complex") == 1
    # `setsar=1` after the overlay, not before it: set on the foreground, the overlay inherits the
    # backdrop's sample aspect and the file lands at 10239:10240. Measured. The roll is joined with a
    # stream copy, so parts that disagree about SAR are parts a player may treat as a new stream.
    assert graph.index("setsar=1") > graph.index("overlay=")


def test_a_clip_with_sound_is_levelled_to_the_programme():
    cmd = BR.normalize_cmd(Path("in.mp4"), Path("out.mp4"), has_audio=True)
    audio = cmd[cmd.index("-af") + 1]
    assert BR.LOUDNORM in audio and BR.HIGHPASS in audio
    assert "-map" in cmd and "0:a:0" in cmd
    assert "anullsrc" not in " ".join(cmd)


def test_a_silent_clip_still_gets_an_audio_track():
    """A track that appears and disappears part-way through the roll is a re-encode at best and a
    mixer noticing mid-break at worst — so every output has one, and the join stays a stream copy."""
    cmd = BR.normalize_cmd(Path("in.mp4"), Path("out.mp4"), has_audio=False)
    assert "anullsrc=channel_layout=stereo:sample_rate=48000" in cmd
    assert "1:a" in cmd
    # Right here and wrong in `record_demo`: the null source is infinite, the picture is finite.
    assert "-shortest" in cmd
    assert "-af" not in cmd


def test_mute_drops_the_clips_own_sound():
    cmd = BR.normalize_cmd(Path("in.mp4"), Path("out.mp4"), has_audio=True, mute=True)
    assert "anullsrc=channel_layout=stereo:sample_rate=48000" in cmd and "-af" not in cmd


def test_a_long_upload_is_truncated():
    assert "-t" in BR.normalize_cmd(Path("i"), Path("o"), max_seconds=45)
    cmd = BR.normalize_cmd(Path("i"), Path("o"), max_seconds=45)
    assert cmd[cmd.index("-t") + 1] == "45"
    assert "-t" not in BR.normalize_cmd(Path("i"), Path("o"), max_seconds=None)


def test_the_encoder_is_the_boxs_and_the_fallback_is_the_one_record_demo_uses():
    assert BR.pick_encoder(FakeFfmpeg(encoders="V..... h264_nvenc  NVIDIA NVENC")) == "h264_nvenc"
    assert BR.pick_encoder(FakeFfmpeg(encoders="V..... libx264")) == "libx264"
    x264 = BR.normalize_cmd(Path("i"), Path("o"), encoder="libx264")
    assert "-crf" in x264 and x264[x264.index("-preset") + 1] == "medium"
    nvenc = BR.normalize_cmd(Path("i"), Path("o"), encoder="h264_nvenc")
    assert "h264_nvenc" in nvenc and "-cq" in nvenc


def test_the_join_is_a_stream_copy():
    cmd = BR.concat_cmd(Path("list.txt"), Path("roll.mp4"))
    assert "-c" in cmd and cmd[cmd.index("-c") + 1] == "copy"
    assert "+faststart" in cmd


def test_the_concat_listing_escapes_a_quote_in_a_path():
    text = BR.concat_listing([Path("/vr/break/clips/a.mp4"), Path("/vr/it's/b.mp4")])
    assert text.splitlines()[0] == "file '/vr/break/clips/a.mp4'"
    assert r"'\''" in text.splitlines()[1]            # the demuxer's own escape for a quote


def test_a_windows_path_is_written_with_forward_slashes():
    """ffmpeg accepts them on Windows, and a backslash in this file is read as an escape — so
    `D:\\vr\\break` would be a path the join cannot find, on the only platform it runs on."""
    text = BR.concat_listing([Path("D:/vr/break") / "clips" / "a.mp4"])
    assert text.strip() == "file 'D:/vr/break/clips/a.mp4'"
    assert "\\" not in text


def test_a_probe_that_cannot_answer_assumes_there_IS_audio():
    """Mapping a track that exists is harmless; adding a null source to a clip that already has one
    gives two tracks and a break whose sound depends on which the player picks."""
    class Broken:
        def __call__(self, *_a, **_kw):
            raise OSError("no ffprobe")
    assert BR.has_audio(Path("x.mp4"), Broken()) is True
    assert BR.has_audio(Path("x.mp4"), FakeFfmpeg(audio=False)) is False


# --- building the roll ---------------------------------------------------------------------------

def test_build_normalizes_and_joins_in_key_order(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "000000000000002_b.mp4", "000000000000001_a.mov")
    fake = FakeFfmpeg()
    result = BR.build(tmp_path, run=fake, log=lambda *_a: None)
    assert result.state == "ok" and result.clips == 2
    assert (tmp_path / "break_roll.mp4").exists()
    listed = (tmp_path / "clips.txt").read_text(encoding="utf-8").splitlines()
    assert [Path(line.split("'")[1]).stem for line in listed] == \
           ["000000000000001_a", "000000000000002_b"]


def test_build_is_idempotent(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "0001_a.mp4")
    BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    again = FakeFfmpeg()
    result = BR.build(tmp_path, run=again, log=lambda *_a: None)
    assert result.normalized == () and result.changed is False
    assert not again.of("-filter_complex"), "a second pass re-encoded something"


def test_one_unreadable_clip_does_not_cost_the_roll(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "0001_good.mp4", "0002_broken.mp4")
    result = BR.build(tmp_path, run=FakeFfmpeg(fail={"0002_broken"}), log=lambda *_a: None)
    assert result.failed == ("0002_broken",)
    assert result.clips == 1 and (tmp_path / "break_roll.mp4").exists()


def test_a_stray_file_is_named_and_skipped_not_fed_to_ffmpeg(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "0001_a.mp4")
    (tmp_path / "source" / "notes.txt").write_text("x", encoding="utf-8")
    (tmp_path / "source" / "0002_b.mp4.part").write_bytes(b"half")
    result = BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    assert result.skipped == ("notes.txt",)          # a half-download is a transient, not a finding
    assert result.clips == 1


def test_a_removed_source_takes_its_derivative_with_it(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "0001_a.mp4", "0002_b.mp4")
    BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    (tmp_path / "source" / "0002_b.mp4").unlink()
    result = BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    assert result.removed == ("0002_b",) and result.clips == 1


def test_an_empty_folder_removes_the_roll_rather_than_leaving_yesterdays(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "0001_a.mp4")
    BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    (tmp_path / "source" / "0001_a.mp4").unlink()
    result = BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    assert result.clips == 0
    assert not (tmp_path / "break_roll.mp4").exists(), "a roll from yesterday is not nothing"


def test_without_ffmpeg_nothing_is_touched_and_it_says_unknown(tmp_path: Path, monkeypatch):
    _seed(tmp_path, "0001_a.mp4")
    monkeypatch.setattr(BR, "have_tool", lambda _n: False)
    result = BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    assert result.state == "unknown" and "ffmpeg" in result.reason
    assert not (tmp_path / "clips").exists() or not list((tmp_path / "clips").glob("*.mp4"))


def test_a_failed_join_leaves_the_previous_roll_alone(tmp_path: Path, have_ffmpeg):
    _seed(tmp_path, "0001_a.mp4")
    BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    before = (tmp_path / "break_roll.mp4").read_bytes()
    _seed(tmp_path, "0002_b.mp4")
    result = BR.build(tmp_path, run=FakeFfmpeg(fail={"break_roll.next"}), log=lambda *_a: None)
    assert result.state == "unknown"
    assert (tmp_path / "break_roll.mp4").read_bytes() == before


# --- keeping up with what was approved -----------------------------------------------------------

def _clip(idx: int, approved_ms: int, url: str = "https://gcs/x.mp4") -> dict:
    return {"id": f"c{idx}", "approvedUtcMs": approved_ms, "url": url}


def test_the_key_sorts_by_approval_and_never_changes():
    """A new approval sorts to the end and nothing is renumbered. 'This one first' is served by
    approving it first."""
    keys = [BR.clip_key(_clip(1, 200)), BR.clip_key(_clip(2, 100)), BR.clip_key(_clip(3, 300))]
    assert sorted(keys) == [keys[1], keys[0], keys[2]]
    assert BR.clip_key(_clip(1, 200)) == keys[0]                 # stable across passes
    # Two approvals in the same millisecond still separate.
    assert BR.clip_key(_clip(1, 100)) != BR.clip_key(_clip(2, 100))


def test_reconcile_fetches_what_is_new(tmp_path: Path, monkeypatch):
    monkeypatch.setattr(BR, "urlopen_bytes", None, raising=False)
    fetched = []

    def fake_urlopen(url, timeout=None):                          # noqa: ARG001
        fetched.append(url)
        return _Body(b"clip bytes")
    monkeypatch.setattr(BR.urllib.request, "urlopen", fake_urlopen)
    n, failures = BR.reconcile(tmp_path, [_clip(1, 100), _clip(2, 200)], log=lambda *_a: None)
    assert (n, failures) == (2, [])
    assert sorted(p.stem for p in (tmp_path / "source").iterdir()) == \
           [BR.clip_key(_clip(1, 100)), BR.clip_key(_clip(2, 200))]
    assert not list((tmp_path / "source").glob("*.part")), "a half-written file must not survive"


def test_a_revoked_approval_is_withdrawn_from_the_box(tmp_path: Path, monkeypatch):
    monkeypatch.setattr(BR.urllib.request, "urlopen", lambda url, timeout=None: _Body(b"x"))  # noqa: ARG005
    BR.reconcile(tmp_path, [_clip(1, 100), _clip(2, 200)], log=lambda *_a: None)
    BR.reconcile(tmp_path, [_clip(1, 100)], log=lambda *_a: None)
    assert [p.stem for p in (tmp_path / "source").iterdir()] == [BR.clip_key(_clip(1, 100))]


def test_a_clip_that_will_not_download_is_named_not_swallowed(tmp_path: Path, monkeypatch):
    def boom(url, timeout=None):                                 # noqa: ARG001
        raise OSError("connection reset")
    monkeypatch.setattr(BR.urllib.request, "urlopen", boom)
    n, failures = BR.reconcile(tmp_path, [_clip(1, 100)], log=lambda *_a: None)
    assert (n, failures) == (0, [BR.clip_key(_clip(1, 100))])
    assert not list((tmp_path / "source").iterdir())


def test_an_unreachable_read_plane_changes_nothing_and_reports_unknown(tmp_path: Path, have_ffmpeg):
    """The one that matters. A break that exists must not vanish because 4G blipped."""
    _seed(tmp_path, "0001_a.mp4")
    BR.build(tmp_path, run=FakeFfmpeg(), log=lambda *_a: None)
    before = sorted(p.name for p in (tmp_path / "source").iterdir())

    def unreachable(*_a, **_kw):
        raise OSError("connection refused")
    result = BR.sync_once(tmp_path, event="e", read_base="https://x", token="t",
                          run=FakeFfmpeg(), fetch=unreachable, log=lambda *_a: None)
    assert result.state == "unknown" and "approved.json" in result.reason
    assert sorted(p.name for p in (tmp_path / "source").iterdir()) == before
    assert (tmp_path / "break_roll.mp4").exists()
    status = json.loads((tmp_path / "status.json").read_text(encoding="utf-8"))
    # A missing reading is stated, never zeroed — "0 approved" over a failed fetch is the same lie
    # as a green probe that never ran.
    assert status["state"] == "unknown" and status["approved"] is None


def test_a_successful_pass_records_what_it_found(tmp_path: Path, have_ffmpeg, monkeypatch):
    monkeypatch.setattr(BR.urllib.request, "urlopen", lambda url, timeout=None: _Body(b"x"))  # noqa: ARG005
    result = BR.sync_once(tmp_path, event="e", read_base="https://x", token="t",
                          run=FakeFfmpeg(), fetch=lambda *_a, **_kw: [_clip(1, 100)],
                          log=lambda *_a: None)
    assert result.state == "ok" and result.clips == 1
    status = json.loads((tmp_path / "status.json").read_text(encoding="utf-8"))
    assert status["approved"] == 1 and status["clips"] == 1 and status["lastChangeUtc"]
    assert status["checkedUtc"].endswith("+00:00")


def test_the_approved_url_is_the_events_allow_list():
    assert BR.approved_url("https://apps.viewregatta.com/", "trwc-2026") == \
           "https://apps.viewregatta.com/v3/break/trwc-2026/approved.json"


def test_where_the_roll_lives_is_a_deployment_fact(tmp_path: Path, monkeypatch):
    monkeypatch.delenv(BR.DIR_ENV, raising=False)
    assert BR.break_dir(tmp_path) == tmp_path / "_cap" / "break_roll"
    monkeypatch.setenv(BR.DIR_ENV, str(tmp_path / "from-env"))
    assert BR.break_dir(tmp_path) == tmp_path / "from-env"
    assert BR.break_dir(tmp_path, str(tmp_path / "explicit")) == tmp_path / "explicit"


class _Body:
    def __init__(self, data: bytes):
        self._data = data

    def read(self, n: int = -1) -> bytes:
        data, self._data = (self._data if n in (-1, None) else self._data[:n]), \
                           (b"" if n in (-1, None) else self._data[n:])
        return data

    def __enter__(self):
        return self

    def __exit__(self, *_a):
        return False