105 lines
4.0 KiB
Python
105 lines
4.0 KiB
Python
"""错误码与异常体系测试."""
|
|
|
|
import pytest
|
|
|
|
from src.ai.errors import (
|
|
AIError,
|
|
AILLMUnavailableError,
|
|
AIQuotaExceededError,
|
|
AIRateLimitedError,
|
|
AIValidationError,
|
|
AIWorkflowNotFoundError,
|
|
AIWorkflowStateInvalidError,
|
|
ErrorCode,
|
|
ErrorCodes,
|
|
)
|
|
|
|
|
|
class TestErrorCodes:
|
|
"""错误码映射测试."""
|
|
|
|
def test_http_status_mapping(self) -> None:
|
|
"""错误码 → HTTP 状态码映射正确."""
|
|
assert ErrorCodes.http_status(ErrorCode.AI_UNAUTHORIZED) == 401
|
|
assert ErrorCodes.http_status(ErrorCode.AI_FORBIDDEN) == 403
|
|
assert ErrorCodes.http_status(ErrorCode.AI_RATE_LIMITED) == 429
|
|
assert ErrorCodes.http_status(ErrorCode.AI_LLM_UNAVAILABLE) == 200
|
|
assert ErrorCodes.http_status(ErrorCode.AI_LLM_TIMEOUT) == 504
|
|
assert ErrorCodes.http_status(ErrorCode.AI_INTERNAL_ERROR) == 500
|
|
|
|
def test_http_status_unknown_fallback(self) -> None:
|
|
"""未知错误码回退 500."""
|
|
# 使用一个不存在的 ErrorCode 值
|
|
assert ErrorCodes.http_status(ErrorCode.AI_INTERNAL_ERROR) == 500
|
|
|
|
def test_error_code_values(self) -> None:
|
|
"""ErrorCode 枚举值与名称一致."""
|
|
assert ErrorCode.AI_UNAUTHORIZED == "AI_UNAUTHORIZED"
|
|
assert ErrorCode.AI_LLM_ALL_PROVIDERS_FAILED == "AI_LLM_ALL_PROVIDERS_FAILED"
|
|
|
|
def test_all_codes_have_http_mapping(self) -> None:
|
|
"""所有错误码都有 HTTP 映射."""
|
|
for code in ErrorCode:
|
|
assert code in ErrorCodes.HTTP_STATUS, f"{code} missing HTTP mapping"
|
|
|
|
|
|
class TestExceptions:
|
|
"""异常类测试."""
|
|
|
|
def test_ai_error_default(self) -> None:
|
|
"""AIError 默认值."""
|
|
err = AIError()
|
|
assert err.code == ErrorCode.AI_INTERNAL_ERROR
|
|
assert err.http_status == 500
|
|
assert err.details == {}
|
|
|
|
def test_ai_validation_error(self) -> None:
|
|
"""AIValidationError 使用 AI_INVALID_MODEL code."""
|
|
err = AIValidationError("invalid model name")
|
|
assert err.code == ErrorCode.AI_INVALID_MODEL
|
|
assert err.http_status == 400
|
|
assert "invalid model name" in str(err)
|
|
|
|
def test_rate_limited_error(self) -> None:
|
|
"""AIRateLimitedError 携带 dimension + limit."""
|
|
err = AIRateLimitedError("user", 10)
|
|
assert err.code == ErrorCode.AI_RATE_LIMITED
|
|
assert err.http_status == 429
|
|
assert err.details["dimension"] == "user"
|
|
assert err.details["limit"] == 10
|
|
|
|
def test_quota_exceeded_error(self) -> None:
|
|
"""AIQuotaExceededError 携带 scope + used + budget."""
|
|
err = AIQuotaExceededError("school", 1_200_000, 1_000_000)
|
|
assert err.code == ErrorCode.AI_QUOTA_EXCEEDED
|
|
assert err.details["scope"] == "school"
|
|
assert err.details["used"] == 1_200_000
|
|
assert err.details["budget"] == 1_000_000
|
|
|
|
def test_llm_unavailable_error(self) -> None:
|
|
"""AILLMUnavailableError 默认消息."""
|
|
err = AILLMUnavailableError()
|
|
assert err.code == ErrorCode.AI_LLM_UNAVAILABLE
|
|
assert err.http_status == 200 # 降级模式仍 200
|
|
assert "all providers failed" in str(err)
|
|
|
|
def test_workflow_not_found_error(self) -> None:
|
|
"""AIWorkflowNotFoundError 携带 workflow_id."""
|
|
err = AIWorkflowNotFoundError("wf-123")
|
|
assert err.code == ErrorCode.AI_WORKFLOW_NOT_FOUND
|
|
assert err.http_status == 404
|
|
assert err.details["workflow_id"] == "wf-123"
|
|
|
|
def test_workflow_state_invalid_error(self) -> None:
|
|
"""AIWorkflowStateInvalidError 携带状态信息."""
|
|
err = AIWorkflowStateInvalidError("wf-1", "pending", "confirm")
|
|
assert err.code == ErrorCode.AI_WORKFLOW_STATE_INVALID
|
|
assert err.http_status == 409
|
|
assert err.details["current_status"] == "pending"
|
|
assert err.details["action"] == "confirm"
|
|
|
|
def test_ai_error_is_exception(self) -> None:
|
|
"""AIError 是 Exception 子类."""
|
|
with pytest.raises(AIError):
|
|
raise AIError(ErrorCode.AI_INTERNAL_ERROR, "test")
|