mitus/tests/test_ffmpeg.py

"""Tests for cht.stream.ffmpeg — pipeline construction and execution."""

import os
import signal
import subprocess
from pathlib import Path
from unittest.mock import patch, MagicMock

import pytest
import ffmpeg as ffmpeg_lib

from cht.stream import ffmpeg as ff


class TestReceiveAndSegment:
    def test_compiles_to_valid_cmd(self, tmp_path):
        node = ff.receive_and_segment(
            "tcp://0.0.0.0:4444?listen", tmp_path, segment_duration=30
        )
        cmd = node.compile()
        cmd_str = " ".join(str(c) for c in cmd)

        assert cmd[0] == "ffmpeg"
        assert "tcp://0.0.0.0:4444?listen" in cmd_str
        assert "segment" in cmd_str
        assert str(tmp_path / "segment_%04d.ts") in cmd_str

    def test_input_has_low_latency_flags(self, tmp_path):
        node = ff.receive_and_segment("tcp://0.0.0.0:4444?listen", tmp_path)
        cmd_str = " ".join(str(c) for c in node.compile())
        assert "nobuffer" in cmd_str
        assert "low_delay" in cmd_str

    def test_copy_codec(self, tmp_path):
        node = ff.receive_and_segment("tcp://0.0.0.0:4444?listen", tmp_path)
        assert "copy" in node.compile()

    def test_has_global_args(self, tmp_path):
        node = ff.receive_and_segment("tcp://0.0.0.0:4444?listen", tmp_path)
        cmd = node.compile()
        assert "-hide_banner" in cmd


class TestReceiveAndSegmentWithMonitor:
    def test_creates_fifo(self, tmp_path):
        fifo = tmp_path / "monitor.pipe"
        ff.receive_and_segment_with_monitor(
            "tcp://0.0.0.0:4444?listen", tmp_path, fifo
        )
        assert fifo.exists()

    def test_does_not_recreate_existing_fifo(self, tmp_path):
        fifo = tmp_path / "monitor.pipe"
        os.mkfifo(str(fifo))
        inode_before = fifo.stat().st_ino
        ff.receive_and_segment_with_monitor(
            "tcp://0.0.0.0:4444?listen", tmp_path, fifo
        )
        assert fifo.stat().st_ino == inode_before

    def test_has_two_outputs(self, tmp_path):
        fifo = tmp_path / "monitor.pipe"
        node = ff.receive_and_segment_with_monitor(
            "tcp://0.0.0.0:4444?listen", tmp_path, fifo
        )
        cmd_str = " ".join(str(c) for c in node.compile())
        assert "segment_%04d.ts" in cmd_str
        assert "monitor.pipe" in cmd_str

    def test_both_outputs_use_copy(self, tmp_path):
        fifo = tmp_path / "monitor.pipe"
        node = ff.receive_and_segment_with_monitor(
            "tcp://0.0.0.0:4444?listen", tmp_path, fifo
        )
        cmd = node.compile()
        assert cmd.count("copy") >= 2

    def test_has_global_args(self, tmp_path):
        fifo = tmp_path / "monitor.pipe"
        node = ff.receive_and_segment_with_monitor(
            "tcp://0.0.0.0:4444?listen", tmp_path, fifo
        )
        assert "-hide_banner" in node.compile()


class TestExtractSceneFrames:
    def test_compiles_select_filter(self, tmp_path):
        stream = ffmpeg_lib.input(str(tmp_path / "test.ts"))
        stream = stream.filter("select", "gt(scene,0.3)+gte(t-prev_selected_t,30)")
        stream = stream.filter("showinfo")
        output = ffmpeg_lib.output(
            stream,
            str(tmp_path / "F%04d.jpg"),
            vsync="vfr",
            **{"q:v": "2"},
            start_number=1,
        )
        cmd_str = " ".join(str(c) for c in output.compile())
        assert "select" in cmd_str
        assert "scene" in cmd_str
        assert "showinfo" in cmd_str
        assert "vfr" in cmd_str

    def test_returns_decoded_strings(self, tmp_path):
        mock_proc = MagicMock()
        mock_proc.communicate.return_value = (b"out", b"err")
        mock_proc.poll.return_value = 0
        with patch("ffmpeg._run.run_async", return_value=mock_proc):
            stdout, stderr = ff.extract_scene_frames(
                tmp_path / "test.ts", tmp_path,
            )
        assert stdout == "out"
        assert stderr == "err"


class TestExtractAudioPcm:
    def test_compiles_audio_extraction(self, tmp_path):
        node = ff.extract_audio_pcm(tmp_path / "test.ts")
        cmd_str = " ".join(str(c) for c in node.compile())
        assert "pcm_s16le" in cmd_str
        assert "16000" in cmd_str
        assert "pipe:" in cmd_str
        assert "wav" in cmd_str

    def test_has_global_args(self, tmp_path):
        node = ff.extract_audio_pcm(tmp_path / "test.ts")
        assert "-hide_banner" in node.compile()


class TestRunAsync:
    @patch("ffmpeg.run_async")
    def test_delegates_to_ffmpeg_python(self, mock_run_async, tmp_path):
        node = ffmpeg_lib.input("test").output("out")
        ff.run_async(node)
        # ffmpeg-python's run_async is called on the node
        # Our function calls node.run_async() which is the bound method

    @patch("ffmpeg.run_async")
    def test_passes_pipe_flags(self, mock_run_async, tmp_path):
        node = ffmpeg_lib.input("test").output("out")
        ff.run_async(node, pipe_stdout=True, pipe_stderr=True)


class TestStopProc:
    def test_sends_sigint_then_waits(self):
        proc = MagicMock()
        proc.poll.return_value = None
        ff.stop_proc(proc, timeout=3)
        proc.send_signal.assert_called_once_with(signal.SIGINT)
        proc.wait.assert_called_once_with(timeout=3)

    def test_kills_on_timeout(self):
        proc = MagicMock()
        proc.poll.return_value = None
        proc.wait.side_effect = subprocess.TimeoutExpired("ffmpeg", 3)
        ff.stop_proc(proc, timeout=3)
        proc.kill.assert_called_once()

    def test_noop_if_already_exited(self):
        proc = MagicMock()
        proc.poll.return_value = 0
        ff.stop_proc(proc)
        proc.send_signal.assert_not_called()

    def test_noop_if_none(self):
        ff.stop_proc(None)