xiaoxia-saas/tests/unit/test_generation_preview.py

"""预览生成（Phase 1）单元测试.

覆盖：
- 领域模型 is_preview 字段
- CreateGenerationTaskCommand is_preview 字段
- UseCase 传递 is_preview
- Schema 校验（CreatePreviewGenerationTaskRequest / PreviewGenerationTaskResponse）
- 仓储层 _to_domain 兼容旧数据（getattr + 默认值）
- 仓储层 create/update 保留 is_preview 字段
- 状态流转验证
"""

from __future__ import annotations

import os
import sys
from datetime import datetime, timezone
from pathlib import Path
from unittest.mock import MagicMock

import pytest

# 设置必要环境变量（必须在导入 app 模块之前）
os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret-key-for-testing")
os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")

# 确保 app 模块可导入
sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "api"))

from packages.application.generation_tasks import (
    CreateGenerationTaskCommand,
    CreateGenerationTaskUseCase,
)
from packages.domain.generation_task import GenerationTask, GenerationTaskStatus

# ── 领域模型测试 ──────────────────────────────────────────────────────────────


class TestGenerationTaskIsPreviewField:
    """GenerationTask 领域模型 is_preview 字段测试"""

    def test_is_preview_default_false(self):
        """默认 is_preview=False"""
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
        )
        assert task.is_preview is False

    def test_is_preview_true_when_specified(self):
        """显式指定 is_preview=True"""
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
            is_preview=True,
        )
        assert task.is_preview is True

    def test_is_preview_false_when_explicit_false(self):
        """显式指定 is_preview=False"""
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
            is_preview=False,
        )
        assert task.is_preview is False

    def test_is_preview_with_template_mode(self):
        """模板模式下 is_preview 正常工作"""
        task = GenerationTask.create(
            project_id="",
            asset_library_id="",
            template_id="tmpl1",
            asset_ids=["a1", "a2"],
            is_preview=True,
        )
        assert task.is_preview is True
        assert task.template_id == "tmpl1"
        assert task.asset_ids == ["a1", "a2"]

    def test_is_preview_preserved_in_dataclass(self):
        """is_preview 是 dataclass 字段，可以被赋值"""
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
        )
        task.is_preview = True
        assert task.is_preview is True


# ── UseCase 测试 ────────────────────────────────────────────────────────────


class TestCreateGenerationTaskUseCaseIsPreview:
    """CreateGenerationTaskUseCase 中 is_preview 传递测试"""

    def test_command_default_is_preview_false(self):
        """CreateGenerationTaskCommand 默认 is_preview=False"""
        cmd = CreateGenerationTaskCommand(
            project_id="proj1",
            asset_library_id="lib1",
        )
        assert cmd.is_preview is False

    def test_command_is_preview_true(self):
        """CreateGenerationTaskCommand 设置 is_preview=True"""
        cmd = CreateGenerationTaskCommand(
            project_id="proj1",
            asset_library_id="lib1",
            is_preview=True,
        )
        assert cmd.is_preview is True

    def test_use_case_passes_is_preview_to_task(self):
        """UseCase 将 is_preview 传递给 GenerationTask"""
        mock_repo = MagicMock()
        mock_repo.create.side_effect = lambda t: t
        use_case = CreateGenerationTaskUseCase(mock_repo)

        command = CreateGenerationTaskCommand(
            project_id="proj1",
            asset_library_id="lib1",
            is_preview=True,
        )
        result = use_case.execute(command)

        assert result.is_preview is True
        mock_repo.create.assert_called_once()

    def test_use_case_default_is_preview_false(self):
        """UseCase 默认 is_preview=False"""
        mock_repo = MagicMock()
        mock_repo.create.side_effect = lambda t: t
        use_case = CreateGenerationTaskUseCase(mock_repo)

        command = CreateGenerationTaskCommand(
            project_id="proj1",
            asset_library_id="lib1",
        )
        result = use_case.execute(command)

        assert result.is_preview is False


# ── Schema 测试 ──────────────────────────────────────────────────────────────


