republisher/tests/test_pipelines.py

import sys
from pathlib import Path
from types import SimpleNamespace

import pytest

from repub import media
from repub.config import (
    FeedConfig,
    RepublisherConfig,
    build_base_settings,
    build_feed_settings,
)
from repub.pipelines import AudioPipeline, FilePipeline, VideoPipeline


def build_test_crawler(tmp_path: Path) -> SimpleNamespace:
    out_dir = (tmp_path / "mirror").resolve()
    config = RepublisherConfig(
        config_path=tmp_path / "repub.toml",
        out_dir=out_dir,
        feeds=(
            FeedConfig(
                name="NASA Breaking News",
                slug="nasa",
                url="https://www.nasa.gov/rss/dyn/breaking_news.rss",
            ),
        ),
        scrapy_settings={},
    )
    base_settings = build_base_settings(config)
    settings = build_feed_settings(base_settings, out_dir=out_dir, feed_slug="nasa")
    return SimpleNamespace(settings=settings, request_fingerprinter=object())


@pytest.mark.parametrize(
    ("pipeline_cls", "store_setting"),
    [
        (AudioPipeline, "AUDIO_STORE"),
        (VideoPipeline, "VIDEO_STORE"),
        (FilePipeline, "FILES_STORE"),
    ],
)
def test_pipeline_from_crawler_uses_configured_store(
    tmp_path: Path, pipeline_cls, store_setting: str
) -> None:
    crawler = build_test_crawler(tmp_path)

    pipeline = pipeline_cls.from_crawler(crawler)

    assert pipeline.settings is crawler.settings
    assert pipeline.store.basedir == crawler.settings[store_setting]


def test_transcode_audio_captures_ffmpeg_output(monkeypatch, tmp_path: Path) -> None:
    input_file = tmp_path / "input.mp3"
    input_file.write_bytes(b"12345")
    output_dir = tmp_path / "audio-out"
    output_dir.mkdir()
    run_calls: list[dict[str, object]] = []

    class FakeOutput:
        def __init__(self, output_path: Path):
            self.output_path = output_path

        def run(self, **kwargs):
            run_calls.append(kwargs)
            self.output_path.write_bytes(b"12")
            return b"", b""

    class FakeInput:
        def output(self, output_file: str, **params):
            del params
            return FakeOutput(Path(output_file))

    monkeypatch.setattr(media.ffmpeg, "input", lambda _: FakeInput())

    result = media.transcode_audio(
        str(input_file),
        str(output_dir),
        {"extension": "mp3", "acodec": "libmp3lame"},
    )

    assert result == str(output_dir / "converted.mp3")
    assert run_calls == [{"capture_stdout": True, "capture_stderr": True}]


def test_transcode_video_two_pass_does_not_print_ffmpeg_output(
    monkeypatch, tmp_path: Path
) -> None:
    input_file = tmp_path / "input.mp4"
    input_file.write_bytes(b"12345")
    output_dir = tmp_path / "video-out"
    output_dir.mkdir()
    run_calls: list[dict[str, object]] = []
    printed: list[tuple[tuple[object, ...], dict[str, object]]] = []

    class FakeOutput:
        def __init__(self, output_path: Path | None):
            self.output_path = output_path

        def global_args(self, *args):
            del args
            return self

        def run(self, **kwargs):
            run_calls.append(kwargs)
            if self.output_path is not None:
                self.output_path.write_bytes(b"12")
            return b"pass-out", b"pass-err"

    class FakeInput:
        video = object()
        audio = object()

        def output(self, *args, **params):
            del params
            output_path = next(
                (
                    Path(arg)
                    for arg in args
                    if isinstance(arg, str) and arg.endswith(".mp4")
                ),
                None,
            )
            return FakeOutput(output_path)

    monkeypatch.setattr(media.ffmpeg, "input", lambda _: FakeInput())
    monkeypatch.setattr(
        "builtins.print", lambda *args, **kwargs: printed.append((args, kwargs))
    )

    result = media.transcode_video(
        str(input_file),
        str(output_dir),
        {
            "extension": "mp4",
            "passes": [
                {"f": "null"},
                {"c:v": "libx264"},
            ],
        },
    )

    assert result == str(output_dir / "converted.mp4")
    assert run_calls == [
        {"capture_stdout": True, "capture_stderr": True},
        {
            "capture_stdout": True,
            "capture_stderr": True,
            "overwrite_output": True,
        },
    ]
    assert printed == []


def test_transcode_video_prints_ffmpeg_output_on_error(
    monkeypatch, tmp_path: Path
) -> None:
    input_file = tmp_path / "input.mp4"
    input_file.write_bytes(b"12345")
    output_dir = tmp_path / "video-out"
    output_dir.mkdir()
    printed: list[tuple[str, bool]] = []

    class FakeOutput:
        def run(self, **kwargs):
            del kwargs
            raise media.ffmpeg.Error("ffmpeg", b"video-stdout", b"video-stderr")

    class FakeInput:
        def output(self, *args, **params):
            del args, params
            return FakeOutput()

    def fake_print(*args, **kwargs):
        printed.append((str(args[0]), kwargs.get("file") is sys.stderr))

    monkeypatch.setattr(media.ffmpeg, "input", lambda _: FakeInput())
    monkeypatch.setattr("builtins.print", fake_print)

    with pytest.raises(RuntimeError):
        media.transcode_video(
            str(input_file),
            str(output_dir),
            {"extension": "mp4", "c:v": "libx264"},
        )

    assert ("video-stderr", True) in printed
    assert ("video-stdout", False) in printed