xiaoxia-saas/tests/unit/test_cover_frame_pre_extract.py

"""Tests for cover frame pre-extraction during rendering.

Tests:
- extract_cover_candidates: FFmpeg frame extraction at 25%/50%/75%
- extract_and_upload_cover_frames: extraction + OSS upload
- RenderAdapterResult.cover_candidates field
- generation_cover route uses pre-stored candidates
- ai_service FFmpeg fallback
"""

from __future__ import annotations

from pathlib import Path
from unittest.mock import MagicMock, Mock, call, patch

import pytest


class TestExtractCoverCandidates:
    """extract_cover_candidates 测试."""

    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
    def test_extracts_3_frames_at_correct_positions(self, mock_probe, mock_run):
        """在 25%/50%/75% 处抽取 3 帧."""
        import tempfile

        from video_processing.thumbnail_generator import extract_cover_candidates

        # Create temp files that look like they were created
        def fake_run(cmd, **kwargs):
            # Find the output path (last arg)
            output_path = cmd[-1]
            Path(output_path).write_bytes(b"\xff\xd8\xff\xe0" + b"\x00" * 100)
            return ("", "")

        mock_run.side_effect = fake_run

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
            tmp.write(b"fake video")
            video_path = tmp.name

        try:
            results = extract_cover_candidates(video_path, num_frames=3)
            assert len(results) == 3

            # Check frame times: 20*0.25=5.0, 20*0.5=10.0, 20*0.75=15.0
            assert results[0]["frame_time"] == 5.0
            assert results[1]["frame_time"] == 10.0
            assert results[2]["frame_time"] == 15.0

            # Check local paths exist
            for r in results:
                assert Path(r["local_path"]).exists()

            # Clean up
            for r in results:
                Path(r["local_path"]).unlink(missing_ok=True)
        finally:
            Path(video_path).unlink(missing_ok=True)

    @patch("video_processing.ffmpeg_utils.run_ffmpeg")
    @patch("video_processing.ffmpeg_utils.probe_duration", return_value=20.0)
    def test_handles_ffmpeg_failure_gracefully(self, mock_probe, mock_run):
        """FFmpeg 失败时跳过该帧，继续抽取其他帧."""
        import tempfile

        from video_processing.thumbnail_generator import extract_cover_candidates

        call_count = 0

        def fake_run(cmd, **kwargs):
            nonlocal call_count
            call_count += 1
            output_path = cmd[-1]
            if call_count == 2:
                # Second frame fails - don't create file
                raise RuntimeError("ffmpeg error")
            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
            return ("", "")

        mock_run.side_effect = fake_run

        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
            tmp.write(b"fake video")
            video_path = tmp.name

        try:
            results = extract_cover_candidates(video_path, num_frames=3)
            # Should get 2 frames (1st and 3rd), 2nd failed
            assert len(results) == 2
        finally:
            Path(video_path).unlink(missing_ok=True)
            for r in results:
                Path(r["local_path"]).unlink(missing_ok=True)

    @patch("video_processing.ffmpeg_utils.probe_duration", side_effect=Exception("probe failed"))
    def test_fallback_duration_when_probe_fails(self, mock_probe):
        """probe 失败时使用默认时长."""
        import tempfile

        from video_processing.thumbnail_generator import extract_cover_candidates

        # Mock run_ffmpeg to create output files
        def fake_run(cmd, **kwargs):
            output_path = cmd[-1]
            Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
            return ("", "")

        with patch("video_processing.ffmpeg_utils.run_ffmpeg", side_effect=fake_run):
            with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as tmp:
                tmp.write(b"fake")
                video_path = tmp.name

            try:
                results = extract_cover_candidates(video_path, num_frames=3)
                assert len(results) == 3
                # Default duration is 5.0, so times should be 5*0.25=1.25, 5*0.5=2.5, 5*0.75=3.75
                assert results[0]["frame_time"] == 1.25
                assert results[1]["frame_time"] == 2.5
                assert results[2]["frame_time"] == 3.75
            finally:
                Path(video_path).unlink(missing_ok=True)
                for r in results:
                    Path(r["local_path"]).unlink(missing_ok=True)


