xiaoxia-saas/tests/unit/test_generated_video_creation_logic.py

"""测试：渲染完成后 generated_videos 写表判断逻辑.

验证：project_id 为空时，只要有 generation_task_id 就应该写入 generated_videos。
"""

from __future__ import annotations

from datetime import datetime, timezone
from unittest.mock import MagicMock, patch

import pytest
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
from packages.domain import GenerationTask, GenerationTaskStatus


def _make_task(**kwargs) -> GenerationTask:
    defaults = dict(
        project_id="proj-1",
        asset_library_id="lib-1",
        created_by_user_id="user-1",
        status=GenerationTaskStatus.PENDING,
        progress=0.0,
        source_edit_plan_id="plan-1",
    )
    defaults.update(kwargs)
    return GenerationTask(id="task-" + kwargs.get("project_id", "t1")[:8], **defaults)


class TestCleanupStaleRunning:
    """复用已有的 repo 测试基础设施，验证写表判断逻辑。"""

    def _setup(self):
        engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(engine)
        session = sessionmaker(bind=engine)()
        return SQLAlchemyGenerationTaskRepository(session), session

    def test_generation_task_without_project_id_can_be_created(self):
        """验证：project_id 为空的 generation_task 可以正常创建（模拟一键生成场景）。"""
        repo, session = self._setup()
        try:
            task = _make_task(
                project_id="",
                status=GenerationTaskStatus.COMPLETED,
                progress=100.0,
                result_count=1,
                source_edit_plan_id="plan-empty-proj",
            )
            task.id = "task-empty-proj"
            created = repo.create(task)
            assert created.project_id == ""

            fetched = repo.get("task-empty-proj")
            assert fetched is not None
            assert fetched.project_id == ""
            assert fetched.status.value == "completed"
            assert fetched.progress == 100.0
        finally:
            session.close()

    def test_generation_task_with_project_id_can_be_created(self):
        """验证：project_id 有值的 generation_task 正常（回归测试）。"""
        repo, session = self._setup()
        try:
            task = _make_task(
                project_id="proj-123",
                status=GenerationTaskStatus.COMPLETED,
                progress=100.0,
                result_count=1,
            )
            task.id = "task-with-proj"
            created = repo.create(task)
            assert created.project_id == "proj-123"

            fetched = repo.get("task-with-proj")
            assert fetched is not None
            assert fetched.project_id == "proj-123"
        finally:
            session.close()


class TestVideoCreationLogic:
    """验证写表判断逻辑（纯逻辑测试，不依赖 DB）。"""

    def test_should_create_video_with_generation_task_id_and_empty_project(self):
        """有 generation_task_id、无 project_id → 应该创建视频（修复点）。"""
        generation_task_id = "task-123"
        project_id = ""
        # 修复后：只要有 generation_task_id 就创建
        assert bool(generation_task_id) is True
        # 之前的错误逻辑：if generation_task_id and project_id → False
        assert bool(generation_task_id and project_id) is False
        # 修复后的逻辑：if generation_task_id → True
        assert bool(generation_task_id) is True

    def test_should_create_video_with_both_ids(self):
        """有 generation_task_id、有 project_id → 应该创建视频（回归）。"""
        generation_task_id = "task-123"
        project_id = "proj-456"
        assert bool(generation_task_id) is True
        assert bool(generation_task_id and project_id) is True

    def test_should_not_create_video_without_generation_task_id(self):
        """无 generation_task_id → 不创建视频（边界）。"""
        generation_task_id = ""
        project_id = "proj-456"
        assert bool(generation_task_id) is False


class TestVideoNameParameter:
    """验证成片库视频名称逻辑：用户设置标题时用标题，没设置时用默认命名。"""

    def test_name_from_user_title(self):
        """用户设置了标题 → 用标题作为视频名称。"""
        # 验证函数签名包含 name 参数
        import inspect

        from video_processing.dedup_helpers import create_video_record_and_dedup

        sig = inspect.signature(create_video_record_and_dedup)
        assert "name" in sig.parameters, "create_video_record_and_dedup 应支持 name 参数"

    def test_generated_video_requires_name(self):
        """GeneratedVideo.create 要求 name 非空。"""
        from packages.domain import GeneratedVideo

        with pytest.raises(ValueError, match="name cannot be empty"):
            GeneratedVideo.create(
                project_id="proj-1",
                user_id="user-1",
                generation_task_id="task-1",
                name="",
                file_url="https://example.com/test.mp4",
            )

    def test_name_fallback_when_empty(self):
        """name 为空或纯空格时，调用方应 fallback 到默认命名。"""
        # 模拟 _finalize_render_success 中的逻辑
        title_text_empty = ""
        title_text_spaces = "   "
        generation_task_id = "task-abcdef123456"

        # 空标题 → fallback
        video_name_1 = title_text_empty.strip() or f"generated-{generation_task_id[:8]}.mp4"
        assert video_name_1 == f"generated-{generation_task_id[:8]}.mp4"

        # 纯空格 → fallback
        video_name_2 = title_text_spaces.strip() or f"generated-{generation_task_id[:8]}.mp4"
        assert video_name_2 == f"generated-{generation_task_id[:8]}.mp4"

        # 有标题 → 用标题
        title_text = "我的旅行vlog"
        video_name_3 = title_text.strip() or f"generated-{generation_task_id[:8]}.mp4"
        assert video_name_3 == "我的旅行vlog"


