xiaoxia-saas/tests/unit/test_lipsync_routes.py

"""对口型 API 路由 + Service 单元测试 — #1796.

CI 增量映射: lipsync.py (route) + lipsync_service.py → test_lipsync_routes.py
"""

import os
from unittest.mock import MagicMock, patch

import pytest

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


@pytest.fixture
def mock_mediakit():
    """Mock MediaKit 客户端."""
    client = MagicMock()
    client.is_available = True
    client.submit_lipsync.return_value = {
        "success": True,
        "task_id": "mk-task-123",
        "request_id": "mk-req-456",
    }
    client.get_task_status.return_value = {
        "success": True,
        "task_id": "mk-task-123",
        "status": "completed",
        "result": {"video_url": "https://output.mp4", "duration": 30.0},
        "created_at": 1777291767,
        "finished_at": 1777291851,
        "expires_at": 1777464650,
    }
    return client


def _make_mock_job(
    job_id="job-1",
    user_id="user-1",
    status="submitted",
    mediakit_task_id="mk-task-123",
    output_video_url="",
    output_duration=0.0,
    error_message="",
    error_code="",
):
    m = MagicMock()
    m.id = job_id
    m.user_id = user_id
    m.project_id = ""
    m.video_url = "https://example.com/video.mp4"
    m.audio_url = "https://example.com/audio.mp3"
    m.enable_video_loop = False
    m.mediakit_task_id = mediakit_task_id
    m.status = status
    m.output_video_url = output_video_url
    m.output_duration = output_duration
    m.error_message = error_message
    m.error_code = error_code
    m.submitted_at = None
    m.completed_at = None
    m.created_at = None
    m.updated_at = None
    return m


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

    def test_valid_video_url(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4",
            audio_url="https://example.com/audio.mp3",
        )
        assert req.video_url == "https://example.com/video.mp4"

    def test_invalid_video_url_not_mp4(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="MP4"):
            CreateLipsyncJobRequest(
                video_url="https://example.com/video.mov",
                audio_url="https://example.com/audio.mp3",
            )

    def test_invalid_video_url_empty(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="不能为空"):
            CreateLipsyncJobRequest(
                video_url="  ",
                audio_url="https://example.com/audio.mp3",
            )

    def test_invalid_video_url_not_http(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="HTTP"):
            CreateLipsyncJobRequest(
                video_url="ftp://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
            )

    def test_valid_audio_formats(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        for ext in [".mp3", ".aac", ".wav", ".m4a", ".flac"]:
            req = CreateLipsyncJobRequest(
                video_url="https://example.com/video.mp4",
                audio_url=f"https://example.com/audio{ext}",
            )
            assert req.audio_url.endswith(ext)

    def test_invalid_audio_format(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        with pytest.raises(ValueError, match="格式不支持"):
            CreateLipsyncJobRequest(
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.ogg",
            )

    def test_enable_video_loop_default(self):
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4",
            audio_url="https://example.com/audio.mp3",
        )
        assert req.enable_video_loop is False

    def test_video_url_strip_query_params(self):
        """视频 URL 含查询参数时，扩展名检查应忽略 ? 后面的部分."""
        from app.schemas.lipsync import CreateLipsyncJobRequest

        req = CreateLipsyncJobRequest(
            video_url="https://example.com/video.mp4?token=abc",
            audio_url="https://example.com/audio.mp3?sign=xyz",
        )
        assert "?token=" in req.video_url


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

    def test_create_job_success(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

        mock_db = MagicMock()
        svc = LipsyncService(mock_db, client=mock_mediakit)

        # 模拟 db.add + db.flush 不报错
        mock_db.add = MagicMock()
        mock_db.flush = MagicMock()
        mock_db.commit = MagicMock()
        mock_db.refresh = 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-task-123"
        mock_mediakit.submit_lipsync.assert_called_once()

    def test_create_job_api_failure(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService
        from app.services.mediakit_client import MediaKitError

        mock_mediakit.submit_lipsync.side_effect = MediaKitError("API 调用失败", code="SubmitFailed")

        mock_db = MagicMock()
        svc = LipsyncService(mock_db, client=mock_mediakit)

        with pytest.raises(MediaKitError, match="API 调用失败"):
            svc.create_job(
                user_id="user-1",
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
            )

    def test_get_job_delegates_to_db(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

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

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.get_job("job-1", "user-1")

        assert result is mock_job
        mock_db.query.assert_called_once()

    def test_get_job_not_found(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

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

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.get_job("nonexistent", "user-1")
        assert result is None

    def test_refresh_job_completed(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

        mock_job = _make_mock_job(status="submitted")
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.refresh_job_status("job-1", "user-1")

        assert result.status == "completed"
        assert result.output_video_url == "https://output.mp4"
        assert result.output_duration == 30.0

    def test_refresh_job_failed(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

        mock_mediakit.get_task_status.return_value = {
            "success": True,
            "task_id": "mk-task-123",
            "status": "failed",
            "error": {"code": "DownloadFailed", "message": "无法下载"},
            "created_at": 1777291767,
            "finished_at": 1777291851,
        }

        mock_job = _make_mock_job(status="submitted")
        mock_db = MagicMock()
        mock_query = MagicMock()
        mock_filter = MagicMock()
        mock_filter.first.return_value = mock_job
        mock_query.filter.return_value = mock_filter
        mock_db.query.return_value = mock_query

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.refresh_job_status("job-1", "user-1")

        assert result.status == "failed"
        assert result.error_code == "DownloadFailed"

    def test_refresh_job_already_completed(self, mock_mediakit):
        """已完成的任务不轮询."""
        from app.services.lipsync_service import LipsyncService

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

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.refresh_job_status("job-1", "user-1")

        # 不应调用 MediaKit
        mock_mediakit.get_task_status.assert_not_called()
        assert result.status == "completed"

    def test_cancel_job_pending(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

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

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.cancel_job("job-1", "user-1")

        assert result.status == "cancelled"

    def test_cancel_job_completed_not_allowed(self, mock_mediakit):
        from app.services.lipsync_service import LipsyncService

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

        svc = LipsyncService(mock_db, client=mock_mediakit)
        result = svc.cancel_job("job-1", "user-1")

        # 已完成不可取消
        assert result.status == "completed"