"""LipsyncService GPU 路径集成测试 (#1978 异步版本).

#1978 性能修复：GPU 推理从同步等待改为异步。
- _submit_audio_direct 创建 GPU 任务后立即返回，job.status="processing"
- Celery 任务 lipsync_gpu_process_async 负责等待结果+回写
- 本测试验证：创建任务、异步派发、音频转存等逻辑
"""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest


@pytest.fixture()
def fake_db():
    db = MagicMock()
    return db


@pytest.fixture()
def fake_mediakit():
    client = MagicMock()
    client.submit_lipsync.return_value = {"task_id": "mk-task-1"}
    return client


def _make_job(video_url="videos/video.mp4", audio_url="audios/audio.wav"):
    job = MagicMock()
    job.id = "job-1"
    job.user_id = "u1"
    job.project_id = "p1"
    job.video_url = video_url
    job.audio_url = audio_url
    job.enable_video_loop = True
    job.script_text = ""
    job.sentence_timings = None
    return job


def _make_svc(db, mediakit, use_gpu=False):
    from app.services.lipsync_service import LipsyncService

    svc = LipsyncService(db=db, client=mediakit)
    svc.settings.use_gpu_lipsync = use_gpu
    svc._sign_media_url = lambda u: (u or "") + "?signed"
    return svc


def _patch_storage(public_url="https://own-bucket.oss-cn-beijing.aliyuncs.com", signed_suffix="?signed-7d"):
    """patch get_shared_storage_service，返回自家 OSS storage mock."""
    storage = MagicMock()
    storage.public_url = public_url
    storage.get_download_url.side_effect = lambda key_or_url, expires_seconds=3600: key_or_url + signed_suffix
    return patch("app.services.lipsync_service.get_shared_storage_service", return_value=storage)


