xiaoxia-saas/tests/unit/test_thumbnail_generator.py

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

from __future__ import annotations

from pathlib import Path

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 抽帧直接返回 dummy 帧，避免真调用 ffmpeg
        def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw):
            results = []
            for i, st in enumerate(seek_points):
                fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg"
                fp.write_bytes(b"\xff\xd8\xff\xe0")  # 最小 jpeg 头
                results.append((st, str(fp)))
            return results

        monkeypatch.setattr(
            "video_processing.thumbnail_generator._extract_frames_single_pass",
            _fake_single_pass,
        )
        # blackdetect 直接返回空
        monkeypatch.setattr(
            "video_processing.thumbnail_generator._detect_black_intervals",
            lambda *a, **kw: [],
        )
        # 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)


class TestSeekPointBlackAvoidance:
    """_adjust_seek_points_avoid_black 纯逻辑测试."""

    def test_no_black_intervals_returns_unchanged(self):
        from video_processing.thumbnail_generator import _adjust_seek_points_avoid_black

        pts = [2.0, 5.0, 8.0]
        out = _adjust_seek_points_avoid_black(pts, [], duration=10.0)
        assert out == [2.0, 5.0, 8.0]

    def test_point_in_black_shifts_forward(self):
        from video_processing.thumbnail_generator import _adjust_seek_points_avoid_black

        # 黑屏 [4, 6]，点在 5.0，向前偏移到 4-0.25=3.75
        pts = [5.0]
        out = _adjust_seek_points_avoid_black(pts, [(4.0, 6.0)], duration=10.0)
        assert out[0] == pytest.approx(3.75, abs=0.01)

    def test_point_at_start_shifts_backward(self):
        from video_processing.thumbnail_generator import _adjust_seek_points_avoid_black

        # 黑屏 [0, 3]，点在 1.0，向前偏移 -0.25 会 <0 → 向后偏移到 3+0.25=3.25
        pts = [1.0]
        out = _adjust_seek_points_avoid_black(pts, [(0.0, 3.0)], duration=10.0)
        assert out[0] == pytest.approx(3.25, abs=0.01)

    def test_all_black_keeps_point(self):
        from video_processing.thumbnail_generator import _adjust_seek_points_avoid_black

        # 全黑，偏移都无效，保留原点
        pts = [5.0]
        out = _adjust_seek_points_avoid_black(pts, [(0.0, 10.0)], duration=10.0)
        assert out[0] == pytest.approx(5.0, abs=0.01)

    def test_multiple_points_decouple(self):
        from video_processing.thumbnail_generator import _adjust_seek_points_avoid_black

        pts = [2.0, 5.0, 8.0]
        black = [(4.5, 5.5)]  # 只有中点在黑屏
        out = _adjust_seek_points_avoid_black(pts, black, duration=10.0)
        assert out[0] == 2.0
        assert out[2] == 8.0
        # 中点必须不在黑屏内
        assert not (4.5 <= out[1] <= 5.5)


