xiaoxia-saas/tests/unit/test_ai_avatar_render_service.py

"""AI数字人渲染 Service 单元测试 — #1798.

至少 15 个测试覆盖 Service 层核心逻辑。
"""

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

import pytest

os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")


def _make_mock_db():
    """创建 mock 数据库 session."""
    mock_db = MagicMock()
    mock_db.add = MagicMock()
    mock_db.flush = MagicMock()
    mock_db.commit = MagicMock()
    mock_db.refresh = MagicMock()
    return mock_db


def _make_mock_render_job(
    job_id="render-1",
    user_id="user-1",
    status="pending",
    progress=0,
    output_video_url="",
    output_cover_url="",
    output_duration=0.0,
    error_message="",
    lipsync_job_id="lipsync-1",
    script_id="script-1",
):
    """创建 mock 渲染任务."""
    m = MagicMock()
    m.id = job_id
    m.user_id = user_id
    m.project_id = ""
    m.lipsync_job_id = lipsync_job_id
    m.script_id = script_id
    m.b_roll_segments = []
    m.title_config = {}
    m.cover_config = {}
    m.status = status
    m.progress = progress
    m.output_video_url = output_video_url
    m.output_cover_url = output_cover_url
    m.output_duration = output_duration
    m.error_message = error_message
    m.submitted_at = None
    m.started_at = None
    m.completed_at = None
    m.created_at = None
    m.updated_at = None
    return m


def _make_mock_lipsync_job(
    job_id="lipsync-1",
    user_id="user-1",
    status="completed",
    output_video_url="https://output.mp4",
    output_duration=30.0,
):
    """创建 mock 对口型任务."""
    m = MagicMock()
    m.id = job_id
    m.user_id = user_id
    m.status = status
    m.output_video_url = output_video_url
    m.output_duration = output_duration
    return m


def _make_mock_script(script_id="script-1", user_id="user-1"):
    """创建 mock 文案."""
    m = MagicMock()
    m.id = script_id
    m.user_id = user_id
    m.title = "测试文案"
    return m


class TestSchemaValidation:
    """Schema 验证测试."""

    def test_valid_broll_segment(self):
        from app.schemas.ai_avatar_render import BRollSegment

        seg = BRollSegment(
            script_segment_index=0,
            asset_url="https://example.com/broll.mp4",
            mode="fullscreen",
            start_time=5.0,
            end_time=10.0,
        )
        assert seg.mode == "fullscreen"
        assert seg.start_time == 5.0

    def test_invalid_mode(self):
        from app.schemas.ai_avatar_render import BRollSegment

        with pytest.raises(ValueError, match="fullscreen 或 pip"):
            BRollSegment(
                script_segment_index=0,
                asset_url="https://example.com/broll.mp4",
                mode="invalid",
                start_time=5.0,
                end_time=10.0,
            )

    def test_end_time_must_exceed_start_time(self):
        from app.schemas.ai_avatar_render import BRollSegment

        with pytest.raises(ValueError, match="end_time 必须大于 start_time"):
            BRollSegment(
                script_segment_index=0,
                asset_url="https://example.com/broll.mp4",
                mode="fullscreen",
                start_time=10.0,
                end_time=5.0,
            )

    def test_asset_url_must_be_http(self):
        from app.schemas.ai_avatar_render import BRollSegment

        with pytest.raises(ValueError, match="HTTP"):
            BRollSegment(
                script_segment_index=0,
                asset_url="ftp://example.com/broll.mp4",
                mode="fullscreen",
                start_time=5.0,
                end_time=10.0,
            )

    def test_asset_url_empty(self):
        from app.schemas.ai_avatar_render import BRollSegment

        with pytest.raises(ValueError, match="不能为空"):
            BRollSegment(
                script_segment_index=0,
                asset_url="  ",
                mode="fullscreen",
                start_time=5.0,
                end_time=10.0,
            )

    def test_create_request_valid(self):
        from app.schemas.ai_avatar_render import BRollSegment, CreateAiAvatarRenderRequest

        req = CreateAiAvatarRenderRequest(
            lipsync_job_id="lipsync-1",
            script_id="script-1",
            b_roll_segments=[
                BRollSegment(
                    script_segment_index=0,
                    asset_url="https://example.com/broll.mp4",
                    mode="pip",
                    start_time=5.0,
                    end_time=10.0,
                )
            ],
        )
        assert req.lipsync_job_id == "lipsync-1"
        assert len(req.b_roll_segments) == 1

    def test_create_request_empty_lipsync_job_id(self):
        from app.schemas.ai_avatar_render import CreateAiAvatarRenderRequest

        with pytest.raises(ValueError, match="lipsync_job_id 不能为空"):
            CreateAiAvatarRenderRequest(
                lipsync_job_id="  ",
                script_id="script-1",
            )

    def test_create_request_empty_script_id(self):
        from app.schemas.ai_avatar_render import CreateAiAvatarRenderRequest

        with pytest.raises(ValueError, match="script_id 不能为空"):
            CreateAiAvatarRenderRequest(
                lipsync_job_id="lipsync-1",
                script_id="  ",
            )


