xiaoxia-saas/tests/unit/test_cover_service.py

"""
封面管理服务单元测试
"""

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

import pytest

from apps.api.app.services.cover_service import (
    COVER_STORAGE_PREFIX,
    DEFAULT_COVER_HEIGHT,
    DEFAULT_COVER_QUALITY,
    DEFAULT_COVER_WIDTH,
    CoverService,
)


class TestGetCoverConfig:
    """get_cover_config 静态方法测试"""

    def test_get_cover_config_default(self):
        """测试默认封面配置"""
        config = {}
        result = CoverService.get_cover_config(config)
        assert result["type"] == "ai_frame"
        assert result["image_url"] == ""
        assert result["frame_time"] is None

    def test_get_cover_config_with_custom_values(self):
        """测试自定义封面配置"""
        config = {
            "cover": {
                "type": "manual",
                "image_url": "https://example.com/cover.jpg",
                "frame_time": 5.5,
            }
        }
        result = CoverService.get_cover_config(config)
        assert result["type"] == "manual"
        assert result["image_url"] == "https://example.com/cover.jpg"
        assert result["frame_time"] == 5.5

    def test_get_cover_config_cover_not_dict(self):
        """测试 cover 不是 dict 时返回默认值"""
        config = {"cover": "not-a-dict"}
        result = CoverService.get_cover_config(config)
        assert result["type"] == "ai_frame"
        assert result["image_url"] == ""
        assert result["frame_time"] is None

    def test_get_cover_config_partial_fields(self):
        """测试部分字段存在时，其余字段用默认值"""
        config = {"cover": {"type": "custom"}}
        result = CoverService.get_cover_config(config)
        assert result["type"] == "custom"
        assert result["image_url"] == ""
        assert result["frame_time"] is None

    def test_get_cover_config_empty_cover_dict(self):
        """测试空的 cover dict"""
        config = {"cover": {}}
        result = CoverService.get_cover_config(config)
        assert result["type"] == "ai_frame"
        assert result["image_url"] == ""