class TestGpuFallback:
    def test_switch_off_uses_mediakit(self, fake_db, fake_mediakit):
        """开关关闭时直接走 MediaKit，不创建 GPU 任务."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=False)
        job = _make_job()
        with patch.object(svc, "_submit_to_gpu_create") as m_sub:
            svc._submit_audio_direct(job=job)
            m_sub.assert_not_called()
        fake_mediakit.submit_lipsync.assert_called_once()
        assert job.status == "submitted"

    def test_switch_on_no_worker_falls_back(self, fake_db, fake_mediakit):
        """开关打开但 has_available_worker=False → 回退 MediaKit."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = False
        with patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
            job = _make_job()
            svc._submit_audio_direct(job=job)
        fake_gpu_svc.create_task.assert_not_called()
        fake_mediakit.submit_lipsync.assert_called_once()
        assert job.status == "submitted"

    def test_gpu_success_dispatches_async(self, fake_db, fake_mediakit):
        """#1978 异步：GPU 任务创建成功 → job.status=processing，Celery 异步派发."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-1")
        with (
            _patch_storage(),
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
            patch("app.services.lipsync_service.lipsync_gpu_process_async") as m_celery,
        ):
            job = _make_job()
            svc._submit_audio_direct(job=job)
        fake_gpu_svc.create_task.assert_called_once()
        fake_mediakit.submit_lipsync.assert_not_called()
        # 异步模式：job 立即设为 processing，Celery 任务派发
        assert job.status == "processing"
        assert job.mediakit_task_id == "gpu:gpu-task-1"
        m_celery.apply_async.assert_called_once_with(args=("job-1", "u1", "gpu-task-1"))

    def test_gpu_celery_dispatch_failure_falls_back_sync(self, fake_db, fake_mediakit):
        """Celery 派发失败 → 降级同步等待 GPU 结果."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        gpu_done = MagicMock(
            id="gpu-task-1",
            status="done",
            result_url="gpu-lipsync/results/gpu-task-1.mp4",
            result_duration=12.5,
        )
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-1")
        fake_gpu_svc.wait_for_result.return_value = gpu_done
        with (
            _patch_storage() as storage_p,
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
            patch("app.services.lipsync_service.lipsync_gpu_process_async") as m_celery,
        ):
            m_celery.apply_async.side_effect = RuntimeError("Celery down")
            storage = storage_p()
            job = _make_job()
            svc._submit_audio_direct(job=job)
        # 降级同步等待完成
        fake_gpu_svc.wait_for_result.assert_called_once()
        assert job.status == "completed"
        assert job.output_duration == 12.5
        storage.get_download_url.assert_called_once_with(
            "gpu-lipsync/results/gpu-task-1.mp4", expires_seconds=7 * 24 * 3600
        )

    def test_gpu_create_failure_falls_back(self, fake_db, fake_mediakit):
        """GPU 任务创建异常 → 回退 MediaKit."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.side_effect = RuntimeError("DB down")
        with _patch_storage(), patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc):
            job = _make_job()
            svc._submit_audio_direct(job=job)
        fake_mediakit.submit_lipsync.assert_called_once()
        assert job.status == "submitted"

    def test_gpu_create_returns_none_falls_back_mediakit(self, fake_db, fake_mediakit):
        """_submit_to_gpu_create 返回 None（create_task 失败被内部吞掉）→ rollback + MediaKit."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        with (
            _patch_storage(),
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
            patch.object(svc, "_submit_to_gpu_create", return_value=None) as m_create,
        ):
            job = _make_job()
            svc._submit_audio_direct(job=job)
        m_create.assert_called_once()
        fake_db.rollback.assert_called_once()
        fake_mediakit.submit_lipsync.assert_called_once()
        assert job.status == "submitted"

    def test_submit_to_gpu_wait_timeout_returns(self, fake_db, fake_mediakit):
        """降级同步等待：wait_for_result 返回 None → 直接返回，job 保持 processing."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.wait_for_result.return_value = None
        job = _make_job()
        job.status = "processing"
        svc._submit_to_gpu_wait(job=job, gpu_svc=fake_gpu_svc, gpu_task=MagicMock(id="gpu-task-x"))
        fake_gpu_svc.wait_for_result.assert_called_once_with("gpu-task-x")
        fake_db.commit.assert_not_called()
        assert job.status == "processing"

    def test_submit_to_gpu_wait_failed_status_returns(self, fake_db, fake_mediakit):
        """降级同步等待：final_task.status != done → 直接返回."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.wait_for_result.return_value = MagicMock(status="failed", result_url="")
        job = _make_job()
        job.status = "processing"
        svc._submit_to_gpu_wait(job=job, gpu_svc=fake_gpu_svc, gpu_task=MagicMock(id="gpu-task-y"))
        fake_db.commit.assert_not_called()
        assert job.status == "processing"

    def test_gpu_external_audio_persisted_to_own_oss(self, fake_db, fake_mediakit):
        """Bug2 回归：dashscope 临时音频 URL 在创建 GPU 任务前转存自家 OSS."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        dashscope_url = "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/tmp/abc.mp3"
        job = _make_job(audio_url=dashscope_url)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-2")
        with (
            _patch_storage() as storage_p,
            patch("app.services.lipsync_service.safe_download_bytes", return_value=b"FAKE-MP3") as m_dl,
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
            patch("app.services.lipsync_service.lipsync_gpu_process_async"),
        ):
            storage = storage_p()
            storage.upload_file.return_value = "https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3"
            svc._submit_audio_direct(job=job)
        # 外部音频在 GPU 分支被额外下载并转存到约定 key
        gpu_dl_calls = [c for c in m_dl.call_args_list if c.kwargs.get("purpose") == "lipsync_gpu_tts_audio"]
        assert len(gpu_dl_calls) == 1
        assert gpu_dl_calls[0].args[0] == dashscope_url
        storage.upload_file.assert_called_once()
        args, kwargs = storage.upload_file.call_args
        assert args[1] == "lipsync-tts/u1/job-1.mp3"
        assert kwargs.get("content_type") == "audio/mpeg"
        # 创建 GPU 任务时用的是自家 OSS URL
        kwargs_create = fake_gpu_svc.create_task.call_args.kwargs
        assert kwargs_create["audio_url"] == "https://own-bucket.oss-cn-beijing.aliyuncs.com/lipsync-tts/u1/job-1.mp3"
        assert kwargs_create["audio_url"] != dashscope_url

    def test_gpu_own_audio_not_repersisted(self, fake_db, fake_mediakit):
        """Bug2：已是自家 OSS 的音频（含裸 key）不重复下载转存."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        job = _make_job(audio_url="lipsync-tts/u1/job-1.mp3")
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-3")
        with (
            _patch_storage() as storage_p,
            patch("app.services.lipsync_service.safe_download_bytes") as m_dl,
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
            patch("app.services.lipsync_service.lipsync_gpu_process_async"),
        ):
            storage = storage_p()
            svc._submit_audio_direct(job=job)
        # GPU 转存分支不应下载/上传
        gpu_dl_calls = [c for c in m_dl.call_args_list if c.kwargs.get("purpose") == "lipsync_gpu_tts_audio"]
        assert gpu_dl_calls == []
        storage.upload_file.assert_not_called()
        assert fake_gpu_svc.create_task.call_args.kwargs["audio_url"] == "lipsync-tts/u1/job-1.mp3"

    def test_gpu_external_audio_persist_fail_uses_original_url(self, fake_db, fake_mediakit):
        """Bug2：外部音频转存失败不阻断，用原始 URL 建任务."""
        svc = _make_svc(fake_db, fake_mediakit, use_gpu=True)
        dashscope_url = "https://dashscope-result-bj.oss-cn-beijing.aliyuncs.com/tmp/abc.mp3"
        job = _make_job(audio_url=dashscope_url)
        fake_gpu_svc = MagicMock()
        fake_gpu_svc.has_available_worker.return_value = True
        fake_gpu_svc.create_task.return_value = MagicMock(id="gpu-task-4")
        with (
            _patch_storage() as storage_p,
            patch("app.services.lipsync_service.safe_download_bytes", side_effect=RuntimeError("network blocked")),
            patch("app.services.gpu_lipsync_service.GpuLipsyncService", return_value=fake_gpu_svc),
            patch("app.services.lipsync_service.lipsync_gpu_process_async"),
        ):
            storage = storage_p()
            svc._submit_audio_direct(job=job)
        storage.upload_file.assert_not_called()
        assert fake_gpu_svc.create_task.call_args.kwargs["audio_url"] == dashscope_url


