"""AI 数字人口型视频生成速度优化 — 单元测试.

验证两个优化点：
1. FFmpeg 编码 preset 从 fast 改为 veryfast（提速 30~50%）
2. TTS 合成从同步改为 Celery 异步任务（API 响应从 6~35s 降到 <1s）

Issue: lipsync-speed-optimization
"""

import os
from unittest.mock import MagicMock, patch

import pytest

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


# ═══════════════════════════════════════════════════════════════════════════════
# 优化1: FFmpeg 编码提速 — preset veryfast
# ═══════════════════════════════════════════════════════════════════════════════


class TestFFmpegPresetOptimization:
    """验证 FFmpeg 编码命令从 -preset fast 改为 -preset veryfast."""

    def test_preset_is_veryfast(self):
        """_build_ffmpeg_command 输出必须包含 -preset veryfast."""
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        svc = AiAvatarRenderService.__new__(AiAvatarRenderService)
        cmd = svc._build_ffmpeg_command(
            input_video="https://example.com/video.mp4",
            b_roll_segments=[],
            filter_complex="",
            final_label=None,
            output_path="/tmp/output.mp4",
        )
        assert "-preset veryfast" in cmd, f"期望 -preset veryfast，实际命令: {cmd}"

    def test_preset_veryfast_with_filter(self):
        """带滤镜场景下也必须使用 veryfast."""
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        svc = AiAvatarRenderService.__new__(AiAvatarRenderService)
        cmd = svc._build_ffmpeg_command(
            input_video="https://example.com/video.mp4",
            b_roll_segments=[],
            filter_complex="overlay=0:0",
            final_label="[v]",
            output_path="/tmp/output.mp4",
        )
        assert "-preset veryfast" in cmd
        assert "-filter_complex" in cmd

    def test_preset_not_fast(self):
        """确保不再使用旧的 -preset fast."""
        from app.services.ai_avatar_render_service import AiAvatarRenderService

        svc = AiAvatarRenderService.__new__(AiAvatarRenderService)
        cmd = svc._build_ffmpeg_command(
            input_video="https://example.com/video.mp4",
            b_roll_segments=[],
            filter_complex="",
            final_label=None,
            output_path="/tmp/output.mp4",
        )
        # 确保是 veryfast 而不是 fast
        assert "-preset veryfast" in cmd
        # 排除 "fast" 单独出现（veryfast 包含 fast 子串，需精确判断）
        parts = cmd.split()
        preset_idx = parts.index("-preset")
        assert parts[preset_idx + 1] == "veryfast"


# ═══════════════════════════════════════════════════════════════════════════════
# 优化2: TTS 合成 Celery 异步化
# ═══════════════════════════════════════════════════════════════════════════════


def _make_service_with_mocks():
    """构造 LipsyncService 测试实例及 mock 依赖."""
    from app.services.lipsync_service import LipsyncService

    db = MagicMock()
    client = MagicMock()
    client.is_available = True
    client.submit_lipsync.return_value = {
        "success": True,
        "task_id": "mk-1",
        "request_id": "req-1",
    }
    cosy = MagicMock()
    cosy.submit_synthesize_task.return_value = {
        "audio_url": "https://tts/raw.mp3",
        "request_id": "tts-req",
        "audio_duration": 3.0,
    }
    svc = LipsyncService(db, client=client, cosyvoice_service=cosy, voice_clone_repo=MagicMock())
    # _resolve_voice_id 默认原样返回（repo.get 返回 None）
    svc._voice_clone_repo.get.return_value = None
    return svc, client, cosy