class TestExtractCoverFromClip:
    """extract_cover_from_clip 测试"""

    @pytest.fixture
    def mock_storage(self):
        storage = Mock()
        storage.download_file = Mock()
        storage.upload_file = Mock()
        storage.get_url = Mock(return_value="https://oss.example.com/covers/plan1/cover_1000.jpg")
        return storage

    @pytest.fixture
    def mock_asset_repo(self):
        repo = Mock()
        repo.get = Mock(return_value=None)
        return repo

    @pytest.fixture
    def video_asset(self):
        asset = Mock()
        asset.storage_key = "videos/test-video.mp4"
        asset.mime_type = "video/mp4"
        return asset

    @pytest.fixture
    def service(self, mock_storage, mock_asset_repo):
        return CoverService(storage_service=mock_storage, asset_repository=mock_asset_repo)

    def test_extract_cover_asset_not_found(self, service, mock_asset_repo):
        """测试素材不存在时报错"""
        mock_asset_repo.get.return_value = None

        with pytest.raises(ValueError, match="素材不存在"):
            service.extract_cover_from_clip(plan_id="plan-1", asset_id="nonexistent")

    def test_extract_cover_asset_no_storage_key(self, service, mock_asset_repo):
        """测试素材没有文件时报错"""
        asset = Mock()
        asset.storage_key = ""
        asset.mime_type = "video/mp4"
        mock_asset_repo.get.return_value = asset

        with pytest.raises(ValueError, match="素材没有文件"):
            service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-no-file")

    def test_extract_cover_asset_not_video(self, service, mock_asset_repo):
        """测试非视频素材报错"""
        asset = Mock()
        asset.storage_key = "images/photo.jpg"
        asset.mime_type = "image/jpeg"
        mock_asset_repo.get.return_value = asset

        with pytest.raises(ValueError, match="素材不是视频类型"):
            service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-img")

    def test_extract_cover_download_failure(self, service, mock_asset_repo, mock_storage, video_asset):
        """测试下载素材失败"""
        mock_asset_repo.get.return_value = video_asset
        mock_storage.download_file.side_effect = Exception("网络错误")

        with pytest.raises(RuntimeError, match="下载素材失败"):
            service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")

    def test_extract_cover_upload_failure(self, service, mock_asset_repo, mock_storage, video_asset):
        """测试上传封面失败"""
        mock_asset_repo.get.return_value = video_asset

        def fake_download(storage_key, local_path):
            # 创建一个假的视频文件
            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
            with open(local_path, "wb") as f:
                f.write(b"fake video data")

        mock_storage.download_file.side_effect = fake_download
        mock_storage.upload_file.side_effect = Exception("上传失败")

        # mock _extract_frame 避免真的调 ffmpeg
        with patch.object(CoverService, "_extract_frame") as mock_extract:

            def fake_extract(video_path, output_path, **kwargs):
                # 创建假的封面文件
                with open(output_path, "wb") as f:
                    f.write(b"\xff\xd8\xff\xe0fake jpeg data")

            mock_extract.side_effect = fake_extract

            with pytest.raises(RuntimeError, match="上传封面失败"):
                service.extract_cover_from_clip(plan_id="plan-1", asset_id="asset-1")

    def test_extract_cover_get_url_falls_back_to_key(self, service, mock_asset_repo, mock_storage, video_asset):
        """测试获取 URL 失败时降级为 storage_key"""
        mock_asset_repo.get.return_value = video_asset

        def fake_download(storage_key, local_path):
            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
            with open(local_path, "wb") as f:
                f.write(b"fake video data")

        mock_storage.download_file.side_effect = fake_download
        mock_storage.get_url.side_effect = Exception("URL服务不可用")

        with patch.object(CoverService, "_extract_frame") as mock_extract:

            def fake_extract(video_path, output_path, **kwargs):
                with open(output_path, "wb") as f:
                    f.write(b"\xff\xd8\xff\xe0fake jpeg")

            mock_extract.side_effect = fake_extract

            result = service.extract_cover_from_clip(plan_id="plan-abc", asset_id="asset-xyz", frame_time=2.5)

        assert result["type"] == "manual"
        assert result["frame_time"] == 2.5
        # URL 失败时返回 storage_key
        assert COVER_STORAGE_PREFIX in result["image_url"]
        assert "plan-abc" in result["image_url"]

    def test_extract_cover_success(self, service, mock_asset_repo, mock_storage, video_asset):
        """测试抽帧成功完整流程"""
        mock_asset_repo.get.return_value = video_asset

        def fake_download(storage_key, local_path):
            Path(local_path).parent.mkdir(parents=True, exist_ok=True)
            with open(local_path, "wb") as f:
                f.write(b"fake video data for testing")

        mock_storage.download_file.side_effect = fake_download

        with patch.object(CoverService, "_extract_frame") as mock_extract:

            def fake_extract(video_path, output_path, **kwargs):
                with open(output_path, "wb") as f:
                    f.write(b"\xff\xd8\xff\xe0fake jpeg image data")

            mock_extract.side_effect = fake_extract

            result = service.extract_cover_from_clip(
                plan_id="plan-123",
                asset_id="asset-456",
                frame_time=3.0,
                width=720,
                height=1280,
                quality=3,
            )

        assert result["type"] == "manual"
        assert result["image_url"] == "https://oss.example.com/covers/plan1/cover_1000.jpg"
        assert result["frame_time"] == 3.0

        # 验证上传被调用
        mock_storage.upload_file.assert_called_once()
        upload_args = mock_storage.upload_file.call_args[1]
        assert upload_args["content_type"] == "image/jpeg"
        assert "plan-123" in upload_args["storage_key"]
        assert "3000" in upload_args["storage_key"]  # frame_time * 1000

        # 验证 _extract_frame 被调用且参数正确
        mock_extract.assert_called_once()
        extract_kwargs = mock_extract.call_args[1]
        assert extract_kwargs["time_sec"] == 3.0
        assert extract_kwargs["width"] == 720
        assert extract_kwargs["height"] == 1280
        assert extract_kwargs["quality"] == 3


class TestGenerateSmartCover:
    """generate_smart_cover 测试"""

    @pytest.fixture
    def service(self):
        return CoverService(storage_service=Mock(), asset_repository=Mock())

    def test_generate_smart_cover_calls_extract_with_default_time(self, service):
        """测试智能封面调用 extract_cover_from_clip 并设置 type 为 ai_frame"""
        fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}

        with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
            result = service.generate_smart_cover(plan_id="plan-1", asset_id="asset-1")

            mock_extract.assert_called_once()
            call_kwargs = mock_extract.call_args[1]
            assert call_kwargs["plan_id"] == "plan-1"
            assert call_kwargs["asset_id"] == "asset-1"
            assert call_kwargs["frame_time"] == 3.0  # 默认第3秒

        assert result["type"] == "ai_frame"
        assert result["image_url"] == "test.jpg"

    def test_generate_smart_cover_passes_dimensions(self, service):
        """测试智能封面传递尺寸和质量参数"""
        fake_result = {"type": "manual", "image_url": "test.jpg", "frame_time": 3.0}

        with patch.object(service, "extract_cover_from_clip", return_value=fake_result) as mock_extract:
            service.generate_smart_cover(
                plan_id="plan-1",
                asset_id="asset-1",
                width=1080,
                height=1920,
                quality=5,
            )

            call_kwargs = mock_extract.call_args[1]
            assert call_kwargs["width"] == 1080
            assert call_kwargs["height"] == 1920
            assert call_kwargs["quality"] == 5