class TestRefreshGpuStale:
    """refresh_job_status 的 GPU 异步 stale 超时分支."""

    def test_stale_gpu_job_marked_failed(self, fake_db):
        from datetime import UTC, datetime, timedelta

        svc = _make_svc(fake_db, MagicMock(), use_gpu=True)
        job = MagicMock()
        job.status = "processing"
        job.mediakit_task_id = "gpu:gpu-task-stale"
        job.updated_at = datetime.now(UTC) - timedelta(minutes=31)
        with patch.object(svc, "get_job", return_value=job):
            result = svc.refresh_job_status("job-stale", "u1")
        assert result is job
        assert job.status == "failed"
        assert job.error_code == "GpuTimeout"
        fake_db.commit.assert_called_once()

    def test_fresh_gpu_job_left_processing(self, fake_db):
        from datetime import UTC, datetime, timedelta

        svc = _make_svc(fake_db, MagicMock(), use_gpu=True)
        job = MagicMock()
        job.status = "processing"
        job.mediakit_task_id = "gpu:gpu-task-fresh"
        job.updated_at = datetime.now(UTC) - timedelta(minutes=2)
        with patch.object(svc, "get_job", return_value=job):
            result = svc.refresh_job_status("job-fresh", "u1")
        assert result is job
        assert job.status == "processing"
        fake_db.commit.assert_not_called()

    def test_naive_updated_at_stale_marked_failed(self, fake_db):
        """updated_at 为 naive datetime 时按 UTC 补时区后再判定."""
        from datetime import UTC, datetime, timedelta

        svc = _make_svc(fake_db, MagicMock(), use_gpu=True)
        job = MagicMock()
        job.status = "gpu_processing"
        job.mediakit_task_id = "gpu:gpu-task-naive"
        job.updated_at = datetime.now(UTC).replace(tzinfo=None) - timedelta(minutes=31)
        with patch.object(svc, "get_job", return_value=job):
            svc.refresh_job_status("job-naive", "u1")
        assert job.status == "failed"
        assert job.error_code == "GpuTimeout"


class TestGpuServiceHelpers:
    """GpuLipsyncService.has_available_worker 测试."""

    def test_no_workers(self, fake_db):
        from app.services.gpu_lipsync_service import GpuLipsyncService

        svc = GpuLipsyncService(db=fake_db)
        fake_db.query.return_value.filter.return_value.first.return_value = None
        assert svc.has_available_worker() is False

    def test_fresh_worker_available(self, fake_db):
        from app.services.gpu_lipsync_service import GpuLipsyncService

        svc = GpuLipsyncService(db=fake_db)
        svc.settings.gpu_worker_stale_seconds = 300
        fake_db.query.return_value.filter.return_value.first.return_value = MagicMock()
        assert svc.has_available_worker() is True

    def test_stale_worker_unavailable(self, fake_db):
        from app.services.gpu_lipsync_service import GpuLipsyncService

        svc = GpuLipsyncService(db=fake_db)
        fake_db.query.return_value.filter.return_value.first.return_value = None
        assert svc.has_available_worker() is False