class TestCreateJobAsyncTTS:
    """验证 TTS 模式改为 Celery 异步后的行为."""

    def test_tts_mode_returns_tts_processing_status(self):
        """TTS 模式下 create_job 立即返回，状态为 tts_processing."""
        svc, client, cosy = _make_service_with_mocks()

        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
            mock_task.apply_async = MagicMock()

            job = svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="longxiaochun_v3",
                script_text="大家好",
                speed=1.0,
                emotion="",
            )

        assert job.status == "tts_processing"

    def test_tts_mode_dispatches_celery_task(self):
        """TTS 模式必须 dispatch Celery 异步任务."""
        svc, client, cosy = _make_service_with_mocks()

        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
            mock_task.apply_async = MagicMock()

            svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="v-1",
                script_text="测试文本",
            )

        mock_task.apply_async.assert_called_once()
        call_kwargs = mock_task.apply_async.call_args
        args = call_kwargs.kwargs.get("args") or call_kwargs[1].get("args", call_kwargs[0][0] if call_kwargs[0] else ())
        assert args[1] == "user-1"  # user_id
        assert args[2] == "v-1"  # voice_id
        assert args[3] == "测试文本"  # script_text

    def test_tts_mode_celery_dispatch_failure_still_creates_job(self):
        """Celery dispatch 失败时，job 记录已创建，状态保持 tts_processing."""
        svc, client, cosy = _make_service_with_mocks()

        with patch("app.services.lipsync_service.tts_synthesize_and_submit") as mock_task:
            mock_task.apply_async = MagicMock(side_effect=Exception("Celery broker down"))

            job = svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                voice_id="v-1",
                script_text="测试文本",
            )

        # job 已创建
        assert job is not None
        assert job.status == "tts_processing"
        # MediaKit 未被调用
        client.submit_lipsync.assert_not_called()

    def test_tts_mode_voice_validation_still_sync(self):
        """TTS 模式下音色校验仍在 HTTP 请求中同步执行."""
        from app.services.mediakit_client import MediaKitError

        svc, client, cosy = _make_service_with_mocks()
        # 模拟音色属于其他用户
        other_profile = MagicMock()
        other_profile.user_id = "user-other"
        svc._voice_clone_repo.get.return_value = other_profile

        with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit"):
            with pytest.raises(MediaKitError) as exc:
                svc.create_job(
                    user_id="user-1",
                    video_url="https://example.com/video.mp4",
                    voice_id="clone-profile-id",
                    script_text="测试",
                )
        assert exc.value.code == "VoiceForbidden"

    def test_tts_mode_missing_input_raises_immediately(self):
        """缺少 voice_id 或 script_text 时立即报错，不 dispatch Celery 任务."""
        from app.services.mediakit_client import MediaKitError

        svc, client, cosy = _make_service_with_mocks()

        with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit") as mock_task:
            mock_task.delay = MagicMock()

            with pytest.raises(MediaKitError) as exc:
                svc.create_job(
                    user_id="user-1",
                    video_url="https://example.com/video.mp4",
                    # 缺少 voice_id 和 script_text
                )
            assert exc.value.code == "InvalidInput"

        # Celery 任务未被 dispatch
        mock_task.delay.assert_not_called()
        # TTS 和 MediaKit 均未调用
        cosy.submit_synthesize_task.assert_not_called()
        client.submit_lipsync.assert_not_called()


class TestCreateJobDirectAudio:
    """验证直接音频模式不受异步化影响."""

    def test_direct_audio_still_submits_synchronously(self):
        """直接音频模式仍然同步提交 MediaKit，状态为 submitted."""
        svc, client, cosy = _make_service_with_mocks()

        with patch("app.tasks.lipsync_tts.tts_synthesize_and_submit") as mock_task:
            mock_task.delay = MagicMock()

            job = svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
            )

        assert job.status == "submitted"
        assert job.mediakit_task_id == "mk-1"
        client.submit_lipsync.assert_called_once()
        # TTS Celery 任务不应被调用
        mock_task.delay.assert_not_called()

    def test_direct_audio_skips_tts(self):
        """直接音频模式不调用 CosyVoice TTS."""
        svc, client, cosy = _make_service_with_mocks()

        job = svc.create_job(
            user_id="user-1",
            video_url="https://example.com/video.mp4",
            audio_url="https://example.com/audio.mp3",
        )

        cosy.submit_synthesize_task.assert_not_called()
        call_kwargs = client.submit_lipsync.call_args
        assert call_kwargs.kwargs["audio_url"] == "https://example.com/audio.mp3"


