xiaoxia-saas/tests/unit/test_ai_avatar_render_routes.py

from datetime import datetime, timezone

"""AI数字人渲染 API 路由测试 — #1798.

至少 10 个测试覆盖路由层逻辑。
"""

import os
from unittest.mock import MagicMock, patch

import pytest

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


def _make_mock_user(user_id="user-1"):
    """创建 mock 认证用户."""
    user = MagicMock()
    user.id = user_id
    return user


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="",
):
    """创建 mock 渲染任务."""
    m = MagicMock()
    m.id = job_id
    m.user_id = user_id
    m.project_id = ""
    m.lipsync_job_id = "lipsync-1"
    m.script_id = "script-1"
    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 = datetime(2026, 1, 1, tzinfo=timezone.utc)
    m.updated_at = datetime(2026, 1, 1, tzinfo=timezone.utc)
    return m


class TestRenderRoutes:
    """路由层测试（通过 mock service 测试路由逻辑）."""

    def _get_client(self):
        """获取测试客户端."""
        from app.main import app
        from fastapi.testclient import TestClient

        return TestClient(app)

    def test_create_render_job_success(self):
        from app.api.routes.ai_avatar_render import router
        from app.schemas.ai_avatar_render import AiAvatarRenderJobResponse
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_job = _make_mock_render_job()
        mock_service.create_render_job.return_value = mock_job

        # 直接测试路由函数
        from app.api.routes.ai_avatar_render import create_render_job

        mock_user = _make_mock_user()
        body = MagicMock()
        body.lipsync_job_id = "lipsync-1"
        body.script_id = "script-1"
        body.b_roll_segments = []
        body.title_config = {}
        body.cover_config = {}
        body.project_id = ""

        result = create_render_job(
            body=body,
            current_user=mock_user,
            svc=mock_service,
        )
        assert result.id == "render-1"
        mock_service.create_render_job.assert_called_once()

    def test_create_render_job_lipsync_not_found(self):
        from app.api.routes.ai_avatar_render import create_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_service.create_render_job.side_effect = AiAvatarRenderError("对口型任务不存在", code="LipsyncJobNotFound")

        mock_user = _make_mock_user()
        body = MagicMock()
        body.lipsync_job_id = "nonexistent"
        body.script_id = "script-1"
        body.b_roll_segments = []
        body.title_config = {}
        body.cover_config = {}
        body.project_id = ""

        with pytest.raises(HTTPException) as exc_info:
            create_render_job(body=body, current_user=mock_user, svc=mock_service)
        assert exc_info.value.status_code == 404

    def test_create_render_job_lipsync_not_completed(self):
        from app.api.routes.ai_avatar_render import create_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderError, AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_service.create_render_job.side_effect = AiAvatarRenderError(
            "对口型任务状态为 processing", code="LipsyncJobNotCompleted"
        )

        mock_user = _make_mock_user()
        body = MagicMock()
        body.lipsync_job_id = "lipsync-1"
        body.script_id = "script-1"
        body.b_roll_segments = []
        body.title_config = {}
        body.cover_config = {}
        body.project_id = ""

        with pytest.raises(HTTPException) as exc_info:
            create_render_job(body=body, current_user=mock_user, svc=mock_service)
        assert exc_info.value.status_code == 400

    def test_get_render_job_success(self):
        from app.api.routes.ai_avatar_render import get_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_job = _make_mock_render_job()
        mock_service.get_render_job.return_value = mock_job

        result = get_render_job(job_id="render-1", current_user=_make_mock_user(), svc=mock_service)
        assert result.id == "render-1"

    def test_get_render_job_not_found(self):
        from app.api.routes.ai_avatar_render import get_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_service.get_render_job.return_value = None

        with pytest.raises(HTTPException) as exc_info:
            get_render_job(job_id="nonexistent", current_user=_make_mock_user(), svc=mock_service)
        assert exc_info.value.status_code == 404

    def test_list_render_jobs(self):
        from app.api.routes.ai_avatar_render import list_render_jobs
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_jobs = [_make_mock_render_job(f"render-{i}") for i in range(3)]
        mock_service.list_render_jobs.return_value = (mock_jobs, 3)

        result = list_render_jobs(
            project_id="",
            status="",
            offset=0,
            limit=20,
            current_user=_make_mock_user(),
            svc=mock_service,
        )
        assert result["total"] == 3
        assert len(result["items"]) == 3

    def test_cancel_render_job_success(self):
        from app.api.routes.ai_avatar_render import cancel_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_job = _make_mock_render_job(status="cancelled")
        mock_service.cancel_render_job.return_value = mock_job

        result = cancel_render_job(job_id="render-1", current_user=_make_mock_user(), svc=mock_service)
        assert result.status == "cancelled"

    def test_cancel_render_job_not_found(self):
        from app.api.routes.ai_avatar_render import cancel_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_service.cancel_render_job.return_value = None

        with pytest.raises(HTTPException) as exc_info:
            cancel_render_job(job_id="nonexistent", current_user=_make_mock_user(), svc=mock_service)
        assert exc_info.value.status_code == 404

    def test_cancel_render_job_not_cancellable(self):
        from app.api.routes.ai_avatar_render import cancel_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_job = _make_mock_render_job(status="completed")
        mock_service.cancel_render_job.return_value = mock_job

        with pytest.raises(HTTPException) as exc_info:
            cancel_render_job(job_id="render-1", current_user=_make_mock_user(), svc=mock_service)
        assert exc_info.value.status_code == 400

    def test_retry_render_job_success(self):
        from app.api.routes.ai_avatar_render import retry_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_job = _make_mock_render_job(status="pending")
        mock_service.retry_render_job.return_value = mock_job

        result = retry_render_job(job_id="render-1", current_user=_make_mock_user(), svc=mock_service)
        assert result.status == "pending"

    def test_retry_render_job_not_failed(self):
        from app.api.routes.ai_avatar_render import retry_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_job = _make_mock_render_job(status="completed")
        mock_service.retry_render_job.return_value = mock_job

        with pytest.raises(HTTPException) as exc_info:
            retry_render_job(job_id="render-1", current_user=_make_mock_user(), svc=mock_service)
        assert exc_info.value.status_code == 400

    def test_retry_render_job_not_found(self):
        from app.api.routes.ai_avatar_render import retry_render_job
        from app.services.ai_avatar_render_service import AiAvatarRenderService
        from fastapi import HTTPException

        mock_service = MagicMock(spec=AiAvatarRenderService)
        mock_service.retry_render_job.return_value = None

        with pytest.raises(HTTPException) as exc_info:
            retry_render_job(job_id="nonexistent", current_user=_make_mock_user(), svc=mock_service)
        assert exc_info.value.status_code == 404