class TestExtractFrame:
    """_extract_frame 静态方法测试（mock subprocess）"""

    @pytest.fixture
    def video_path(self, tmp_path):
        path = tmp_path / "test_video.mp4"
        path.write_bytes(b"fake video")
        return path

    @pytest.fixture
    def output_path(self, tmp_path):
        return tmp_path / "cover.jpg"

    def test_extract_frame_success(self, video_path, output_path):
        """测试 FFmpeg 抽帧成功"""
        fake_result = Mock()
        fake_result.returncode = 0

        with patch("subprocess.run", return_value=fake_result) as mock_run:
            CoverService._extract_frame(
                video_path=video_path,
                output_path=output_path,
                time_sec=2.5,
                width=1080,
                height=1920,
                quality=5,
            )

            assert mock_run.call_count == 1
            cmd = mock_run.call_args[0][0]
            assert cmd[0] == "ffmpeg"
            assert "-ss" in cmd
            assert "2.500" in cmd
            assert "-vframes" in cmd
            # 验证 scale+crop 滤镜存在
            vf_index = cmd.index("-vf") + 1
            assert "scale=" in cmd[vf_index]
            assert "crop=" in cmd[vf_index]

    def test_extract_frame_fallback_to_simple_command(self, video_path, output_path):
        """测试主命令失败时回退到简化命令"""
        fail_result = Mock()
        fail_result.returncode = 1
        fail_result.stderr = "Filter graph error"

        success_result = Mock()
        success_result.returncode = 0

        call_count = 0

        def fake_run(*args, **kwargs):
            nonlocal call_count
            call_count += 1
            if call_count == 1:
                return fail_result
            return success_result

        with patch("subprocess.run", side_effect=fake_run) as mock_run:
            CoverService._extract_frame(
                video_path=video_path,
                output_path=output_path,
                time_sec=1.0,
                width=1080,
                height=1920,
                quality=5,
            )

            assert mock_run.call_count == 2
            # 第二次是简化命令（没有 -vf 参数）
            second_cmd = mock_run.call_args_list[1][0][0]
            assert "-vf" not in second_cmd

    def test_extract_frame_both_commands_fail(self, video_path, output_path):
        """测试两个命令都失败时报错"""
        fail_result = Mock()
        fail_result.returncode = 1
        fail_result.stderr = "Invalid data found when processing input"

        with patch("subprocess.run", return_value=fail_result):
            with pytest.raises(RuntimeError, match="FFmpeg 抽帧失败"):
                CoverService._extract_frame(
                    video_path=video_path,
                    output_path=output_path,
                    time_sec=1.0,
                    width=1080,
                    height=1920,
                    quality=5,
                )

    def test_extract_frame_timeout(self, video_path, output_path):
        """测试 FFmpeg 抽帧超时"""
        with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="ffmpeg", timeout=60)):
            with pytest.raises(RuntimeError, match="FFmpeg 抽帧超时"):
                CoverService._extract_frame(
                    video_path=video_path,
                    output_path=output_path,
                    time_sec=1.0,
                    width=1080,
                    height=1920,
                    quality=5,
                )

    def test_extract_frame_ffmpeg_not_found(self, video_path, output_path):
        """测试 FFmpeg 不可用"""
        with patch("subprocess.run", side_effect=FileNotFoundError("ffmpeg not found")):
            with pytest.raises(RuntimeError, match="FFmpeg 不可用"):
                CoverService._extract_frame(
                    video_path=video_path,
                    output_path=output_path,
                    time_sec=1.0,
                    width=1080,
                    height=1920,
                    quality=5,
                )


class TestDefaults:
    """默认常量测试"""

    def test_default_dimensions(self):
        """测试默认尺寸常量"""
        assert DEFAULT_COVER_WIDTH == 1080
        assert DEFAULT_COVER_HEIGHT == 1920
        assert DEFAULT_COVER_QUALITY == 5
        assert COVER_STORAGE_PREFIX == "covers"