voice-transcriptor/tests/test_media_probe.py

import json
import subprocess
from pathlib import Path

import pytest

from voice_transcriptor.services.media_probe import (
    InvalidMediaPathError,
    MediaProbeError,
    MediaProbeService,
    ProbeExecutionError,
    ProbeOutputError,
    ProbeTimeoutError,
    detect_tools,
    parse_probe_output,
    validate_media_path,
)


def test_detects_ffmpeg_and_ffprobe_executables(monkeypatch: pytest.MonkeyPatch) -> None:
    discovered = {"ffmpeg": r"C:\\tools\\ffmpeg.exe", "ffprobe": r"C:\\tools\\ffprobe.exe"}
    monkeypatch.setattr("voice_transcriptor.services.media_probe.shutil.which", discovered.get)

    status = detect_tools()

    assert status.ffmpeg_path == Path(r"C:\tools\ffmpeg.exe")
    assert status.ffprobe_path == Path(r"C:\tools\ffprobe.exe")


def test_rejects_directory_path(tmp_path: Path) -> None:
    with pytest.raises(InvalidMediaPathError, match="not a file"):
        validate_media_path(tmp_path)


def test_parses_audio_only_media(tmp_path: Path) -> None:
    media = tmp_path / "memo.m4a"
    payload = json.dumps({
        "format": {"duration": "4.5"},
        "streams": [{"codec_type": "audio", "codec_name": "aac"}],
    })

    info = parse_probe_output(media, 3, payload)

    assert info.duration_seconds == 4.5
    assert info.audio_codec == "aac"
    assert info.duration_text == "4.5"
    assert info.has_audio is True
    assert info.has_video is False


def test_parses_video_with_audio(tmp_path: Path) -> None:
    media = tmp_path / "meeting.mov"
    media.write_bytes(b"x" * 12)
    payload = json.dumps({
        "format": {"duration": "3661.25"},
        "streams": [{"codec_type": "video", "codec_name": "h264"}, {"codec_type": "audio", "codec_name": "aac"}],
    })
    info = parse_probe_output(media, 12, payload)
    assert info.duration_seconds == 3661.25
    assert info.audio_codec == "aac"
    assert info.has_video is True


def test_missing_metadata_is_unknown(tmp_path: Path) -> None:
    media = tmp_path / "memo.m4a"
    info = parse_probe_output(media, 0, '{"format": {}, "streams": []}')
    assert info.duration_seconds is None
    assert info.audio_codec is None


def test_rejects_malformed_probe_json(tmp_path: Path) -> None:
    with pytest.raises(MediaProbeError, match="invalid data"):
        parse_probe_output(tmp_path / "bad.mov", 0, "not json")


@pytest.mark.parametrize("payload", ["[]", '{"format": []}', '{"streams": [null]}'])
def test_rejects_structurally_invalid_probe_json(tmp_path: Path, payload: str) -> None:
    with pytest.raises(ProbeOutputError, match="invalid data"):
        parse_probe_output(tmp_path / "bad.mov", 0, payload)


def test_rejects_missing_file(tmp_path: Path) -> None:
    with pytest.raises(MediaProbeError, match="does not exist"):
        validate_media_path(tmp_path / "missing.mov")


def test_maps_nonzero_ffprobe_result(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    media = tmp_path / "bad.mov"
    media.write_bytes(b"bad")
    monkeypatch.setattr(subprocess, "run", lambda *a, **k: subprocess.CompletedProcess(a, 1, "", "invalid input"))
    with pytest.raises(MediaProbeError, match="could not inspect"):
        MediaProbeService(Path("ffprobe")).probe(media)


def test_maps_ffprobe_timeout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    media = tmp_path / "slow.mov"
    media.write_bytes(b"slow")
    def timeout(*args: object, **kwargs: object) -> None:
        raise subprocess.TimeoutExpired("ffprobe", 30)
    monkeypatch.setattr(subprocess, "run", timeout)
    with pytest.raises(MediaProbeError, match="timed out"):
        MediaProbeService(Path("ffprobe")).probe(media)


def test_maps_probe_failures_to_typed_errors(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    missing = tmp_path / "missing.mov"
    with pytest.raises(InvalidMediaPathError):
        validate_media_path(missing)

    with pytest.raises(ProbeOutputError):
        parse_probe_output(tmp_path / "bad.mov", 0, "not json")

    media = tmp_path / "bad.mov"
    media.write_bytes(b"bad")
    monkeypatch.setattr(subprocess, "run", lambda *a, **k: subprocess.CompletedProcess(a, 1, "", "invalid input"))
    with pytest.raises(ProbeExecutionError):
        MediaProbeService(Path("ffprobe")).probe(media)

    def timeout(*args: object, **kwargs: object) -> None:
        raise subprocess.TimeoutExpired("ffprobe", 30)

    monkeypatch.setattr(subprocess, "run", timeout)
    with pytest.raises(ProbeTimeoutError):
        MediaProbeService(Path("ffprobe")).probe(media)