class TestExtractAndUploadCoverFrames:
    """extract_and_upload_cover_frames 测试."""

    @patch("video_processing.oss_helpers.upload_to_oss")
    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
    def test_uploads_and_returns_correct_format(self, mock_extract, mock_upload):
        """上传帧到 OSS 并返回正确格式."""
        import tempfile

        from video_processing.thumbnail_generator import extract_and_upload_cover_frames

        # Create actual temp files
        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
        tmp1.close()
        tmp2 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp2.write(b"\xff\xd8" + b"\x00" * 50)
        tmp2.close()

        mock_extract.return_value = [
            {"local_path": tmp1.name, "frame_time": 5.0},
            {"local_path": tmp2.name, "frame_time": 10.0},
        ]
        mock_upload.side_effect = [
            "https://oss.example.com/covers/plan1/frame_0.jpg",
            "https://oss.example.com/covers/plan1/frame_1.jpg",
        ]

        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")

        assert len(results) == 2
        assert results[0]["image_url"] == "https://oss.example.com/covers/plan1/frame_0.jpg"
        assert results[0]["frame_time"] == 5.0
        assert results[0]["storage_key"] == "covers/plan1/frame_0.jpg"

        assert results[1]["image_url"] == "https://oss.example.com/covers/plan1/frame_1.jpg"
        assert results[1]["frame_time"] == 10.0

    @patch("video_processing.thumbnail_generator.extract_cover_candidates", return_value=[])
    def test_returns_empty_when_no_candidates(self, mock_extract):
        """没有候选帧时返回空列表."""
        from video_processing.thumbnail_generator import extract_and_upload_cover_frames

        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
        assert results == []

    @patch("video_processing.oss_helpers.upload_to_oss", side_effect=Exception("OSS error"))
    @patch("video_processing.thumbnail_generator.extract_cover_candidates")
    def test_handles_upload_failure_gracefully(self, mock_extract, mock_upload):
        """上传失败时跳过该帧."""
        import tempfile

        from video_processing.thumbnail_generator import extract_and_upload_cover_frames

        tmp1 = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp1.write(b"\xff\xd8" + b"\x00" * 50)
        tmp1.close()

        mock_extract.return_value = [
            {"local_path": tmp1.name, "frame_time": 5.0},
        ]

        results = extract_and_upload_cover_frames("/tmp/video.mp4", "plan1")
        assert results == []


class TestRenderAdapterResultCoverCandidates:
    """RenderAdapterResult 的 cover_candidates 字段."""

    def test_default_none(self):
        """默认为 None."""
        from video_processing.render_adapter import RenderAdapterResult

        result = RenderAdapterResult(success=True)
        assert result.cover_candidates is None

    def test_can_set_candidates(self):
        """可以设置候选帧列表."""
        from video_processing.render_adapter import RenderAdapterResult

        candidates = [
            {"image_url": "https://example.com/frame_0.jpg", "frame_time": 5.0, "storage_key": "covers/p1/frame_0.jpg"},
        ]
        result = RenderAdapterResult(success=True, cover_candidates=candidates)
        assert len(result.cover_candidates) == 1
        assert result.cover_candidates[0]["frame_time"] == 5.0