class TestUserIdFilter:
    """验证成片库按 user_id 过滤的核心逻辑。"""

    def _setup_repo(self):
        from sqlalchemy import create_engine
        from sqlalchemy.orm import sessionmaker

        from packages.adapters.sqlalchemy_impl.generated_video_repository import (
            SQLAlchemyGeneratedVideoRepository,
        )
        from packages.adapters.sqlalchemy_impl.models import Base

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

    def test_list_paginated_filters_by_user_id(self):
        """list_paginated 传入 user_id 时只返回该用户的视频。"""
        from packages.domain import GeneratedVideo

        repo, session = self._setup_repo()
        try:
            # 用户A的2个视频
            for i in range(2):
                v = GeneratedVideo.create(
                    project_id=f"proj-a-{i}",
                    user_id="user-a",
                    generation_task_id=f"task-a-{i}",
                    name=f"video-a-{i}.mp4",
                    file_url=f"https://oss.example.com/a-{i}.mp4",
                )
                repo.create(v)

            # 用户B的3个视频
            for i in range(3):
                v = GeneratedVideo.create(
                    project_id=f"proj-b-{i}",
                    user_id="user-b",
                    generation_task_id=f"task-b-{i}",
                    name=f"video-b-{i}.mp4",
                    file_url=f"https://oss.example.com/b-{i}.mp4",
                )
                repo.create(v)

            # 查用户A → 2条
            items, total = repo.list_paginated(user_id="user-a", page=1, page_size=10)
            assert total == 2
            assert len(items) == 2
            assert all(it.user_id == "user-a" for it in items)

            # 查用户B → 3条
            items, total = repo.list_paginated(user_id="user-b", page=1, page_size=10)
            assert total == 3
            assert len(items) == 3
            assert all(it.user_id == "user-b" for it in items)
        finally:
            session.close()

    def test_list_paginated_user_id_plus_project_id(self):
        """同时传 user_id 和 project_id 时两个条件同时过滤。"""
        from packages.domain import GeneratedVideo

        repo, session = self._setup_repo()
        try:
            # 用户A的proj-1视频
            v1 = GeneratedVideo.create(
                project_id="proj-1",
                user_id="user-a",
                generation_task_id="task-1",
                name="v1.mp4",
                file_url="https://oss.example.com/v1.mp4",
            )
            repo.create(v1)

            # 用户B的proj-1视频（不同用户同项目）
            v2 = GeneratedVideo.create(
                project_id="proj-1",
                user_id="user-b",
                generation_task_id="task-2",
                name="v2.mp4",
                file_url="https://oss.example.com/v2.mp4",
            )
            repo.create(v2)

            # 用户A + proj-1 → 只有1条
            items, total = repo.list_paginated(user_id="user-a", project_id="proj-1", page=1, page_size=10)
            assert total == 1
            assert items[0].user_id == "user-a"
            assert items[0].generation_task_id == "task-1"
        finally:
            session.close()

    def test_generated_video_has_user_id_field(self):
        """GeneratedVideo domain 对象有 user_id 字段。"""
        from packages.domain import GeneratedVideo

        v = GeneratedVideo.create(
            project_id="proj-1",
            user_id="user-123",
            generation_task_id="task-1",
            name="test.mp4",
            file_url="https://example.com/test.mp4",
        )
        assert v.user_id == "user-123"

    def test_created_video_persists_user_id(self):
        """创建视频后 user_id 能正确持久化和读取。"""
        from packages.domain import GeneratedVideo

        repo, session = self._setup_repo()
        try:
            v = GeneratedVideo.create(
                project_id="proj-1",
                user_id="user-persist-test",
                generation_task_id="task-persist",
                name="persist.mp4",
                file_url="https://example.com/persist.mp4",
            )
            repo.create(v)

            fetched = repo.get(v.id)
            assert fetched is not None
            assert fetched.user_id == "user-persist-test"
        finally:
            session.close()