class TestBrollOverlayFilter:
    """FFmpeg B-roll 滤镜构建测试."""

    def test_empty_segments_returns_empty(self):
        from packages.domain.video_filter_builder import build_broll_overlay_filter

        result = build_broll_overlay_filter([], 30.0)
        assert result == ""

    def test_pip_mode_generates_overlay(self):
        from packages.domain.video_filter_builder import build_broll_overlay_filter

        segments = [
            {
                "script_segment_index": 0,
                "asset_url": "https://example.com/broll.mp4",
                "mode": "pip",
                "start_time": 5.0,
                "end_time": 10.0,
                "pip_position": "bottom_right",
                "pip_scale": 0.3,
            }
        ]
        result = build_broll_overlay_filter(segments, 30.0)
        assert "overlay" in result or "scale=" in result

    def test_fullscreen_mode_generates_concat(self):
        from packages.domain.video_filter_builder import build_broll_overlay_filter

        segments = [
            {
                "script_segment_index": 0,
                "asset_url": "https://example.com/broll.mp4",
                "mode": "fullscreen",
                "start_time": 5.0,
                "end_time": 10.0,
            }
        ]
        result = build_broll_overlay_filter(segments, 30.0)
        assert "trim" in result or "concat" in result

    def test_cover_extract_command(self):
        from packages.domain.video_filter_builder import build_cover_extract_command

        cmd = build_cover_extract_command({"timestamp": 5.0}, "/tmp/cover.jpg")
        assert "ffmpeg" in cmd
        assert "5.0" in cmd
        assert "/tmp/cover.jpg" in cmd

    def test_cover_extract_empty_config(self):
        from packages.domain.video_filter_builder import build_cover_extract_command

        cmd = build_cover_extract_command({}, "/tmp/cover.jpg")
        assert "ffmpeg" in cmd

    def test_cover_extract_with_size(self):
        from packages.domain.video_filter_builder import build_cover_extract_command

        cmd = build_cover_extract_command(
            {"timestamp": 3.0, "width": 1280, "height": 720},
            "/tmp/cover.jpg",
        )
        assert "scale=" in cmd