xiaoxia-saas/tests/unit/test_dashscope_client.py

"""tests for packages/shared/dashscope_client.py (#2159 Wan 3.0 DashScope client)."""

from __future__ import annotations

from unittest.mock import MagicMock, mock_open, patch

import pytest

_SINGLETON = "_DASHSCOPE_CLIENT_SINGLETON"


@pytest.fixture(autouse=True)
def reset_singleton():
    import packages.shared.dashscope_client as d

    # 兼容实际 singleton 名
    for name in ("_DASHSCOPE_CLIENT_SINGLETON", "_dashscope_client"):
        if hasattr(d, name):
            setattr(d, name, None)
    yield
    for name in ("_DASHSCOPE_CLIENT_SINGLETON", "_dashscope_client"):
        if hasattr(d, name):
            setattr(d, name, None)


def _make_settings(api_key="test-key"):
    return MagicMock(
        dashscope_api_key=api_key,
        dashscope_base_url="https://dashscope.aliyuncs.com/api/v1",
        dashscope_video_timeout=10,
        dashscope_video_poll_interval=0,
        video_dir="/tmp/videos",
    )


class TestDashScopeAvailability:
    def test_unavailable_without_key(self):
        from packages.shared.dashscope_client import get_dashscope_client

        with patch("packages.shared.dashscope_client.get_shared_settings") as ms:
            ms.return_value = _make_settings(api_key="")
            assert get_dashscope_client() is None

    def test_available_with_key(self):
        from packages.shared.dashscope_client import get_dashscope_client

        with patch("packages.shared.dashscope_client.get_shared_settings") as ms:
            ms.return_value = _make_settings()
            c = get_dashscope_client()
            assert c is not None
            assert c.is_available is True


def _mock_stream_response(min_size=2048):
    """构造 httpx.stream 上下文返回值，模拟返回若干字节的 mp4 内容。"""
    m = MagicMock()
    m.status_code = 200
    chunk = b"x" * min_size
    m.iter_bytes.return_value = [chunk]
    ctx = MagicMock()
    ctx.__enter__.return_value = m
    return ctx


class TestDashScopeVideoGeneration:
    def test_happy_path_returns_video_path(self):
        """POST create → GET poll (SUCCEEDED) → download → returns path + correct payload."""
        import packages.shared.dashscope_client as d

        with patch("packages.shared.dashscope_client.get_shared_settings") as ms:
            ms.return_value = _make_settings()
            c = d.DashScopeClient()

        create_resp = MagicMock(status_code=200)
        create_resp.json.return_value = {"output": {"task_id": "task-abc"}}
        poll_resp = MagicMock(status_code=200)
        poll_resp.json.return_value = {
            "output": {"task_status": "SUCCEEDED", "video_url": "http://x/y.mp4"},
            "usage": {"billed_duration": 10},
        }
        # fake file: write enough bytes to pass the size>=1024 check
        m_open = mock_open()
        m_open.return_value.write.return_value = None
        fake_size = {"/tmp/videos/wan_task-abc.mp4": 4096}

        def fake_getsize(p):
            return fake_size.get(p, 0)

        def fake_exists(p):
            return p in fake_size

        with (
            patch.object(d.httpx, "post", return_value=create_resp) as mock_post,
            patch.object(d.httpx, "get", return_value=poll_resp),
            patch.object(d.httpx, "stream", return_value=_mock_stream_response()),
            patch("packages.shared.dashscope_client.time.sleep"),
            patch("packages.shared.dashscope_client.os.makedirs"),
            patch("builtins.open", m_open),
            patch("packages.shared.dashscope_client.os.path.getsize", side_effect=fake_getsize),
            patch("packages.shared.dashscope_client.os.path.exists", side_effect=fake_exists),
        ):
            res = c.video_generation(
                prompt="test",
                duration=5,
                ratio="9:16",
                resolution="720p",
                output_dir="/tmp/videos",
            )
        assert res is not None, "expected success"
        assert res["video_path"] == "/tmp/videos/wan_task-abc.mp4"
        _, kwargs = mock_post.call_args
        body = kwargs["json"]
        assert body["parameters"]["resolution"] == "720P"
        assert body["model"] == "wan3.0-video"

    def test_create_http_error_returns_none(self):
        import packages.shared.dashscope_client as d

        with patch("packages.shared.dashscope_client.get_shared_settings") as ms:
            ms.return_value = _make_settings()
            c = d.DashScopeClient()
        err_resp = MagicMock(status_code=400, text="bad")
        err_resp.raise_for_status.side_effect = RuntimeError("bad")
        with patch.object(d.httpx, "post", return_value=err_resp):
            res = c.video_generation(
                prompt="test", duration=5, ratio="9:16", resolution="720p", output_dir="/tmp/videos"
            )
        assert res is None

    def test_poll_failed_returns_none(self):
        import packages.shared.dashscope_client as d

        with patch("packages.shared.dashscope_client.get_shared_settings") as ms:
            ms.return_value = _make_settings()
            c = d.DashScopeClient()
        create_resp = MagicMock(status_code=200)
        create_resp.json.return_value = {"output": {"task_id": "task-abc"}}
        poll_resp = MagicMock(status_code=200)
        poll_resp.json.return_value = {"output": {"task_status": "FAILED", "message": "nope"}}
        with (
            patch.object(d.httpx, "post", return_value=create_resp),
            patch.object(d.httpx, "get", return_value=poll_resp),
            patch("packages.shared.dashscope_client.time.sleep"),
        ):
            res = c.video_generation(
                prompt="test", duration=5, ratio="9:16", resolution="720p", output_dir="/tmp/videos"
            )
        assert res is None

    def test_empty_prompt_returns_none(self):
        import packages.shared.dashscope_client as d

        with patch("packages.shared.dashscope_client.get_shared_settings") as ms:
            ms.return_value = _make_settings()
            c = d.DashScopeClient()
        assert (
            c.video_generation(prompt="  ", duration=5, ratio="9:16", resolution="720p", output_dir="/tmp/videos")
            is None
        )