class TestScorerRunsInsideTempDir:
    """P1 修复：scorer 必须在 TemporaryDirectory 块内调用（帧文件还在时）。"""

    def _setup_mocks(self, monkeypatch, tmp_path, *, scorer_should_read=True):
        from video_processing.thumbnail_generator import extract_and_upload_cover_frames

        class FakeClient:
            is_available = False

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

        self._frames_on_disk_when_called = []

        def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw):
            results = []
            for i, st in enumerate(seek_points):
                fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg"
                fp.write_bytes(b"\xff\xd8\xff\xe0" + b"X" * 200)
                results.append((st, str(fp)))
            return results

        monkeypatch.setattr(
            "video_processing.thumbnail_generator._extract_frames_single_pass",
            _fake_single_pass,
        )
        monkeypatch.setattr(
            "video_processing.thumbnail_generator._detect_black_intervals",
            lambda *a, **kw: [],
        )
        monkeypatch.setattr(
            "video_processing.ffmpeg_utils.probe_duration",
            lambda path: 60.0,
        )
        monkeypatch.setattr(
            "video_processing.oss_helpers.upload_to_oss",
            lambda path, key: f"https://oss.example.com/{key}",
        )
        # 标题叠加 no-op
        monkeypatch.setattr(
            "video_processing.thumbnail_generator.apply_title_overlay",
            lambda *a, **kw: None,
        )

        # 记录 scorer 被调用时各 image_path 是否存在
        def _fake_scorer(candidates):
            for c in candidates:
                self._frames_on_disk_when_called.append(Path(c["image_path"]).exists())
            # 给个假评分：倒序排，验证顺序被应用
            scored = list(candidates)
            for i, c in enumerate(scored):
                c["score"] = float(len(scored) - i)
            scored.sort(key=lambda c: c["score"], reverse=True)
            return scored

        monkeypatch.setattr(
            "packages.shared.cover_frame_scorer.score_frames",
            _fake_scorer,
        )
        return extract_and_upload_cover_frames

    def test_scorer_reads_files_while_they_exist(self, tmp_path, monkeypatch):
        """核心 P1：评分时帧文件必须还在磁盘上（在 TemporaryDirectory 内调用）。"""
        extract = self._setup_mocks(monkeypatch, tmp_path)
        video_file = tmp_path / "t.mp4"
        video_file.write_bytes(b"fake")
        result = extract(str(video_file), "plan1", num_frames=3)
        # scorer 看到的 3 个文件都必须存在
        assert len(self._frames_on_disk_when_called) == 3
        assert all(self._frames_on_disk_when_called), f"scorer 调用时有文件已被删除: {self._frames_on_disk_when_called}"
        # 结果按评分降序排列（is_best 在第一个）
        assert len(result) == 3
        assert result[0].get("is_best") is True
        # 结果中不应该再暴露 image_path
        assert all("image_path" not in c for c in result)

    def test_scorer_failure_falls_back_gracefully(self, tmp_path, monkeypatch):
        """评分抛异常时不应中断上传，仍返回所有候选帧。"""
        from video_processing.thumbnail_generator import extract_and_upload_cover_frames

        class FakeClient:
            is_available = False

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

        def _fake_single_pass(video_path, seek_points, out_dir, prefix="frame", **kw):
            results = []
            for i, st in enumerate(seek_points):
                fp = Path(out_dir) / f"{prefix}_{i + 1:02d}.jpg"
                fp.write_bytes(b"\xff\xd8\xff\xe0" + b"X" * 100)
                results.append((st, str(fp)))
            return results

        monkeypatch.setattr("video_processing.thumbnail_generator._extract_frames_single_pass", _fake_single_pass)
        monkeypatch.setattr("video_processing.thumbnail_generator._detect_black_intervals", lambda *a, **kw: [])
        monkeypatch.setattr("video_processing.ffmpeg_utils.probe_duration", lambda p: 60.0)
        monkeypatch.setattr(
            "video_processing.oss_helpers.upload_to_oss",
            lambda path, key: f"https://oss/{key}",
        )
        monkeypatch.setattr("video_processing.thumbnail_generator.apply_title_overlay", lambda *a, **kw: None)

        def _boom(candidates):
            raise RuntimeError("cv2 crashed")

        monkeypatch.setattr("packages.shared.cover_frame_scorer.score_frames", _boom)

        video_file = tmp_path / "t.mp4"
        video_file.write_bytes(b"fake")
        # 不应抛出
        result = extract_and_upload_cover_frames(str(video_file), "plan1", num_frames=3)
        assert len(result) == 3
        assert all("url" in c for c in result)

    def test_best_frame_is_first_after_scoring(self, tmp_path, monkeypatch):
        """评分后 best 帧（score 最高）在 candidates[0]，is_best=True。"""
        extract = self._setup_mocks(monkeypatch, tmp_path)
        video_file = tmp_path / "t.mp4"
        video_file.write_bytes(b"fake")
        result = extract(str(video_file), "plan1", num_frames=5)
        assert result[0]["is_best"] is True
        scores = [c.get("score", 0.0) for c in result]
        assert scores == sorted(scores, reverse=True)