class TestAICoverServiceFFmpegFallback:
    """AI 封面服务 FFmpeg 兜底测试."""

    @patch("packages.shared.ai_service.http_requests.head")
    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
    def test_ffmpeg_fallback_success(self, mock_ffmpeg, mock_head):
        """FFmpeg 兜底抽帧成功."""
        import tempfile

        mock_head.return_value.status_code = 200

        tmp = tempfile.NamedTemporaryFile(suffix=".jpg", delete=False)
        tmp.write(b"\xff\xd8" + b"\x00" * 50)
        tmp.close()

        mock_ffmpeg.return_value = [{"local_path": tmp.name, "frame_time": 5.0}]

        # Mock storage
        with patch("packages.shared.storage.get_shared_storage_service") as mock_storage_fn:
            mock_storage = Mock()
            mock_storage.upload_file = Mock()
            mock_storage.get_url.return_value = "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
            mock_storage_fn.return_value = mock_storage

            from packages.shared.ai_service import _call_ai_cover_service

            result = _call_ai_cover_service(
                plan_id="plan1",
                asset_ids=["a1"],
                cover_type="ai_frame",
                primary_video_url="https://example.com/video.mp4",
            )

            assert result["type"] == "ai_frame"
            assert result["image_url"] == "https://oss.example.com/covers/plan1/ffmpeg_frame_abc.jpg"
            assert result["frame_time"] == 5.0
            assert result["confidence"] == 0.85

        Path(tmp.name).unlink(missing_ok=True)

    @patch("packages.shared.ai_service.http_requests.head")
    def test_ffmpeg_no_video_url_raises(self, mock_head):
        """没有视频 URL 时抛出 RuntimeError."""
        from packages.shared.ai_service import _call_ai_cover_service

        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
            _call_ai_cover_service(
                plan_id="plan1",
                asset_ids=["a1"],
                cover_type="ai_frame",
                primary_video_url=None,
            )

    @patch("packages.shared.ai_service.http_requests.head")
    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
    def test_ffmpeg_no_frames_raises(self, mock_ffmpeg, mock_head):
        """FFmpeg 抽帧为空时抛出 RuntimeError."""
        mock_head.return_value.status_code = 200
        mock_ffmpeg.return_value = []

        from packages.shared.ai_service import _call_ai_cover_service

        with pytest.raises(RuntimeError, match="无法从视频抽帧"):
            _call_ai_cover_service(
                plan_id="plan1",
                asset_ids=["a1"],
                cover_type="ai_frame",
                primary_video_url="https://example.com/video.mp4",
            )

    def test_upload_type_returns_immediately(self):
        """upload 类型直接返回."""
        from packages.shared.ai_service import _call_ai_cover_service

        result = _call_ai_cover_service(
            plan_id="plan1",
            asset_ids=["a1"],
            cover_type="upload",
            primary_video_url="https://example.com/video.mp4",
        )
        assert result["type"] == "upload"

    def test_manual_type_returns_immediately(self):
        """manual 类型直接返回."""
        from packages.shared.ai_service import _call_ai_cover_service

        result = _call_ai_cover_service(
            plan_id="plan1",
            asset_ids=["a1"],
            cover_type="manual",
            frame_time=5.0,
            primary_video_url="https://example.com/video.mp4",
        )
        assert result["type"] == "manual"
        assert result["frame_time"] == 5.0

    @patch("packages.shared.ai_service.http_requests.head")
    @patch("packages.shared.ai_service._extract_frames_with_ffmpeg")
    def test_video_url_unreachable_raises(self, mock_ffmpeg, mock_head):
        """视频 URL 不可访问时抛出 RuntimeError."""
        mock_head.return_value.status_code = 404

        from packages.shared.ai_service import _call_ai_cover_service

        with pytest.raises(RuntimeError, match="预览视频URL不可访问"):
            _call_ai_cover_service(
                plan_id="plan1",
                asset_ids=["a1"],
                cover_type="ai_frame",
                primary_video_url="https://example.com/video.mp4",
            )


class TestCoverTemplatesFix:
    """CoverTemplateResponse config=None 修复测试."""

    def test_config_none_becomes_empty_dict(self):
        """config=None 时 CoverTemplateResponse 不报 ValidationError."""
        from datetime import datetime

        from app.schemas.cover_template import CoverTemplateResponse

        # This should not raise
        resp = CoverTemplateResponse(
            id="1",
            name="test",
            thumbnail_url="",
            is_system=True,
            created_at=datetime.now(),
            config={},
        )
        assert resp.config == {}


