xiaoxia-saas/tests/unit/test_generated_videos_use_cases.py

"""生成视频 UseCase 单元测试."""

from __future__ import annotations

from unittest.mock import MagicMock

import pytest

from packages.application.generated_videos import (
    GetGeneratedVideoDownloadUrlUseCase,
    GetGeneratedVideoUseCase,
    GetVideosByIdsUseCase,
    ListGeneratedVideosByTaskUseCase,
    ListGeneratedVideosPaginatedUseCase,
    ListGeneratedVideosUseCase,
    UpdateVideoReviewStatusUseCase,
)
from packages.domain.generated_video import GeneratedVideo


@pytest.fixture
def mock_repo():
    return MagicMock()


@pytest.fixture
def sample_video():
    v = GeneratedVideo.create(
        project_id="proj_1",
        generation_task_id="task_1",
        name="测试视频",
        file_url="https://oss.example.com/videos/test.mp4",
        user_id="user_1",
        file_size=1024000,
        duration=30.5,
        width=1920,
        height=1080,
        fps=30.0,
    )
    v.id = "video_123"
    return v


class TestListGeneratedVideosUseCase:
    """ListGeneratedVideosUseCase 测试"""

    def test_list_returns_results(self, mock_repo, sample_video):
        """正常返回项目生成视频列表"""
        mock_repo.list_by_project.return_value = [sample_video]
        use_case = ListGeneratedVideosUseCase(mock_repo)

        result = use_case.execute("proj_1")

        assert len(result) == 1
        assert result[0].id == "video_123"
        mock_repo.list_by_project.assert_called_once_with("proj_1")

    def test_empty_project_id_raises(self, mock_repo):
        """空 project_id 抛出"""
        use_case = ListGeneratedVideosUseCase(mock_repo)

        with pytest.raises(ValueError, match="project_id 不能为空"):
            use_case.execute("")

        mock_repo.list_by_project.assert_not_called()

    def test_whitespace_project_id_raises(self, mock_repo):
        """纯空格 project_id 抛出"""
        use_case = ListGeneratedVideosUseCase(mock_repo)

        with pytest.raises(ValueError, match="project_id 不能为空"):
            use_case.execute("   ")

    def test_project_id_stripped(self, mock_repo, sample_video):
        """project_id 会被 strip"""
        mock_repo.list_by_project.return_value = [sample_video]
        use_case = ListGeneratedVideosUseCase(mock_repo)

        use_case.execute("  proj_1  ")

        mock_repo.list_by_project.assert_called_once_with("proj_1")


class TestListGeneratedVideosPaginatedUseCase:
    """ListGeneratedVideosPaginatedUseCase 测试"""

    def test_paginated_default_params(self, mock_repo, sample_video):
        """默认分页参数正确传递"""
        mock_repo.list_paginated.return_value = ([sample_video], 1)
        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)

        results, total = use_case.execute(user_id="user_1")

        assert len(results) == 1
        assert total == 1
        mock_repo.list_paginated.assert_called_once_with(
            user_id="user_1",
            project_id=None,
            status=None,
            review_status=None,
            page=1,
            page_size=20,
        )

    def test_page_less_than_1_clamped(self, mock_repo, sample_video):
        """page < 1 被修正为 1"""
        mock_repo.list_paginated.return_value = ([sample_video], 1)
        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)

        use_case.execute(page=0)

        call_kwargs = mock_repo.list_paginated.call_args[1]
        assert call_kwargs["page"] == 1

    def test_page_size_less_than_1_clamped(self, mock_repo, sample_video):
        """page_size < 1 被修正为 20"""
        mock_repo.list_paginated.return_value = ([sample_video], 1)
        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)

        use_case.execute(page_size=0)

        call_kwargs = mock_repo.list_paginated.call_args[1]
        assert call_kwargs["page_size"] == 20

    def test_page_size_greater_than_100_clamped(self, mock_repo, sample_video):
        """page_size > 100 被修正为 20"""
        mock_repo.list_paginated.return_value = ([sample_video], 1)
        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)

        use_case.execute(page_size=200)

        call_kwargs = mock_repo.list_paginated.call_args[1]
        assert call_kwargs["page_size"] == 20

    def test_full_filter_params(self, mock_repo, sample_video):
        """所有过滤参数正确传递"""
        mock_repo.list_paginated.return_value = ([sample_video], 1)
        use_case = ListGeneratedVideosPaginatedUseCase(mock_repo)

        use_case.execute(
            user_id="user_1",
            project_id="proj_1",
            status="completed",
            review_status="approved",
            page=2,
            page_size=50,
        )

        mock_repo.list_paginated.assert_called_once_with(
            user_id="user_1",
            project_id="proj_1",
            status="completed",
            review_status="approved",
            page=2,
            page_size=50,
        )


class TestGetGeneratedVideoUseCase:
    """GetGeneratedVideoUseCase 测试"""

    def test_get_existing(self, mock_repo, sample_video):
        """获取存在的视频"""
        mock_repo.get.return_value = sample_video
        use_case = GetGeneratedVideoUseCase(mock_repo)

        result = use_case.execute("video_123")

        assert result is not None
        assert result.id == "video_123"

    def test_get_nonexistent_returns_none(self, mock_repo):
        """获取不存在的视频返回 None"""
        mock_repo.get.return_value = None
        use_case = GetGeneratedVideoUseCase(mock_repo)

        result = use_case.execute("nonexistent")

        assert result is None


