xiaoxia-saas/tests/unit/test_mediakit_client.py

"""MediaKit 客户端单元测试 — #1796."""

import os
from unittest.mock import MagicMock, patch

import httpx
import pytest

# 确保测试环境有 JWT_SECRET_KEY
os.environ.setdefault("JWT_SECRET_KEY", "dev-secret-key-for-testing")

from app.services.mediakit_client import (
    MediaKitClient,
    MediaKitError,
    get_mediakit_client,
    reset_mediakit_client,
)


@pytest.fixture(autouse=True)
def _reset_client():
    """每个测试前后重置单例."""
    reset_mediakit_client()
    yield
    reset_mediakit_client()


@pytest.fixture
def mock_settings():
    with patch("app.services.mediakit_client.get_api_settings") as m:
        settings = MagicMock()
        settings.mediakit_api_key = "test-api-key"
        settings.mediakit_base_url = "https://mediakit.cn-beijing.volces.com/api/v1"
        settings.mediakit_timeout = 30
        m.return_value = settings
        yield settings


@pytest.fixture
def mock_settings_no_key():
    with patch("app.services.mediakit_client.get_api_settings") as m:
        settings = MagicMock()
        settings.mediakit_api_key = ""
        settings.mediakit_base_url = "https://mediakit.cn-beijing.volces.com/api/v1"
        settings.mediakit_timeout = 30
        m.return_value = settings
        yield settings


class TestMediaKitClientInit:
    """客户端初始化测试."""

    def test_is_available_with_key(self, mock_settings):
        client = MediaKitClient()
        assert client.is_available is True

    def test_is_available_without_key(self, mock_settings_no_key):
        client = MediaKitClient()
        assert client.is_available is False

    def test_get_client_singleton(self, mock_settings):
        c1 = get_mediakit_client()
        c2 = get_mediakit_client()
        assert c1 is c2


class TestSubmitLipsync:
    """提交对口型任务测试."""

    def test_submit_success(self, mock_settings):
        client = MediaKitClient()
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "success": True,
            "task_id": "amk-tool-lip-sync-123",
            "request_id": "req-456",
        }
        mock_response.raise_for_status = MagicMock()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.post.return_value = mock_response
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            result = client.submit_lipsync(
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
            )

            assert result["success"] is True
            assert result["task_id"] == "amk-tool-lip-sync-123"
            assert result["request_id"] == "req-456"

    def test_submit_without_api_key(self, mock_settings_no_key):
        client = MediaKitClient()
        with pytest.raises(MediaKitError, match="未配置"):
            client.submit_lipsync(
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
            )

    def test_submit_api_error(self, mock_settings):
        client = MediaKitClient()
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "success": False,
            "task_id": "",
            "request_id": "req-789",
            "error": {
                "code": "InvalidParameter",
                "message": "must specify audio_url",
                "param": "audio_url",
                "type": "BadRequest",
            },
        }
        mock_response.raise_for_status = MagicMock()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.post.return_value = mock_response
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            with pytest.raises(MediaKitError) as exc_info:
                client.submit_lipsync(
                    video_url="https://example.com/video.mp4",
                    audio_url="https://example.com/audio.mp3",
                )
            assert exc_info.value.code == "InvalidParameter"
            assert "audio_url" in str(exc_info.value)

    def test_submit_timeout(self, mock_settings):
        client = MediaKitClient()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.post.side_effect = httpx.TimeoutException("timeout")
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            with pytest.raises(MediaKitError, match="超时"):
                client.submit_lipsync(
                    video_url="https://example.com/video.mp4",
                    audio_url="https://example.com/audio.mp3",
                )

    def test_submit_with_all_params(self, mock_settings):
        client = MediaKitClient()
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "success": True,
            "task_id": "task-1",
            "request_id": "req-1",
        }
        mock_response.raise_for_status = MagicMock()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.post.return_value = mock_response
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            result = client.submit_lipsync(
                video_url="https://example.com/video.mp4",
                audio_url="https://example.com/audio.mp3",
                enable_video_loop=True,
                callback_url="https://callback.example.com",
                callback_args="my_args",
                client_token="token-123",
            )

            assert result["success"] is True
            # 验证请求参数
            call_args = mock_client.post.call_args
            payload = call_args.kwargs["json"]
            assert payload["enable_video_loop"] is True
            assert payload["callback_url"] == "https://callback.example.com"
            assert payload["callback_args"] == "my_args"
            assert payload["client_token"] == "token-123"


class TestGetTaskStatus:
    """查询任务状态测试."""

    def test_get_status_running(self, mock_settings):
        client = MediaKitClient()
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "success": True,
            "task_id": "task-123",
            "status": "running",
            "created_at": 1777291767,
        }
        mock_response.raise_for_status = MagicMock()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.get.return_value = mock_response
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            result = client.get_task_status("task-123")
            assert result["status"] == "running"
            assert result["result"] is None

    def test_get_status_completed(self, mock_settings):
        client = MediaKitClient()
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "success": True,
            "task_id": "task-123",
            "status": "completed",
            "result": {"video_url": "https://output.mp4", "duration": 60.5},
            "created_at": 1777291767,
            "finished_at": 1777291851,
            "expires_at": 1777464650,
        }
        mock_response.raise_for_status = MagicMock()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.get.return_value = mock_response
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            result = client.get_task_status("task-123")
            assert result["status"] == "completed"
            assert result["result"]["video_url"] == "https://output.mp4"
            assert result["result"]["duration"] == 60.5

    def test_get_status_failed(self, mock_settings):
        client = MediaKitClient()
        mock_response = MagicMock()
        mock_response.json.return_value = {
            "success": True,
            "task_id": "task-123",
            "status": "failed",
            "error": {"code": "DownloadFailed", "message": "无法下载视频"},
            "created_at": 1777291767,
            "finished_at": 1777291851,
        }
        mock_response.raise_for_status = MagicMock()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.get.return_value = mock_response
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            result = client.get_task_status("task-123")
            assert result["status"] == "failed"
            assert result["error"]["code"] == "DownloadFailed"

    def test_get_status_without_api_key(self, mock_settings_no_key):
        client = MediaKitClient()
        with pytest.raises(MediaKitError, match="未配置"):
            client.get_task_status("task-123")

    def test_get_status_network_error(self, mock_settings):
        client = MediaKitClient()

        with patch("httpx.Client") as mock_http:
            mock_client = MagicMock()
            mock_client.get.side_effect = httpx.RequestError("connection refused")
            mock_client.__enter__ = MagicMock(return_value=mock_client)
            mock_client.__exit__ = MagicMock(return_value=False)
            mock_http.return_value = mock_client

            with pytest.raises(MediaKitError, match="网络错误"):
                client.get_task_status("task-123")