class TestCancelJobTtsProcessing:
    """验证 tts_processing 状态的任务可以被取消."""

    def test_cancel_tts_processing(self):
        """tts_processing 状态的任务可以成功取消."""
        svc, client, cosy = _make_service_with_mocks()

        mock_job = MagicMock()
        mock_job.status = "tts_processing"
        mock_job.id = "job-1"
        svc.get_job = MagicMock(return_value=mock_job)

        result = svc.cancel_job("job-1", "user-1")
        assert result.status == "cancelled"

    def test_cancel_pending_still_works(self):
        """pending 状态仍可取消."""
        svc, client, cosy = _make_service_with_mocks()

        mock_job = MagicMock()
        mock_job.status = "pending"
        mock_job.id = "job-1"
        svc.get_job = MagicMock(return_value=mock_job)

        result = svc.cancel_job("job-1", "user-1")
        assert result.status == "cancelled"

    def test_cancel_submitted_still_works(self):
        """submitted 状态仍可取消."""
        svc, client, cosy = _make_service_with_mocks()

        mock_job = MagicMock()
        mock_job.status = "submitted"
        mock_job.id = "job-1"
        svc.get_job = MagicMock(return_value=mock_job)

        result = svc.cancel_job("job-1", "user-1")
        assert result.status == "cancelled"


# ═══════════════════════════════════════════════════════════════════════════════
# lipsync_tts.py — Celery 异步任务单元测试
# ═══════════════════════════════════════════════════════════════════════════════

import sys
import types


class _FakeQuery:
    """模拟 SQLAlchemy query.filter().first() 链式调用."""

    def __init__(self, job):
        self._job = job

    def filter(self, *args, **kwargs):
        return self

    def first(self):
        return self._job


def _make_fake_job(**kwargs):
    """构造可 setattr 的 job 记录."""
    job = MagicMock()
    job.id = kwargs.get("job_id", "job-1")
    job.user_id = kwargs.get("user_id", "user-1")
    job.status = kwargs.get("status", "tts_processing")
    job.audio_url = kwargs.get("audio_url", "")
    job.video_url = kwargs.get("video_url", "https://oss/video.mp4")
    job.mediakit_task_id = kwargs.get("mediakit_task_id", "")
    job.enable_video_loop = kwargs.get("enable_video_loop", False)
    job.error_code = ""
    job.error_message = ""
    job.submitted_at = None
    job.updated_at = None
    return job


def _build_session(job):
    """构造 mock DB session + factory. 返回 (session, factory_patch_ctx_value)."""
    session = MagicMock()
    session.query.return_value = _FakeQuery(job)
    session.commit = MagicMock()
    session.close = MagicMock()
    factory = MagicMock(return_value=session)
    return session, factory


