xiaoxia-saas/tests/unit/test_thumbnail_generator.py

"""缩略图生成器单元测试 - 纯逻辑函数."""

from __future__ import annotations

import pytest
from video_processing.thumbnail_generator import _format_seek_time


class TestFormatSeekTime:
    """_format_seek_time 时间格式化测试."""

    def test_zero_seconds(self):
        """0秒."""
        result = _format_seek_time(0)
        assert result == "00:00:00.00"

    def test_less_than_one_second(self):
        """小于1秒."""
        result = _format_seek_time(0.5)
        assert result == "00:00:00.50"

    def test_few_seconds(self):
        """几秒."""
        result = _format_seek_time(5.5)
        assert result == "00:00:05.50"

    def test_one_minute(self):
        """1分钟."""
        result = _format_seek_time(60.0)
        assert result == "00:01:00.00"

    def test_minutes_and_seconds(self):
        """分+秒."""
        result = _format_seek_time(125.5)
        assert result == "00:02:05.50"

    def test_one_hour(self):
        """1小时."""
        result = _format_seek_time(3600.0)
        assert result == "01:00:00.00"

    def test_hours_minutes_seconds(self):
        """时+分+秒."""
        result = _format_seek_time(3725.25)
        assert result == "01:02:05.25"

    def test_long_duration(self):
        """长视频（2小时以上）."""
        result = _format_seek_time(7384.12)
        assert result == "02:03:04.12"

    def test_precision_two_decimal(self):
        """两位小数精度."""
        result = _format_seek_time(3.14159)
        assert result == "00:00:03.14"

    def test_always_two_digit_hours(self):
        """小时始终两位数字."""
        result = _format_seek_time(3600 * 9)
        assert result.startswith("09:")

    def test_always_two_digit_minutes(self):
        """分钟始终两位数字."""
        result = _format_seek_time(300)  # 5分钟
        parts = result.split(":")
        assert parts[1] == "05"

    def test_float_input(self):
        """浮点数输入."""
        result = _format_seek_time(10.0)
        assert isinstance(result, str)
        assert result == "00:00:10.00"

    def test_int_input(self):
        """整数输入."""
        result = _format_seek_time(30)
        assert result == "00:00:30.00"

    def test_format_structure(self):
        """格式结构正确：HH:MM:SS.xx."""
        result = _format_seek_time(3661.5)
        # 格式: HH:MM:SS.xx
        parts = result.split(":")
        assert len(parts) == 3
        assert "." in parts[2]
        sec_parts = parts[2].split(".")
        assert len(sec_parts) == 2
        assert len(sec_parts[1]) == 2  # 两位小数


# ── MediaKit 智能抽帧集成测试 ──────────────────────────────────────────


class TestExtractFramesViaMediakit:
    """_extract_frames_via_mediakit 函数测试."""

    def test_mediakit_not_configured_returns_none(self, tmp_path, monkeypatch):
        """MediaKit 未配置时返回 None."""
        from video_processing.thumbnail_generator import _extract_frames_via_mediakit

        # Mock get_mediakit_client 返回不可用客户端
        class FakeClient:
            is_available = False

        monkeypatch.setattr(
            "packages.shared.mediakit_client.get_mediakit_client",
            lambda: FakeClient(),
        )

        video_file = tmp_path / "test.mp4"
        video_file.write_bytes(b"fake video")

        result = _extract_frames_via_mediakit(str(video_file), "plan1", 3)
        assert result is None

    def test_mediakit_success_returns_frames(self, tmp_path, monkeypatch):
        """MediaKit 成功时返回帧列表."""
        from video_processing.thumbnail_generator import _extract_frames_via_mediakit

        class FakeClient:
            is_available = True

            def extract_frames(self, video_url, strategy, max_frames):
                return [
                    {"image_url": "https://example.com/frame1.jpg", "timestamp": 1.5},
                    {"image_url": "https://example.com/frame2.jpg", "timestamp": 3.2},
                ]

        monkeypatch.setattr(
            "packages.shared.mediakit_client.get_mediakit_client",
            lambda: FakeClient(),
        )
        # Mock upload_to_oss
        monkeypatch.setattr(
            "video_processing.oss_helpers.upload_to_oss",
            lambda path, key: f"https://oss.example.com/{key}",
        )
        # Mock httpx.get for downloading frame
        import httpx

        class FakeResponse:
            status_code = 200
            content = b"fake image data"

            def raise_for_status(self):
                pass

        monkeypatch.setattr("httpx.get", lambda url, **kw: FakeResponse())

        video_file = tmp_path / "test.mp4"
        video_file.write_bytes(b"fake video")

        result = _extract_frames_via_mediakit(str(video_file), "plan1", 2)
        assert result is not None
        assert len(result) == 2
        assert result[0]["timestamp"] == 1.5
        assert result[1]["timestamp"] == 3.2

    def test_mediakit_failure_returns_none(self, tmp_path, monkeypatch):
        """MediaKit 调用失败时返回 None."""
        from video_processing.thumbnail_generator import _extract_frames_via_mediakit

        class FakeClient:
            is_available = True

            def extract_frames(self, video_url, strategy, max_frames):
                return None

        monkeypatch.setattr(
            "packages.shared.mediakit_client.get_mediakit_client",
            lambda: FakeClient(),
        )
        monkeypatch.setattr(
            "video_processing.oss_helpers.upload_to_oss",
            lambda path, key: f"https://oss.example.com/{key}",
        )

        video_file = tmp_path / "test.mp4"
        video_file.write_bytes(b"fake video")

        result = _extract_frames_via_mediakit(str(video_file), "plan1", 3)
        assert result is None


class TestExtractAndUploadCoverFramesFallback:
    """extract_and_upload_cover_frames 降级逻辑测试."""

    def test_fallback_to_ffmpeg_when_mediakit_fails(self, tmp_path, monkeypatch):
        """MediaKit 失败时降级到 ffmpeg 抽帧."""
        from video_processing.thumbnail_generator import extract_and_upload_cover_frames

        # Mock MediaKit 返回 None（未配置或失败）
        class FakeClient:
            is_available = False

        monkeypatch.setattr(
            "packages.shared.mediakit_client.get_mediakit_client",
            lambda: FakeClient(),
        )

        # Mock ffmpeg 抽帧
        monkeypatch.setattr(
            "video_processing.thumbnail_generator.extract_first_frame",
            lambda video_path, output_path, **kw: output_path,
        )
        # Mock upload
        monkeypatch.setattr(
            "video_processing.oss_helpers.upload_to_oss",
            lambda path, key: f"https://oss.example.com/{key}",
        )
        # Mock probe_duration
        monkeypatch.setattr(
            "video_processing.ffmpeg_utils.probe_duration",
            lambda path: 60.0,
        )

        video_file = tmp_path / "test.mp4"
        video_file.write_bytes(b"fake video")

        result = extract_and_upload_cover_frames(str(video_file), "plan1", num_frames=2)
        assert len(result) == 2
        assert all("url" in item for item in result)
        assert all("position" in item for item in result)