patbond-ai/tests/test_seedance.py

"""Seedance 客户端的测试。全部离线——用 httpx 的 MockTransport 拦请求，不打真实厂商。

夹具里的响应体逐字取自官方文档的「返回格式」示例，避免测试通过而真实字段名对不上。
"""

from __future__ import annotations

import httpx
import pytest

from app.providers.errors import Retry, UserFacing
from app.providers.seedance import (
    SeedanceClient,
    SeedanceError,
    SeedanceSubmission,
    TaskStatus,
)

#: 官方查询文档「返回格式」一节的原样示例。
OFFICIAL_SUCCESS_PAYLOAD = {
    "id": "cgt-2026****-****",
    "model": "doubao-seedance-2-0-260128",
    "status": "succeeded",
    "content": {
        "video_url": "https://ark-content-generation-cn-beijing.tos-cn-beijing.volces.com/xxx"
    },
    "usage": {"completion_tokens": 108900, "total_tokens": 108900},
    "created_at": 1779348818,
    "updated_at": 1779348874,
    "seed": 78674,
    "resolution": "720p",
    "ratio": "16:9",
    "duration": 5,
    "framespersecond": 24,
    "service_tier": "default",
    "execution_expires_after": 172800,
    "generate_audio": True,
    "draft": False,
    "priority": 0,
}


#: 显式的假值。防泄漏扫描会拦下 `api_key = "..."` 形态的赋值（规则正确——这正是
#: 真实密钥最常见的泄漏形态），故测试夹具用 ALLOW 清单认可的占位词而非随手起名。
FAKE_API_KEY = "dummy-api-key-for-tests"


def client_with(handler) -> SeedanceClient:
    return SeedanceClient(
        api_key=FAKE_API_KEY,
        client=httpx.Client(transport=httpx.MockTransport(handler)),
    )


class TestSubmissionBody:
    """请求体必须符合官方 content 数组的形状。"""

    def test_text_to_video_sends_only_text(self):
        body = SeedanceSubmission(model="m", prompt="一只猫").to_body()
        assert body["content"] == [{"type": "text", "text": "一只猫"}]
        assert body["model"] == "m"

    def test_image_to_video_adds_first_frame_with_explicit_role(self):
        body = SeedanceSubmission(
            model="m", prompt="动起来", first_frame_url="https://x/a.png"
        ).to_body()
        assert body["content"][0]["type"] == "text"
        image = body["content"][1]
        assert image["type"] == "image_url"
        assert image["image_url"] == {"url": "https://x/a.png"}
        # role 省略时厂商按 first_frame 处理，但显式写出以免依赖默认行为。
        assert image["role"] == "first_frame"

    def test_image_only_without_prompt_is_allowed(self):
        body = SeedanceSubmission(
            model="m", first_frame_url="https://x/a.png"
        ).to_body()
        assert len(body["content"]) == 1
        assert body["content"][0]["type"] == "image_url"

    def test_empty_content_is_rejected_before_hitting_the_vendor(self):
        # 提交空内容必然被厂商拒，但那要花一次往返。本地先拦。
        with pytest.raises(ValueError, match="至少要有一个"):
            SeedanceSubmission(model="m").to_body()

    def test_expiry_defaults_to_the_minimum_not_the_vendor_default(self):
        # 厂商默认 48 小时，对交互式应用过长——被放弃的任务会在厂商侧挂两天。
        # 取值范围 [3600, 259200]，故取下限。
        body = SeedanceSubmission(model="m", prompt="x").to_body()
        assert body["execution_expires_after"] == 3600

    def test_last_frame_requested_by_default_for_free_cover_image(self):
        body = SeedanceSubmission(model="m", prompt="x").to_body()
        assert body["return_last_frame"] is True

    def test_optional_fields_are_omitted_when_unset(self):
        body = SeedanceSubmission(model="m", prompt="x").to_body()
        assert "safety_identifier" not in body
        assert "callback_url" not in body

    def test_safety_identifier_is_passed_through_when_given(self):
        body = SeedanceSubmission(
            model="m", prompt="x", safety_identifier="sha256hash"
        ).to_body()
        assert body["safety_identifier"] == "sha256hash"


class TestSubmit:
    def test_returns_vendor_task_id(self):
        def handler(request: httpx.Request) -> httpx.Response:
            assert request.headers["Authorization"] == f"Bearer {FAKE_API_KEY}"
            assert request.url.path.endswith("/contents/generations/tasks")
            return httpx.Response(200, json={"id": "cgt-123"})

        assert (
            client_with(handler).submit(SeedanceSubmission(model="m", prompt="x"))
            == "cgt-123"
        )

    def test_missing_id_in_success_response_is_an_error(self):
        # 2xx 但没有 ID 意味着我们无法再查询这个任务——不能当成功放过。
        def handler(request):
            return httpx.Response(200, json={"unexpected": True})

        with pytest.raises(SeedanceError, match="缺少任务 ID"):
            client_with(handler).submit(SeedanceSubmission(model="m", prompt="x"))

    def test_network_failure_is_classified_retryable(self):
        def handler(request):
            raise httpx.ConnectError("connection refused")

        with pytest.raises(SeedanceError) as exc:
            client_with(handler).submit(SeedanceSubmission(model="m", prompt="x"))
        assert exc.value.classification.retry is Retry.BACKOFF