def _apply_all_patches(
    *,
    job=None,
    cosyvoice_service=None,
    cosyvoice_side_effect=None,
    cosyvoice_error=None,
    download_bytes=b"AUDIO",
    download_error=None,
    storage=None,
    mk_client=None,
    mk_submit_return=None,
    mk_submit_error=None,
):
    """统一构造测试需要的 patch 列表.

    lipsync_tts.run() 在函数体内部懒 import 多个模块，通过 sys.modules 注入
    伪造包路径避免真实导入；对存在的模块用 patch() 替换返回值/side_effect。
    """
    # 构造不存在的 database 模块
    fake_db_mod = types.ModuleType("packages.adapters.sqlalchemy_impl.database")
    session, factory = _build_session(job)
    fake_db_mod.SessionLocal = factory

    patches = [
        patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.database": fake_db_mod}),
        patch(
            "app.tasks.lipsync_tts._sign_media_url",
            side_effect=lambda url: url + "?signed" if url else url,
        ),
    ]

    # CosyVoice
    if cosyvoice_service is not None:
        cosy_instance = cosyvoice_service
    else:
        cosy_instance = MagicMock()
        if cosyvoice_side_effect is not None:
            cosy_instance.submit_synthesize_task.side_effect = cosyvoice_side_effect
        elif cosyvoice_error is not None:
            cosy_instance.submit_synthesize_task.side_effect = cosyvoice_error
        else:
            cosy_instance.submit_synthesize_task.return_value = {"audio_url": "https://tts/raw.mp3"}
    patches.append(patch("packages.application.cosyvoice_service.CosyVoiceService", return_value=cosy_instance))

    # safe_download_bytes
    if download_error is not None:
        patches.append(patch("packages.shared.url_security.safe_download_bytes", side_effect=download_error))
    else:
        patches.append(patch("packages.shared.url_security.safe_download_bytes", return_value=download_bytes))

    # Storage
    if storage is None:
        storage = MagicMock()
        storage.public_url = "https://oss.example.com"
        storage.upload_file.return_value = "https://oss.example.com/tts.mp3"
    patches.append(patch("packages.shared.storage.get_shared_storage_service", return_value=storage))

    # MediaKit client
    if mk_client is not None:
        patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=mk_client))
    else:
        client = MagicMock()
        if mk_submit_error is not None:
            client.submit_lipsync.side_effect = mk_submit_error
        else:
            client.submit_lipsync.return_value = mk_submit_return or {"task_id": "mk-1"}
        patches.append(patch("app.services.mediakit_client.get_mediakit_client", return_value=client))

    return session, patches


