patbond-ai/tests/providers/volcengine/test_errors.py

"""火山方舟错误分类的测试。

每个用例对应官方错误码表里的一行或一类。重点在三处**如果搞错只会在生产暴露**
的地方：`QuotaExceeded` 的一码两义、内容安全类必须不可重试、未知码必须偏向
可重试而不是直接判死。
"""

from app.domain.generation import Retry, UserFacing
from app.providers.volcengine.errors import classify


class TestQuotaExceededIsTwoDifferentThings:
    """同一个 code，两种完全不同的处置。只看 code 必然错一半。"""

    def test_queue_backlog_is_a_seconds_long_wait(self):
        # 官方原文：「当前账号处于排队中状态的任务数已超过限制，请稍后重试。」
        result = classify(
            "QuotaExceeded", "The request has exceeded the quota. Request ID: x."
        )
        assert result.retry is Retry.BACKOFF
        assert result.user_facing is UserFacing.RATE_LIMITED

    def test_queue_backlog_recognised_from_chinese_message(self):
        result = classify("QuotaExceeded", "当前账号处于排队中状态的任务数已超过限制")
        assert result.retry is Retry.BACKOFF
        assert result.user_facing is UserFacing.RATE_LIMITED

    def test_free_trial_exhausted_needs_billing_action(self):
        # 官方原文：「免费试用额度已消耗完毕，请前往控制台开通对应模型服务。」
        result = classify(
            "QuotaExceeded",
            "Your account [x] has exhausted its free trial quota for the [y] model.",
        )
        assert result.retry is Retry.QUOTA
        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED

    def test_periodic_limit_exhausted_needs_to_wait_for_reset(self):
        result = classify(
            "QuotaExceeded",
            "You have exceeded the 5-hour/weekly/monthly usage quota. It will reset at T.",
        )
        assert result.retry is Retry.QUOTA
        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED


class TestContentRejectionMustNotBeRetried:
    """同样的输入必然同样被拒。重试只会白烧额度再失败。"""

    def test_input_image_sensitive(self):
        result = classify("InputImageSensitiveContentDetected")
        assert result.retry is Retry.NEVER
        assert result.user_facing is UserFacing.CONTENT_REJECTED

    def test_output_video_sensitive(self):
        result = classify("OutputVideoSensitiveContentDetected")
        assert result.retry is Retry.NEVER
        assert result.user_facing is UserFacing.CONTENT_REJECTED

    def test_policy_violation_subcode(self):
        result = classify("InputVideoSensitiveContentDetected.PolicyViolation")
        assert result.retry is Retry.NEVER

    def test_real_person_subcode(self):
        result = classify("InputImageSensitiveContentDetected.PrivacyInformation")
        assert result.retry is Retry.NEVER

    def test_risk_detection_product_codes(self):
        assert classify("InputTextRiskDetection").retry is Retry.NEVER
        assert classify("OutputImageRiskDetection").retry is Retry.NEVER

    def test_but_the_detector_itself_failing_is_retryable(self):
        # 风险识别产品请求失败 ≠ 内容真的违规。
        result = classify("ContentSecurityDetectionError", "Internal error.")
        assert result.retry is Retry.BACKOFF
        assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE


class TestRateLimitFamily:
    def test_endpoint_rpm(self):
        result = classify("RateLimitExceeded.EndpointRPMExceeded")
        assert result.retry is Retry.BACKOFF
        assert result.user_facing is UserFacing.RATE_LIMITED

    def test_model_account_tpm(self):
        assert classify("ModelAccountTpmRateLimitExceeded").retry is Retry.BACKOFF

    def test_server_overloaded(self):
        assert classify("ServerOverloaded").retry is Retry.BACKOFF

    def test_request_burst_too_fast(self):
        # doubao-seed-2.0 之后版本的突增流量保护码。
        assert classify("RequestBurstTooFast").retry is Retry.BACKOFF

    def test_images_per_minute(self):
        assert classify("ModelAccountIpmRateLimitExceeded").retry is Retry.BACKOFF


class TestQuotaAndBillingFamily:
    def test_inference_limit_set_by_account(self):
        result = classify("SetLimitExceeded")
        assert result.retry is Retry.QUOTA
        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED

    def test_concurrency_cap_needs_topup(self):
        assert classify("InflightBatchsizeExceeded").retry is Retry.QUOTA

    def test_account_overdue_is_403_but_same_user_action(self):
        # HTTP 状态码不同（403 而非 429），但用户要做的事一样：去充值。
        result = classify("AccountOverdueError")
        assert result.retry is Retry.QUOTA
        assert result.user_facing is UserFacing.QUOTA_EXHAUSTED

    def test_model_not_activated(self):
        assert classify("ModelNotOpen").retry is Retry.QUOTA


class TestOurOwnFaults:
    """重试无意义，需要人改代码或改配置。"""

    def test_bad_api_key(self):
        result = classify("AuthenticationError")
        assert result.retry is Retry.NEVER
        assert result.user_facing is UserFacing.SERVER_ERROR

    def test_unknown_model_id(self):
        assert classify("InvalidEndpointOrModel.NotFound").retry is Retry.NEVER

    def test_invalid_parameter(self):
        assert classify("InvalidParameter").retry is Retry.NEVER

    def test_task_type_constraint_is_our_parameter_problem(self):
        # Seedance 2.5 全模态任务的参数与模型判定的任务类型不兼容。
        assert classify("InvalidParameter.TaskTypeConstraint").retry is Retry.NEVER

    def test_task_type_mismatch(self):
        assert classify("InvalidParameter.TaskTypeMismatch").retry is Retry.NEVER

    def test_missing_parameter(self):
        assert classify("MissingParameter").retry is Retry.NEVER

    def test_payload_too_large(self):
        assert classify("RequestTooLarge").retry is Retry.NEVER


class TestUpstreamErrors:
    def test_internal_service_error_is_retryable(self):
        result = classify("InternalServiceError")
        assert result.retry is Retry.BACKOFF
        assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE


class TestUnknownCodesLeanTowardsRetry:
    """厂商在持续新增错误码。把未知码判死会让一次抖动变成用户看到的失败；
    重试预算本身有上限，最坏只是多试两次。"""

    def test_unknown_code(self):
        result = classify("SomeBrandNewCodeVolcengineAddedLastWeek")
        assert result.retry is Retry.BACKOFF
        assert result.user_facing is UserFacing.UPSTREAM_UNAVAILABLE

    def test_missing_code(self):
        assert classify(None).retry is Retry.BACKOFF
        assert classify("").retry is Retry.BACKOFF


class TestProviderCodeIsAlwaysPreserved:
    """原始码绝不丢弃——排查时它是唯一线索。"""

    def test_known_code(self):
        assert classify("ServerOverloaded").provider_code == "ServerOverloaded"

    def test_unknown_code(self):
        assert classify("WeirdCode").provider_code == "WeirdCode"

    def test_whitespace_is_trimmed(self):
        assert classify("  ServerOverloaded  ").provider_code == "ServerOverloaded"