class TestGetSuccessfulTask:
    def test_parses_the_official_example_payload(self):
        def handler(request):
            return httpx.Response(200, json=OFFICIAL_SUCCESS_PAYLOAD)

        task = client_with(handler).get("cgt-1")
        assert task.status is TaskStatus.SUCCEEDED
        assert task.status.is_terminal
        assert task.video_url.endswith("/xxx")
        assert task.ratio == "16:9"
        assert task.resolution == "720p"
        assert task.fps == 24
        assert task.completion_tokens == 108900
        assert task.failure is None

    def test_duration_from_frames_is_exact_when_frames_returned(self):
        # 官方例子：133 帧 / 24fps = 5.54 秒。按帧算才精确。
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "id": "x",
                    "status": "succeeded",
                    "frames": 133,
                    "framespersecond": 24,
                    "content": {},
                },
            )

        assert client_with(handler).get("x").duration_ms == 5542

    def test_duration_from_seconds_is_the_vendors_floored_approximation(self):
        # duration 与 frames 只会返回一个。只有 duration 时只能用它，
        # 但它是 帧数/24 向下取整的约数，最多可能短 1 秒。
        def handler(request):
            return httpx.Response(
                200,
                json={"id": "x", "status": "succeeded", "duration": 5, "content": {}},
            )

        assert client_with(handler).get("x").duration_ms == 5000

    def test_duration_is_none_rather_than_a_made_up_zero(self):
        def handler(request):
            return httpx.Response(
                200, json={"id": "x", "status": "running", "content": {}}
            )

        assert client_with(handler).get("x").duration_ms is None

    def test_last_frame_url_is_picked_up_as_cover_image(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "id": "x",
                    "status": "succeeded",
                    "content": {
                        "video_url": "https://v",
                        "last_frame_url": "https://f.png",
                    },
                },
            )

        assert client_with(handler).get("x").last_frame_url == "https://f.png"


class TestGetFailedTask:
    def test_failure_is_classified(self):
        def handler(request):
            return httpx.Response(
                200,
                json={
                    "id": "x",
                    "status": "failed",
                    "content": {},
                    "error": {
                        "code": "OutputVideoSensitiveContentDetected",
                        "message": "output may contain sensitive information",
                    },
                },
            )

        task = client_with(handler).get("x")
        assert task.failure.retry is Retry.NEVER
        assert task.failure.user_facing is UserFacing.CONTENT_REJECTED

    def test_expired_is_handled_even_though_the_query_doc_omits_it(self):
        """查询接口文档只列了 5 个状态，`expired` 仅见于创建接口的回调说明。

        只按查询文档实现会让超时任务掉进未知分支。它是超时而非内容问题，
        因此可重试；且厂商在 expired 时不一定给 error.code，故单独处理。
        """

        def handler(request):
            return httpx.Response(
                200, json={"id": "x", "status": "expired", "content": {}}
            )

        task = client_with(handler).get("x")
        assert task.status is TaskStatus.EXPIRED
        assert task.status.is_terminal
        assert task.failure.retry is Retry.BACKOFF
        assert task.failure.provider_code == "TaskExpired"

    def test_cancelled_is_terminal_but_not_a_failure(self):
        def handler(request):
            return httpx.Response(
                200, json={"id": "x", "status": "cancelled", "content": {}}
            )

        task = client_with(handler).get("x")
        assert task.status.is_terminal
        assert task.failure is None


class TestUnknownStatusDoesNotBlowUp:
    """厂商新增状态时抛异常会让所有在途任务一起失败。"""

    def test_unknown_status_yields_none_and_preserves_raw(self):
        def handler(request):
            return httpx.Response(
                200, json={"id": "x", "status": "brand_new", "content": {}}
            )

        task = client_with(handler).get("x")
        assert task.status is None
        assert task.raw_status == "brand_new"
        assert task.failure is None  # 当作未结束，继续轮询


class TestHttpErrorMapping:
    def test_rate_limit_is_retryable(self):
        def handler(request):
            return httpx.Response(
                429,
                json={
                    "error": {
                        "code": "RateLimitExceeded.EndpointRPMExceeded",
                        "message": "RPM limit exceeded",
                    }
                },
            )

        with pytest.raises(SeedanceError) as exc:
            client_with(handler).get("x")
        assert exc.value.classification.retry is Retry.BACKOFF
        assert exc.value.classification.user_facing is UserFacing.RATE_LIMITED
        assert exc.value.http_status == 429

    def test_quota_exhausted_is_not_worth_retrying(self):
        def handler(request):
            return httpx.Response(
                429,
                json={
                    "error": {
                        "code": "QuotaExceeded",
                        "message": "Your account has exhausted its free trial quota.",
                    }
                },
            )

        with pytest.raises(SeedanceError) as exc:
            client_with(handler).submit(SeedanceSubmission(model="m", prompt="x"))
        assert exc.value.classification.retry is Retry.QUOTA

    def test_bad_api_key_is_our_own_fault(self):
        def handler(request):
            return httpx.Response(
                401,
                json={
                    "error": {
                        "code": "AuthenticationError",
                        "message": "invalid api key",
                    }
                },
            )

        with pytest.raises(SeedanceError) as exc:
            client_with(handler).get("x")
        assert exc.value.classification.retry is Retry.NEVER
        assert exc.value.classification.user_facing is UserFacing.SERVER_ERROR

    def test_non_json_error_body_is_not_swallowed(self):
        # 打到错误地址时常见：网关返回 HTML 错误页。静默会让排查绕远路。
        def handler(request):
            return httpx.Response(502, text="<html>bad gateway</html>")

        with pytest.raises(SeedanceError) as exc:
            client_with(handler).get("x")
        assert "bad gateway" in str(exc.value)
        assert exc.value.classification.retry is Retry.BACKOFF

    def test_non_json_success_body_is_an_error_too(self):
        def handler(request):
            return httpx.Response(200, text="not json")

        with pytest.raises(SeedanceError, match="不是合法 JSON"):
            client_with(handler).get("x")


class TestConstruction:
    def test_missing_api_key_fails_loudly_at_construction(self):
        with pytest.raises(ValueError, match="ARK_API_KEY"):
            SeedanceClient(api_key="")