class TestListGeneratedVideosByTaskUseCase:
    """ListGeneratedVideosByTaskUseCase 测试"""

    def test_list_by_task(self, mock_repo, sample_video):
        """按任务ID查询视频"""
        mock_repo.list_by_generation_task.return_value = [sample_video]
        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)

        result = use_case.execute("task_1")

        assert len(result) == 1
        mock_repo.list_by_generation_task.assert_called_once_with("task_1")

    def test_empty_task_id_raises(self, mock_repo):
        """空任务ID抛出"""
        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)

        with pytest.raises(ValueError, match="generation_task_id 不能为空"):
            use_case.execute("")

    def test_task_id_stripped(self, mock_repo, sample_video):
        """任务ID被 strip"""
        mock_repo.list_by_generation_task.return_value = [sample_video]
        use_case = ListGeneratedVideosByTaskUseCase(mock_repo)

        use_case.execute("  task_1  ")

        mock_repo.list_by_generation_task.assert_called_once_with("task_1")


class TestGetGeneratedVideoDownloadUrlUseCase:
    """GetGeneratedVideoDownloadUrlUseCase 测试"""

    def test_get_url_success(self, mock_repo, sample_video):
        """成功获取下载URL"""
        mock_repo.get.return_value = sample_video
        use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo)

        result = use_case.execute("video_123")

        assert result == sample_video.file_url
        assert "test.mp4" in result

    def test_get_url_nonexistent_returns_none(self, mock_repo):
        """视频不存在返回 None"""
        mock_repo.get.return_value = None
        use_case = GetGeneratedVideoDownloadUrlUseCase(mock_repo)

        result = use_case.execute("nonexistent")

        assert result is None


class TestUpdateVideoReviewStatusUseCase:
    """UpdateVideoReviewStatusUseCase 测试"""

    def test_update_status_pending_review(self, mock_repo, sample_video):
        """更新为待审核"""
        mock_repo.update_review_status.return_value = sample_video
        use_case = UpdateVideoReviewStatusUseCase(mock_repo)

        result = use_case.execute("video_123", "pending_review")

        assert result is not None
        mock_repo.update_review_status.assert_called_once_with("video_123", "pending_review")

    def test_update_status_approved(self, mock_repo, sample_video):
        """更新为审核通过"""
        mock_repo.update_review_status.return_value = sample_video
        use_case = UpdateVideoReviewStatusUseCase(mock_repo)

        use_case.execute("video_123", "approved")

        mock_repo.update_review_status.assert_called_once_with("video_123", "approved")

    def test_update_status_rejected(self, mock_repo, sample_video):
        """更新为审核拒绝"""
        mock_repo.update_review_status.return_value = sample_video
        use_case = UpdateVideoReviewStatusUseCase(mock_repo)

        use_case.execute("video_123", "rejected")

        mock_repo.update_review_status.assert_called_once_with("video_123", "rejected")

    def test_invalid_status_raises(self, mock_repo):
        """无效状态抛出 ValueError"""
        use_case = UpdateVideoReviewStatusUseCase(mock_repo)

        with pytest.raises(ValueError, match="无效的 review_status"):
            use_case.execute("video_123", "invalid_status")

        mock_repo.update_review_status.assert_not_called()

    def test_empty_video_id_raises(self, mock_repo):
        """空视频ID抛出"""
        use_case = UpdateVideoReviewStatusUseCase(mock_repo)

        with pytest.raises(ValueError, match="video_id 不能为空"):
            use_case.execute("", "approved")

    def test_video_id_stripped(self, mock_repo, sample_video):
        """视频ID被 strip"""
        mock_repo.update_review_status.return_value = sample_video
        use_case = UpdateVideoReviewStatusUseCase(mock_repo)

        use_case.execute("  video_123  ", "approved")

        mock_repo.update_review_status.assert_called_once_with("video_123", "approved")


class TestGetVideosByIdsUseCase:
    """GetVideosByIdsUseCase 测试"""

    def test_get_by_ids(self, mock_repo, sample_video):
        """按ID批量获取"""
        video2 = GeneratedVideo.create(
            project_id="proj_1",
            generation_task_id="task_2",
            name="视频2",
            file_url="https://oss.example.com/videos/v2.mp4",
        )
        video2.id = "video_456"
        mock_repo.get_by_ids.return_value = [sample_video, video2]
        use_case = GetVideosByIdsUseCase(mock_repo)

        result = use_case.execute(["video_123", "video_456"])

        assert len(result) == 2
        mock_repo.get_by_ids.assert_called_once_with(["video_123", "video_456"])

    def test_empty_list(self, mock_repo):
        """空ID列表返回空"""
        mock_repo.get_by_ids.return_value = []
        use_case = GetVideosByIdsUseCase(mock_repo)

        result = use_case.execute([])

        assert result == []