# ── cancel_job 取消链路 (#2009) ─────────────────────────────────────


def _build_sqlite_session():
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker

    from packages.adapters.sqlalchemy_impl import models as _  # noqa: F401
    from packages.adapters.sqlalchemy_impl.models import Base

    engine = create_engine("sqlite:///:memory:", future=True)
    Base.metadata.create_all(engine)
    Session = sessionmaker(bind=engine, future=True)
    return Session()


def _make_real_job(db, *, status="processing", mediakit_task_id="gpu:gpu-task-1"):
    import uuid

    from packages.adapters.sqlalchemy_impl.models import LipsyncJobModel

    job = LipsyncJobModel(
        id=str(uuid.uuid4()),
        user_id="u1",
        project_id="p1",
        video_url="videos/v.mp4",
        audio_url="audios/a.wav",
        enable_video_loop=True,
        mediakit_task_id=mediakit_task_id,
        status=status,
    )
    db.add(job)
    db.commit()
    return job


def test_cancel_processing_gpu_job_marks_gpu_task_cancelled():
    """processing 的 GPU job 取消时，关联 GpuLipsyncTask 必须同步置 cancelled."""
    import uuid

    from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel

    db = _build_sqlite_session()
    gpu_task_id = str(uuid.uuid4())
    gpu_task = GpuLipsyncTaskModel(
        id=gpu_task_id,
        video_url="v",
        audio_url="a",
        status="processing",
        worker_id="w-1",
        attempt=1,
    )
    db.add(gpu_task)
    db.commit()

    job = _make_real_job(db, mediakit_task_id=f"gpu:{gpu_task_id}")
    from app.services.lipsync_service import LipsyncService

    svc = LipsyncService(db=db, client=MagicMock())
    result = svc.cancel_job(job.id, "u1")

    assert result.status == "cancelled"
    db.refresh(gpu_task)
    assert gpu_task.status == "cancelled"
    assert gpu_task.error_msg == "用户取消"
    assert gpu_task.finished_at is not None


def test_cancel_processing_gpu_job_skips_non_processing_gpu_task():
    """GPU task 已不在 processing（如已 done）时，取消 job 不应改它，也不报错."""
    import uuid

    from packages.adapters.sqlalchemy_impl.models import GpuLipsyncTaskModel

    db = _build_sqlite_session()
    gpu_task_id = str(uuid.uuid4())
    gpu_task = GpuLipsyncTaskModel(
        id=gpu_task_id, video_url="v", audio_url="a", status="done", worker_id="w-1", attempt=1
    )
    db.add(gpu_task)
    db.commit()

    job = _make_real_job(db, mediakit_task_id=f"gpu:{gpu_task_id}")
    from app.services.lipsync_service import LipsyncService

    svc = LipsyncService(db=db, client=MagicMock())
    result = svc.cancel_job(job.id, "u1")

    assert result.status == "cancelled"
    db.refresh(gpu_task)
    assert gpu_task.status == "done"  # 没被动


def test_cancel_processing_non_gpu_job_does_not_touch_gpu_table():
    """mediakit_task_id 不是 gpu: 前缀（普通 MediaKit 任务）时，不查 GPU task."""
    db = _build_sqlite_session()
    job = _make_real_job(db, mediakit_task_id="mk-task-99")
    from app.services.lipsync_service import LipsyncService

    svc = LipsyncService(db=db, client=MagicMock())
    result = svc.cancel_job(job.id, "u1")
    assert result.status == "cancelled"


def test_cancel_completed_job_unchanged():
    """completed 状态不可取消，cancel_job 原样返回."""
    db = _build_sqlite_session()
    job = _make_real_job(db, status="completed", mediakit_task_id="gpu:x")
    from app.services.lipsync_service import LipsyncService

    svc = LipsyncService(db=db, client=MagicMock())
    result = svc.cancel_job(job.id, "u1")
    assert result.status == "completed"


def test_cancel_job_not_found_returns_none():
    db = _build_sqlite_session()
    from app.services.lipsync_service import LipsyncService

    svc = LipsyncService(db=db, client=MagicMock())
    assert svc.cancel_job("nonexistent", "u1") is None