class TestExtractFramesWithFFmpeg:
    """_extract_frames_with_ffmpeg 单元测试."""

    def test_extracts_frames_with_correct_seek_times(self):
        """抽帧时间点正确计算."""
        import subprocess
        import tempfile

        # Mock ffprobe to return duration
        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="20.0\n", stderr="")
        with patch("subprocess.run", return_value=mock_probe_result) as mock_subproc:
            # First call is ffprobe, rest are ffmpeg
            call_count = 0

            def side_effect(cmd, **kwargs):
                nonlocal call_count
                call_count += 1
                if call_count == 1:
                    # ffprobe call
                    return mock_probe_result
                else:
                    # ffmpeg call - create output file
                    output_path = cmd[-1]
                    from pathlib import Path

                    Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
                    return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")

            mock_subproc.side_effect = side_effect

            from packages.shared.ai_service import _extract_frames_with_ffmpeg

            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)

            assert len(results) == 3
            # 20 * 0.25 = 5.0, 20 * 0.5 = 10.0, 20 * 0.75 = 15.0
            assert results[0]["frame_time"] == 5.0
            assert results[1]["frame_time"] == 10.0
            assert results[2]["frame_time"] == 15.0

            # Clean up
            for r in results:
                from pathlib import Path

                Path(r["local_path"]).unlink(missing_ok=True)

    def test_handles_ffmpeg_failure(self):
        """FFmpeg 失败时跳过该帧."""
        import subprocess
        import tempfile
        from pathlib import Path

        mock_probe_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="10.0\n", stderr="")

        call_count = 0

        def side_effect(cmd, **kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                return mock_probe_result
            output_path = cmd[-1]
            if call_count == 2:
                # First frame succeeds
                Path(output_path).write_bytes(b"\xff\xd8" + b"\x00" * 50)
                return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
            else:
                # Other frames fail
                raise subprocess.CalledProcessError(1, cmd)

        with patch("subprocess.run", side_effect=side_effect):
            from packages.shared.ai_service import _extract_frames_with_ffmpeg

            results = _extract_frames_with_ffmpeg("https://example.com/video.mp4", num_frames=3)
            assert len(results) == 1
            Path(results[0]["local_path"]).unlink(missing_ok=True)


class TestGenerationCoverPreStored:
    """generation_cover.py 预存帧逻辑测试."""

    def test_pre_stored_candidates_used_when_available(self):
        """有预存帧时直接使用，不调用 AI 服务."""
        from unittest.mock import patch

        # Mock the dependencies
        mock_plan = MagicMock()
        mock_plan.config = {
            "cover_candidates": [
                {
                    "image_url": "https://oss.example.com/covers/p1/frame_0.jpg",
                    "frame_time": 5.0,
                    "storage_key": "covers/p1/frame_0.jpg",
                },
                {
                    "image_url": "https://oss.example.com/covers/p1/frame_1.jpg",
                    "frame_time": 10.0,
                    "storage_key": "covers/p1/frame_1.jpg",
                },
            ],
            "rendered_storage_key": "rendered/p1/video.mp4",
        }

        mock_plan_svc = MagicMock()
        mock_plan_svc.get_plan_or_raise.return_value = mock_plan

        mock_body = MagicMock()
        mock_body.asset_ids = ["a1"]
        mock_body.cover_type = "ai_frame"
        mock_body.frame_time = None

        with (
            patch("app.api.routes.generation_cover.get_editor_services") as mock_services,
            patch("app.api.routes.generation_cover.get_db_session"),
            patch("app.api.routes.generation_cover.get_current_user"),
            patch("app.api.routes.generation_cover.get_draft_plan_id", return_value="p1"),
            patch("app.api.routes.generation_cover.normalize_plan_config") as mock_normalize,
        ):

            mock_services.return_value = (MagicMock(), mock_plan_svc)
            mock_normalize.side_effect = lambda c: c

            from app.api.routes.generation_cover import GenerateCoverRequest, generate_cover

            result = generate_cover(
                body=mock_body,
                template_id="t1",
                plan_id="p1",
                services=(MagicMock(), mock_plan_svc),
                db=MagicMock(),
                current_user=MagicMock(),
            )

            assert result.plan_id == "p1"
            assert result.cover["type"] == "ai_frame"
            assert result.cover["image_url"] == "https://oss.example.com/covers/p1/frame_0.jpg"
            assert result.cover["frame_time"] == 5.0