class TestCreatePreviewGenerationTaskRequest:
    """CreatePreviewGenerationTaskRequest schema 校验测试"""

    def test_valid_request_with_asset_ids(self):
        """有效请求：template_id + asset_ids"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            asset_ids=["asset_1", "asset_2"],
        )
        assert req.template_id == "tmpl_123"
        assert req.asset_ids == ["asset_1", "asset_2"]
        assert req.title_ids == []
        assert req.voice_ids == []
        assert req.video_title == ""
        assert req.duration == 0.0
        assert req.bgm_config == {}

    def test_valid_request_with_title_ids_only(self):
        """有效请求：template_id + title_ids（替代 asset_ids）"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            title_ids=["title_1"],
        )
        assert req.template_id == "tmpl_123"
        assert req.title_ids == ["title_1"]

    def test_valid_request_with_voice_ids_only(self):
        """有效请求：template_id + voice_ids"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            voice_ids=["voice_1"],
        )
        assert req.voice_ids == ["voice_1"]

    def test_missing_template_id_raises(self):
        """缺少 template_id 报错"""
        from pydantic import ValidationError

        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        with pytest.raises(ValidationError, match="template_id"):
            CreatePreviewGenerationTaskRequest(
                template_id="",
                asset_ids=["asset_1"],
            )

    def test_missing_template_id_not_provided_raises(self):
        """完全不提供 template_id 报错"""
        from pydantic import ValidationError

        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        with pytest.raises(ValidationError):
            CreatePreviewGenerationTaskRequest(
                asset_ids=["asset_1"],
            )

    def test_empty_asset_ids_raises(self):
        """asset_ids/title_ids/voice_ids 全空报错"""
        from pydantic import ValidationError

        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        with pytest.raises(ValidationError, match="asset_ids"):
            CreatePreviewGenerationTaskRequest(
                template_id="tmpl_123",
                asset_ids=[],
                title_ids=[],
                voice_ids=[],
            )

    def test_request_with_all_fields(self):
        """所有字段都设置的请求"""
        from app.schemas.generation_task import CreatePreviewGenerationTaskRequest

        req = CreatePreviewGenerationTaskRequest(
            template_id="tmpl_123",
            asset_ids=["a1", "a2"],
            title_ids=["t1"],
            voice_ids=["v1"],
            video_title="测试预览视频",
            duration=30.0,
            video_ratio="9:16",
            bgm_config={"enabled": True, "volume": 0.5},
        )
        assert req.video_title == "测试预览视频"
        assert req.duration == 30.0
        assert req.video_ratio == "9:16"
        assert req.bgm_config["enabled"] is True
        assert req.bgm_config["volume"] == 0.5


class TestPreviewGenerationTaskResponse:
    """PreviewGenerationTaskResponse schema 测试"""

    def test_pending_state_response(self):
        """pending 状态的响应"""
        from app.schemas.generation_task import PreviewGenerationTaskResponse

        resp = PreviewGenerationTaskResponse(
            task_id="task_123",
            status="pending",
            progress=0.0,
        )
        assert resp.task_id == "task_123"
        assert resp.status == "pending"
        assert resp.progress == 0.0
        assert resp.is_preview is True
        assert resp.resolution == ""
        assert resp.video_url == ""
        assert resp.duration == 0.0
        assert resp.file_size == 0
        assert resp.clip_count == 0
        assert resp.error_message == ""

    def test_completed_state_response(self):
        """completed 状态的响应"""
        from datetime import datetime, timezone

        from app.schemas.generation_task import PreviewGenerationTaskResponse

        now = datetime.now(timezone.utc)
        resp = PreviewGenerationTaskResponse(
            task_id="task_123",
            status="completed",
            progress=100.0,
            is_preview=True,
            resolution="854x480",
            video_url="https://example.com/preview.mp4",
            duration=30.5,
            file_size=5_000_000,
            clip_count=5,
            transition_count=4,
            material_usage={"videos": 5, "images": 2},
            created_at=now,
            started_at=now,
            finished_at=now,
            generate_duration=12.5,
        )
        assert resp.status == "completed"
        assert resp.progress == 100.0
        assert resp.resolution == "854x480"
        assert resp.video_url == "https://example.com/preview.mp4"
        assert resp.duration == 30.5
        assert resp.file_size == 5_000_000
        assert resp.clip_count == 5
        assert resp.transition_count == 4
        assert resp.generate_duration == 12.5

    def test_failed_state_response(self):
        """failed 状态的响应"""
        from app.schemas.generation_task import PreviewGenerationTaskResponse

        resp = PreviewGenerationTaskResponse(
            task_id="task_123",
            status="failed",
            progress=30.0,
            error_message="渲染失败：素材格式不支持",
        )
        assert resp.status == "failed"
        assert resp.error_message == "渲染失败：素材格式不支持"
        assert resp.video_url == ""


# ── 仓储层兼容测试 ──────────────────────────────────────────────────────────


class TestRepositoryIsPreviewCompatibility:
    """仓储层 is_preview 向后兼容测试"""

    def test_to_domain_with_is_preview_true(self):
        """新数据 is_preview=True 时正确映射"""
        mock_model = MagicMock()
        mock_model.id = "task_123"
        mock_model.project_id = "proj1"
        mock_model.strategy_id = ""
        mock_model.asset_library_id = "lib1"
        mock_model.voice_library_id = ""
        mock_model.template_id = "tmpl1"
        mock_model.asset_ids = ["a1"]
        mock_model.title_ids = []
        mock_model.voice_ids = []
        mock_model.status = "pending"
        mock_model.progress = 0.0
        mock_model.result_count = 0
        mock_model.error_message = ""
        mock_model.error_info = {}
        mock_model.retry_count = 0
        mock_model.auto_retry_enabled = False
        mock_model.auto_retry_max = 0
        mock_model.started_at = None
        mock_model.completed_at = None
        mock_model.created_by_user_id = "user1"
        mock_model.source_edit_plan_id = None
        mock_model.asset_select_mode = ""
        mock_model.batch_id = ""
        mock_model.video_title = ""
        mock_model.resolution = "854x480"
        mock_model.bgm_config = {}
        mock_model.is_preview = True
        mock_model.logs = "[]"
        mock_model.created_at = datetime.now(timezone.utc)
        mock_model.updated_at = datetime.now(timezone.utc)

        from packages.adapters.sqlalchemy_impl.generation_task_repository import _to_domain

        result = _to_domain(mock_model)
        assert result.is_preview is True
        assert result.resolution == "854x480"

    def test_to_domain_with_is_preview_false(self):
        """is_preview=False 时正确映射"""
        mock_model = MagicMock()
        mock_model.id = "task_123"
        mock_model.project_id = "proj1"
        mock_model.strategy_id = ""
        mock_model.asset_library_id = "lib1"
        mock_model.voice_library_id = ""
        mock_model.template_id = ""
        mock_model.asset_ids = []
        mock_model.title_ids = []
        mock_model.voice_ids = []
        mock_model.status = "pending"
        mock_model.progress = 0.0
        mock_model.result_count = 0
        mock_model.error_message = ""
        mock_model.error_info = {}
        mock_model.retry_count = 0
        mock_model.auto_retry_enabled = False
        mock_model.auto_retry_max = 0
        mock_model.started_at = None
        mock_model.completed_at = None
        mock_model.created_by_user_id = "user1"
        mock_model.source_edit_plan_id = None
        mock_model.asset_select_mode = ""
        mock_model.batch_id = ""
        mock_model.video_title = ""
        mock_model.resolution = ""
        mock_model.bgm_config = {}
        mock_model.is_preview = False
        mock_model.logs = "[]"
        mock_model.created_at = datetime.now(timezone.utc)
        mock_model.updated_at = datetime.now(timezone.utc)

        from packages.adapters.sqlalchemy_impl.generation_task_repository import _to_domain

        result = _to_domain(mock_model)
        assert result.is_preview is False

    def test_repository_create_includes_is_preview(self):
        """repository.create() 包含 is_preview 字段"""
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
            SQLAlchemyGenerationTaskRepository,
        )
        from packages.adapters.sqlalchemy_impl.models import Base, GenerationTaskModel

        # 使用内存数据库
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        SessionLocal = sessionmaker(bind=engine)
        session = SessionLocal()

        try:
            repo = SQLAlchemyGenerationTaskRepository(session)
            task = GenerationTask.create(
                project_id="proj1",
                asset_library_id="lib1",
                is_preview=True,
                resolution="854x480",
            )
            repo.create(task)

            # 直接查 model 验证
            model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task.id).first()
            assert model is not None
            assert model.is_preview is True
            assert model.resolution == "854x480"
        finally:
            session.close()

    def test_repository_update_preserves_is_preview(self):
        """repository.update() 保留 is_preview 字段"""
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from packages.adapters.sqlalchemy_impl.generation_task_repository import (
            SQLAlchemyGenerationTaskRepository,
        )
        from packages.adapters.sqlalchemy_impl.models import Base, GenerationTaskModel

        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        SessionLocal = sessionmaker(bind=engine)
        session = SessionLocal()

        try:
            repo = SQLAlchemyGenerationTaskRepository(session)
            task = GenerationTask.create(
                project_id="proj1",
                asset_library_id="lib1",
                is_preview=True,
            )
            repo.create(task)

            # 更新任务状态
            task.mark_processing()
            repo.update(task)

            model = session.query(GenerationTaskModel).filter(GenerationTaskModel.id == task.id).first()
            assert model is not None
            assert model.is_preview is True  # is_preview 应该保持不变
            assert model.status == "running"
        finally:
            session.close()


# ── 状态流转测试 ────────────────────────────────────────────────────────────


class TestPreviewTaskStatusFlow:
    """预览任务状态流转测试"""

    def test_pending_to_running(self):
        """pending → running"""
        task = GenerationTask.create(
            project_id="",
            asset_library_id="",
            template_id="tmpl1",
            asset_ids=["a1"],
            is_preview=True,
        )
        assert task.status == GenerationTaskStatus.PENDING
        assert task.is_preview is True

        task.mark_processing()
        assert task.status == GenerationTaskStatus.RUNNING
        assert task.started_at is not None

    def test_running_to_completed(self):
        """running → completed"""
        task = GenerationTask.create(
            project_id="",
            asset_library_id="",
            template_id="tmpl1",
            asset_ids=["a1"],
            is_preview=True,
        )
        task.mark_processing()
        task.mark_completed()

        assert task.status == GenerationTaskStatus.COMPLETED
        assert task.progress == 100.0
        assert task.completed_at is not None
        assert task.is_preview is True

    def test_running_to_failed(self):
        """running → failed"""
        task = GenerationTask.create(
            project_id="",
            asset_library_id="",
            template_id="tmpl1",
            asset_ids=["a1"],
            is_preview=True,
        )
        task.mark_processing()
        task.mark_failed("渲染错误")

        assert task.status == GenerationTaskStatus.FAILED
        assert task.error_message == "渲染错误"
        assert task.is_preview is True

    def test_pending_to_cancelled(self):
        """pending → cancelled"""
        task = GenerationTask.create(
            project_id="",
            asset_library_id="",
            template_id="tmpl1",
            asset_ids=["a1"],
            is_preview=True,
        )
        task.mark_cancelled()

        assert task.status == GenerationTaskStatus.CANCELLED
        assert task.is_preview is True

    def test_preview_resolution_is_480p(self):
        """预览任务分辨率为 854x480"""
        task = GenerationTask.create(
            project_id="",
            asset_library_id="",
            template_id="tmpl1",
            asset_ids=["a1"],
            resolution="854x480",
            is_preview=True,
        )
        assert task.resolution == "854x480"
        assert task.is_preview is True

    def test_non_preview_task_default_false(self):
        """非预览任务 is_preview 默认 False"""
        task = GenerationTask.create(
            project_id="proj1",
            asset_library_id="lib1",
        )
        assert task.is_preview is False