"""
The GPU re-encoding plan, and the one thing that must stay true across both
encoders.

`hwaccel.py` exists so that a low-power node can re-encode a browser-hostile
video at all — see its module docstring. What is tested here is the part that
holds whether or not the machine running the tests has a GPU: the shape of the
plan, what a demotion does, and the agreement between the codec string the node
announces and the arguments each encoder is actually given.

The end-to-end fallback — a hardware mode that fails, the stream playing anyway
— is in `test_stream_video_transcode.py`, because it needs a real ffmpeg and a
real file.
"""

import re
import sys

import pytest
from meshbay_node import hwaccel
from node_source import webrtc_source

from conftest import needs_subprocess

VAAPI = next(c for c in hwaccel.CANDIDATES if c.name == "vaapi")
FAKE = hwaccel.Encoder(candidate=VAAPI, variant="standard", extra=(),
                       device="/dev/dri/renderD128")


@pytest.fixture(autouse=True)
def _clean():
    hwaccel._reset_for_tests()
    yield
    hwaccel._reset_for_tests()
    hwaccel.set_enabled(True)


async def test_no_working_encoder_means_the_software_path_and_nothing_else(monkeypatch):
    """
    The guarantee this module owes every node that has no GPU, no driver, or a
    driver that fails its test encode: one mode, the one that was always there.
    A second mode would mean a second ffmpeg spawn and a second wait on a
    machine that can never benefit from either.
    """
    monkeypatch.setattr(hwaccel, "encoder", _fixed(None))

    assert await hwaccel.modes_for("hevc") == [hwaccel.SW]


async def test_the_plan_always_ends_in_software(monkeypatch):
    monkeypatch.setattr(hwaccel, "encoder", _fixed(FAKE))

    assert await hwaccel.modes_for("hevc") == [hwaccel.HW, hwaccel.HWENC, hwaccel.SW]


async def test_a_demotion_is_remembered_for_that_source_codec_only(monkeypatch):
    """
    iHD decodes HEVC and has no MPEG-4 Part 2 decoder at all, so "the GPU
    cannot do this one" is a fact about the pair, not about the machine.
    Demoting the machine would cost every HEVC film the acceleration that one
    Xvid file proved was unavailable to *it*.
    """
    monkeypatch.setattr(hwaccel, "encoder", _fixed(FAKE))
    hwaccel.demote("mpeg4", hwaccel.HW, "no decoder")

    assert await hwaccel.modes_for("mpeg4") == [hwaccel.HWENC, hwaccel.SW]
    assert await hwaccel.modes_for("hevc") == [hwaccel.HW, hwaccel.HWENC, hwaccel.SW]


async def test_the_operator_switch_turns_off_the_probe_itself(monkeypatch):
    """`hardware_video_encode = false` must cost no subprocess, not merely go
    unused: the probe is a test encode, and a node whose operator said no
    should never spawn it."""
    spawned = []
    monkeypatch.setattr(hwaccel, "_test_encode",
                        lambda *a, **k: spawned.append(a) or None)
    hwaccel.set_enabled(False)

    assert await hwaccel.encoder() is None
    assert await hwaccel.modes_for("hevc") == [hwaccel.SW]
    assert spawned == []


def test_every_encoder_produces_what_the_node_announces():
    """
    `stream_init` carries `avc1.640029` and the client puts it through
    MediaSource.isTypeSupported before it trusts a byte of the stream. The
    string is not a label chosen next to libx264 — it is a claim about the
    bytes, and it has to stay true for whichever encoder produced them.

    So the claim is read out of the streaming code and decoded (profile_idc
    0x64 = 100 = High, level_idc 0x29 = 41 = 4.1) rather than restated here,
    and every mode is checked against it. Changing the arguments of one encoder
    without the other, or changing either without the announced string, fails.
    """
    source = webrtc_source()
    announced = set(re.findall(r"avc1\.([0-9a-f]{6})", source))
    assert announced == {"640029"}, (
        "the streaming path announces a codec string this test does not know "
        f"how to check: {announced}")

    (code,) = announced
    profile_idc, constraints, level_idc = (int(code[i:i + 2], 16) for i in (0, 2, 4))
    assert (profile_idc, constraints, level_idc) == (100, 0, 41), (
        "the announced string no longer says High/4.1, so the arguments below "
        "are no longer what it claims")

    for mode in (hwaccel.SW, hwaccel.HW, hwaccel.HWENC):
        args = hwaccel.codec_args(mode, FAKE)
        assert args[args.index("-profile:v") + 1] == "high", \
            f"{mode} does not encode the High profile the node announces"
        assert args[args.index("-level") + 1] == "4.1", \
            f"{mode} does not encode at the level the node announces"


def test_the_software_mode_never_touches_the_gpu():
    """The fallback has to be a real fallback: if `sw` carried a VA-API
    argument, the mode that exists for machines where the GPU failed would
    fail with it."""
    for args in (hwaccel.input_args(hwaccel.SW, FAKE),
                 hwaccel.codec_args(hwaccel.SW, FAKE)):
        joined = " ".join(args)
        assert "vaapi" not in joined and "hwupload" not in joined, joined
    assert hwaccel.input_args(hwaccel.HW, None) == []
    assert "libx264" in hwaccel.codec_args(hwaccel.HW, None)


def _fixed(value):
    async def _f():
        return value
    return _f


@needs_subprocess
async def test_the_probe_accepts_only_what_the_node_announces():
    """The acceptance test is neither always true nor always false.

    Both halves are run with libx264, which is on any machine that can run
    these tests, so what is being checked is the judgement itself rather than
    anybody's GPU: the same encoder is accepted when it writes High at level
    4.1 and rejected when it writes level 3.1. Without this, an encoder whose
    arguments were spelled in a way ffmpeg accepts but misreads would pass the
    probe and then announce a level it had not produced.
    """
    def _software(label: str, extra: tuple[str, ...]) -> hwaccel.Candidate:
        return hwaccel.Candidate(
            name=label, encoder="libx264", platforms=(sys.platform,),
            hwaccel="none", hw_filter="null", sw_filter="format=yuv420p",
            variants=((label, extra),))

    right = _software("as announced", ("-preset", "veryfast", "-crf", "30"))
    assert await hwaccel._test_encode(right, None, right.variants[0][1]) is None

    wrong = _software("a level lower",
                      ("-preset", "veryfast", "-crf", "30", "-level", "3.1"))
    why = await hwaccel._test_encode(wrong, None, wrong.variants[0][1])
    assert why and "31" in why, (
        f"an encoder that wrote another level must be rejected, and say so: {why!r}")


@needs_subprocess
async def test_a_missing_encoder_is_rejected_rather_than_raised():
    """ffmpeg exits non-zero for an encoder it does not have; nothing here may
    turn that into an exception, because it happens on every machine that lacks
    one of the three candidates — which is most of them."""
    absent = hwaccel.Candidate(
        name="not a thing", encoder="h264_definitely_not", platforms=(sys.platform,),
        hwaccel="none", hw_filter="null", sw_filter="null", variants=(("x", ()),))

    assert await hwaccel._test_encode(absent, None, ()) is not None