class TestTtsSynthesizeAndSubmit:
    """测试 Celery 任务 tts_synthesize_and_submit.run 的所有分支."""

    def test_job_not_found_returns_early(self):
        """Job 不存在 → 日志报错直接返回，不抛异常."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        session, patches = _apply_all_patches(job=None)
        entered = [p.__enter__() for p in patches]
        try:
            tts_synthesize_and_submit.run("missing-job", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        session.commit.assert_not_called()
        session.close.assert_called_once()

    def test_cancelled_job_skipped(self):
        """Job 已 cancelled → 跳过不处理，不调用 TTS/MediaKit."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="cancelled")
        session, patches = _apply_all_patches(job=job)
        entered = [p.__enter__() for p in patches]
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        session.commit.assert_not_called()
        session.close.assert_called_once()
        assert job.status == "cancelled"

    def test_happy_path_tts_to_mediakit(self):
        """正常流程：TTS 合成 → 下载 → OSS → 签名 → 提交 MediaKit → submitted."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="tts_processing")
        storage = MagicMock()
        storage.public_url = "https://oss.example.com"
        storage.upload_file.return_value = "https://oss.example.com/lipsync-tts/u/j.mp3"

        session, patches = _apply_all_patches(
            job=job,
            storage=storage,
            mk_submit_return={"task_id": "mk-999"},
        )
        for p in patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        assert job.audio_url == "https://oss.example.com/lipsync-tts/u/j.mp3"
        assert job.mediakit_task_id == "mk-999"
        assert job.status == "submitted"
        assert job.submitted_at is not None
        session.close.assert_called_once()

    def test_tts_cosyvoice_error_marks_failed(self):
        """CosyVoiceError → 标记 failed，error_code=TTSSynthesisFailed."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        from packages.application.cosyvoice_service import CosyVoiceError

        job = _make_fake_job(status="tts_processing")
        session, patches = _apply_all_patches(
            job=job,
            cosyvoice_error=CosyVoiceError("TTS 服务异常"),
        )
        for p in patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        assert job.status == "failed"
        assert job.error_code == "TTSSynthesisFailed"
        assert "TTS 合成失败" in job.error_message
        session.close.assert_called_once()

    def test_tts_value_error_marks_failed(self):
        """ValueError → 标记 failed，error_code=TTSInvalidParam."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="tts_processing")
        session, patches = _apply_all_patches(
            job=job,
            cosyvoice_error=ValueError("speed 参数非法"),
        )
        for p in patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        assert job.status == "failed"
        assert job.error_code == "TTSInvalidParam"
        assert "TTS 参数错误" in job.error_message
        session.close.assert_called_once()

    def test_tts_no_audio_url_marks_failed(self):
        """TTS 返回空 audio_url → 标记 failed，error_code=TTSNoAudio."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="tts_processing")
        cosy = MagicMock()
        cosy.submit_synthesize_task.return_value = {"audio_url": ""}
        session, patches = _apply_all_patches(job=job, cosyvoice_service=cosy)
        for p in patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        assert job.status == "failed"
        assert job.error_code == "TTSNoAudio"
        session.close.assert_called_once()

    def test_oss_upload_failure_falls_back_to_temp_url(self):
        """OSS 上传失败 → 回退临时 URL，继续提交 MediaKit."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="tts_processing")
        storage = MagicMock()
        storage.public_url = "https://oss.example.com"
        storage.upload_file.side_effect = Exception("OSS 上传超时")
        session, patches = _apply_all_patches(
            job=job,
            storage=storage,
            mk_submit_return={"task_id": "mk-77"},
        )
        for p in patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        # 回退到临时 URL
        assert job.audio_url == "https://tts/raw.mp3"
        assert job.status == "submitted"
        assert job.mediakit_task_id == "mk-77"
        session.close.assert_called_once()

    def test_mediakit_submit_failure_marks_failed(self):
        """MediaKit 提交失败（MediaKitError）→ 标记 failed."""
        from app.services.mediakit_client import MediaKitError
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="tts_processing")
        err = MediaKitError("GPU 不可用", code="MediaKitUnavailable")
        session, patches = _apply_all_patches(
            job=job,
            mk_submit_error=err,
        )
        for p in patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(patches):
                p.__exit__(None, None, None)

        assert job.status == "failed"
        assert job.error_code == "MediaKitUnavailable"
        session.close.assert_called_once()

    def test_top_level_exception_marks_async_task_error(self):
        """顶层意外异常 → except 分支回写 failed，error_code=AsyncTaskError."""
        from app.tasks.lipsync_tts import tts_synthesize_and_submit

        job = _make_fake_job(status="tts_processing")
        # 不调用 _apply_all_patches，手动构造所有 patch，让 CosyVoiceService 抛异常
        fake_db_mod = types.ModuleType("packages.adapters.sqlalchemy_impl.database")
        session_mock = MagicMock()
        session_mock.query.return_value = _FakeQuery(job)
        session_mock.commit = MagicMock()
        session_mock.close = MagicMock()
        fake_db_mod.SessionLocal = MagicMock(return_value=session_mock)

        all_patches = [
            patch.dict(sys.modules, {"packages.adapters.sqlalchemy_impl.database": fake_db_mod}),
            patch(
                "app.tasks.lipsync_tts._sign_media_url",
                side_effect=lambda url: url + "?signed" if url else url,
            ),
            patch(
                "packages.application.cosyvoice_service.CosyVoiceService",
                side_effect=RuntimeError("unexpected init failure"),
            ),
            patch("packages.shared.url_security.safe_download_bytes", return_value=b"AUDIO"),
            patch("packages.shared.storage.get_shared_storage_service", return_value=MagicMock()),
            patch("app.services.mediakit_client.get_mediakit_client", return_value=MagicMock()),
        ]
        for p in all_patches:
            p.__enter__()
        try:
            tts_synthesize_and_submit.run("job-1", "user-1", "v1", "你好", 1.0, "")
        finally:
            for p in reversed(all_patches):
                p.__exit__(None, None, None)

        assert job.status == "failed"
        assert job.error_code == "AsyncTaskError"
        assert "TTS 异步任务执行异常" in job.error_message
        session_mock.close.assert_called_once()