class TestAiAvatarRenderService:
    """Service 层单元测试（纯 mock，不依赖数据库）."""

    def test_create_job_success(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        # 模拟 query 链式调用
        mock_query = MagicMock()

        # 第一次 query: LipsyncJobModel
        mock_lipsync_filter = MagicMock()
        mock_lipsync_filter.first.return_value = _make_mock_lipsync_job()
        mock_lipsync_query = MagicMock()
        mock_lipsync_query.filter.return_value = mock_lipsync_filter

        # 第二次 query: ScriptModel
        mock_script_filter = MagicMock()
        mock_script_filter.first.return_value = _make_mock_script()
        mock_script_query = MagicMock()
        mock_script_query.filter.return_value = mock_script_filter

        mock_db.query.side_effect = [mock_lipsync_query, mock_script_query]

        svc = AiAvatarRenderService(mock_db)
        job = svc.create_render_job(
            user_id="user-1",
            lipsync_job_id="lipsync-1",
            script_id="script-1",
            b_roll_segments=[],
            title_config={},
            cover_config={},
        )
        assert job.status == "pending"
        mock_db.add.assert_called_once()
        mock_db.commit.assert_called_once()

    def test_create_job_lipsync_not_found(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_filter = MagicMock()
        mock_filter.first.return_value = None
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        with pytest.raises(AiAvatarRenderError, match="对口型任务不存在"):
            svc.create_render_job(
                user_id="user-1",
                lipsync_job_id="nonexistent",
                script_id="script-1",
                b_roll_segments=[],
                title_config={},
                cover_config={},
            )

    def test_create_job_lipsync_not_completed(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_lipsync_job = _make_mock_lipsync_job(status="processing")

        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_lipsync_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        with pytest.raises(AiAvatarRenderError, match="仅 completed 状态可渲染"):
            svc.create_render_job(
                user_id="user-1",
                lipsync_job_id="lipsync-1",
                script_id="script-1",
                b_roll_segments=[],
                title_config={},
                cover_config={},
            )

    def test_create_job_lipsync_no_output(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_lipsync_job = _make_mock_lipsync_job(status="completed", output_video_url="")

        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_lipsync_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        with pytest.raises(AiAvatarRenderError, match="输出视频 URL 为空"):
            svc.create_render_job(
                user_id="user-1",
                lipsync_job_id="lipsync-1",
                script_id="script-1",
                b_roll_segments=[],
                title_config={},
                cover_config={},
            )

    def test_create_job_script_not_found(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_lipsync_query = MagicMock()
        mock_lipsync_filter = MagicMock()
        mock_lipsync_filter.first.return_value = _make_mock_lipsync_job()
        mock_lipsync_query.filter.return_value = mock_lipsync_filter

        mock_script_query = MagicMock()
        mock_script_filter = MagicMock()
        mock_script_filter.first.return_value = None
        mock_script_query.filter.return_value = mock_script_filter

        mock_db.query.side_effect = [mock_lipsync_query, mock_script_query]

        svc = AiAvatarRenderService(mock_db)
        with pytest.raises(AiAvatarRenderError, match="文案不存在或无权访问"):
            svc.create_render_job(
                user_id="user-1",
                lipsync_job_id="lipsync-1",
                script_id="nonexistent",
                b_roll_segments=[],
                title_config={},
                cover_config={},
            )

    def test_get_render_job_found(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_job = _make_mock_render_job()
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.get_render_job("render-1", "user-1")
        assert result is mock_job

    def test_get_render_job_not_found(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_filter = MagicMock()
        mock_filter.first.return_value = None
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.get_render_job("nonexistent", "user-1")
        assert result is None

    def test_list_render_jobs(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_jobs = [_make_mock_render_job(f"render-{i}") for i in range(3)]

        mock_query = MagicMock()
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 3
        mock_query.order_by.return_value = mock_query
        mock_query.offset.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = mock_jobs
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        items, total = svc.list_render_jobs(user_id="user-1")
        assert total == 3
        assert len(items) == 3

    def test_list_render_jobs_with_project_filter(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_query
        mock_query.count.return_value = 1
        mock_query.order_by.return_value = mock_query
        mock_query.offset.return_value = mock_query
        mock_query.limit.return_value = mock_query
        mock_query.all.return_value = [_make_mock_render_job()]
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        items, total = svc.list_render_jobs(user_id="user-1", project_id="proj-1")
        assert total == 1
        # filter should be called for user_id and project_id
        assert mock_query.filter.call_count >= 2

    def test_cancel_render_job_success(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_job = _make_mock_render_job(status="pending")
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.cancel_render_job("render-1", "user-1")
        assert result is mock_job
        assert mock_job.status == "cancelled"

    def test_cancel_render_job_not_pending(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_job = _make_mock_render_job(status="completed")
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.cancel_render_job("render-1", "user-1")
        # 非 pending 状态不可取消，状态不变
        assert result.status == "completed"

    def test_cancel_render_job_not_found(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_filter = MagicMock()
        mock_filter.first.return_value = None
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.cancel_render_job("nonexistent", "user-1")
        assert result is None

    def test_retry_render_job_success(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_job = _make_mock_render_job(status="failed", error_message="渲染失败")
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.retry_render_job("render-1", "user-1")
        assert result.status == "pending"
        assert result.progress == 0
        assert result.error_message == ""

    def test_retry_render_job_not_failed(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_job = _make_mock_render_job(status="completed")
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.retry_render_job("render-1", "user-1")
        assert result is None

    def test_retry_render_job_not_found(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_filter = MagicMock()
        mock_filter.first.return_value = None
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        result = svc.retry_render_job("nonexistent", "user-1")
        assert result is None

    def test_execute_render_job_not_found(self):
        """execute_render 在任务不存在时应静默返回."""
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_filter = MagicMock()
        mock_filter.first.return_value = None
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        # 不应抛异常
        svc.execute_render("nonexistent")

    def test_execute_render_cancelled_job(self):
        """execute_render 在任务已取消时应静默返回."""
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_db = _make_mock_db()
        mock_job = _make_mock_render_job(status="cancelled")
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query = MagicMock()
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = AiAvatarRenderService(mock_db)
        svc.execute_render("render-1")
        # 不应执行渲染逻辑
        mock_db.commit.assert_not_called()

    def test_error_exception_has_code(self):
        from app.services.ai_avatar_render_service import AiAvatarRenderError

        err = AiAvatarRenderError("测试错误", code="TestCode")
        assert err.code == "TestCode"
        assert str(err) == "测试错误"