xiaoxia-saas/tests/unit/test_mediakit_cover.py

"""MediaKit AI 封面生成单元测试.

测试 #1208: AI封面接入MediaKit视频截帧
"""

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

import pytest

from packages.shared.mediakit_client import MediaKitClient, get_mediakit_client


class TestMediaKitClient:
    """MediaKit 客户端测试."""

    def test_is_available_with_key(self):
        """配置了 API Key 时可用."""
        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value.mediakit_api_key = "test-key"
            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
            mock_settings.return_value.mediakit_timeout = 60

            client = MediaKitClient()
            assert client.is_available is True

    def test_is_available_without_key(self):
        """未配置 API Key 时不可用."""
        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value.mediakit_api_key = ""
            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
            mock_settings.return_value.mediakit_timeout = 60

            client = MediaKitClient()
            assert client.is_available is False

    @patch("packages.shared.mediakit_client.httpx.post")
    @patch("packages.shared.mediakit_client.httpx.get")
    def test_extract_frames_success(self, mock_get, mock_post):
        """抽帧成功."""
        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value.mediakit_api_key = "test-key"
            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
            mock_settings.return_value.mediakit_timeout = 60

            # Mock submit task
            mock_post.return_value.status_code = 200
            mock_post.return_value.json.return_value = {"task_id": "task-123"}
            mock_post.return_value.raise_for_status = Mock()

            # Mock poll task
            mock_get.return_value.status_code = 200
            mock_get.return_value.json.return_value = {
                "status": "success",
                "result": {
                    "snapshots": [
                        {"image_url": "https://example.com/frame1.jpg", "timestamp": 2.5},
                        {"image_url": "https://example.com/frame2.jpg", "timestamp": 5.0},
                    ]
                },
            }
            mock_get.return_value.raise_for_status = Mock()

            client = MediaKitClient()
            frames = client.extract_frames(
                video_url="https://example.com/video.mp4",
                strategy="SceneChange",
                max_frames=5,
            )

            assert frames is not None
            assert len(frames) == 2
            assert frames[0]["image_url"] == "https://example.com/frame1.jpg"
            assert frames[0]["timestamp"] == 2.5

    @patch("packages.shared.mediakit_client.httpx.post")
    def test_extract_frames_submit_failed(self, mock_post):
        """提交任务失败."""
        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value.mediakit_api_key = "test-key"
            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
            mock_settings.return_value.mediakit_timeout = 60

            mock_post.side_effect = Exception("Network error")

            client = MediaKitClient()
            frames = client.extract_frames(
                video_url="https://example.com/video.mp4",
                strategy="SceneChange",
                max_frames=5,
            )

            assert frames is None

    @patch("packages.shared.mediakit_client.httpx.post")
    @patch("packages.shared.mediakit_client.httpx.get")
    def test_extract_frames_task_failed(self, mock_get, mock_post):
        """任务执行失败."""
        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value.mediakit_api_key = "test-key"
            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
            mock_settings.return_value.mediakit_timeout = 60

            # Mock submit task
            mock_post.return_value.status_code = 200
            mock_post.return_value.json.return_value = {"task_id": "task-123"}
            mock_post.return_value.raise_for_status = Mock()

            # Mock poll task - failed
            mock_get.return_value.status_code = 200
            mock_get.return_value.json.return_value = {
                "status": "failed",
                "error": "Video processing failed",
            }
            mock_get.return_value.raise_for_status = Mock()

            client = MediaKitClient()
            frames = client.extract_frames(
                video_url="https://example.com/video.mp4",
                strategy="SceneChange",
                max_frames=5,
            )

            assert frames is None

    def test_extract_frames_not_available(self):
        """客户端不可用时返回 None."""
        with patch("packages.shared.mediakit_client.get_shared_settings") as mock_settings:
            mock_settings.return_value.mediakit_api_key = ""
            mock_settings.return_value.mediakit_base_url = "https://test.com/api/v1"
            mock_settings.return_value.mediakit_timeout = 60

            client = MediaKitClient()
            frames = client.extract_frames(
                video_url="https://example.com/video.mp4",
                strategy="SceneChange",
                max_frames=5,
            )

            assert frames is None


class TestAICoverService:
    """AI 封面服务测试（统一封面管道后）。"""

    def test_call_ai_cover_ai_frame_raises(self):
        """ai_frame type raises RuntimeError in unified pipeline."""
        from packages.shared.ai_service import _call_ai_cover_service

        with pytest.raises(RuntimeError, match="封面数据不可用"):
            _call_ai_cover_service(
                plan_id="plan-123",
                asset_ids=["asset-1"],
                cover_type="ai_frame",
                primary_video_url="https://example.com/video.mp4",
            )

    def test_call_ai_cover_ai_regenerate_raises(self):
        """ai_regenerate type raises RuntimeError in unified pipeline."""
        from packages.shared.ai_service import _call_ai_cover_service

        with pytest.raises(RuntimeError, match="封面数据不可用"):
            _call_ai_cover_service(
                plan_id="plan-123",
                asset_ids=["asset-1"],
                cover_type="ai_regenerate",
            )

    def test_call_ai_cover_without_video_url_raises(self):
        """ai_frame without video URL still raises RuntimeError."""
        from packages.shared.ai_service import _call_ai_cover_service

        with pytest.raises(RuntimeError, match="封面数据不可用"):
            _call_ai_cover_service(
                plan_id="plan-123",
                asset_ids=["asset-1"],
                cover_type="ai_frame",
                primary_video_url=None,
            )

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

        result = _call_ai_cover_service(
            plan_id="plan-123",
            asset_ids=["asset-1"],
            cover_type="upload",
            primary_video_url="https://example.com/video.mp4",
        )

        assert result["type"] == "upload"
        assert result["image_url"] == ""

    def test_call_ai_cover_manual_type(self):
        """manual 类型返回指定时间点."""
        from packages.shared.ai_service import _call_ai_cover_service

        result = _call_ai_cover_service(
            plan_id="plan-123",
            asset_ids=["asset-1"],
            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


class TestGenerateCover:
    """run_generate_cover 测试."""

    @patch("packages.shared.ai_service._call_ai_cover_service")
    def test_run_generate_cover_passes_video_url(self, mock_cover_service):
        """正确传递 video_url 参数."""
        mock_cover_service.return_value = {
            "type": "ai_frame",
            "image_url": "https://example.com/cover.jpg",
            "frame_time": 2.0,
            "confidence": 0.9,
        }

        from packages.shared.ai_service import run_generate_cover

        result = run_generate_cover(
            plan_id="plan-123",
            asset_ids=["asset-1"],
            cover_type="ai_frame",
            primary_video_url="https://example.com/video.mp4",
        )

        assert result["type"] == "ai_frame"
        mock_cover_service.assert_called_once()
        call_kwargs = mock_cover_service.call_args[1]
        assert call_kwargs["primary_video_url"] == "https://example.com/video.mp4"

    @patch("packages.shared.ai_service._call_ai_cover_service")
    def test_run_generate_cover_without_video_url(self, mock_cover_service):
        """不传 video_url 时默认为 None."""
        mock_cover_service.return_value = {
            "type": "ai_frame",
            "image_url": "https://example.com/cover.jpg",
            "frame_time": 2.0,
            "confidence": 0.9,
        }

        from packages.shared.ai_service import run_generate_cover

        result = run_generate_cover(
            plan_id="plan-123",
            asset_ids=["asset-1"],
            cover_type="ai_frame",
        )

        assert result["type"] == "ai_frame"
        call_kwargs = mock_cover_service.call_args[1]
        assert call_kwargs["primary_video_url"] is None