b322ef1d7a
- Add 111 unit tests covering all client classes and exceptions - Test BaseClientMixin: initialization, validation, retry logic - Test FastGPTClient: HTTP requests, streaming, error handling, context manager - Test ChatClient: all chat operations (completion, histories, records, feedback) - Test AppClient: app analytics and logs - Test all exception classes with various configurations - Add shared pytest fixtures in conftest.py 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
358 lines
12 KiB
Python
358 lines
12 KiB
Python
"""Tests for FastGPT exception classes."""
|
|
|
|
import pytest
|
|
|
|
from fastgpt_client.exceptions import (
|
|
FastGPTError,
|
|
APIError,
|
|
AuthenticationError,
|
|
RateLimitError,
|
|
ValidationError,
|
|
StreamParseError,
|
|
)
|
|
|
|
|
|
class TestFastGPTError:
|
|
"""Test suite for FastGPTError base exception."""
|
|
|
|
def test_init_with_message_only(self):
|
|
"""Test initialization with message only."""
|
|
error = FastGPTError("Test error message")
|
|
|
|
assert error.message == "Test error message"
|
|
assert error.status_code is None
|
|
assert error.response_data == {}
|
|
|
|
def test_init_with_message_and_status_code(self):
|
|
"""Test initialization with message and status code."""
|
|
error = FastGPTError("Test error", status_code=404)
|
|
|
|
assert error.message == "Test error"
|
|
assert error.status_code == 404
|
|
assert error.response_data == {}
|
|
|
|
def test_init_with_all_parameters(self):
|
|
"""Test initialization with all parameters."""
|
|
response_data = {"code": "test_error", "details": "Something went wrong"}
|
|
error = FastGPTError(
|
|
"Test error",
|
|
status_code=500,
|
|
response_data=response_data
|
|
)
|
|
|
|
assert error.message == "Test error"
|
|
assert error.status_code == 500
|
|
assert error.response_data == response_data
|
|
|
|
def test_str_representation(self):
|
|
"""Test string representation of exception."""
|
|
error = FastGPTError("Test error message")
|
|
|
|
assert str(error) == "Test error message"
|
|
|
|
def test_inheritance(self):
|
|
"""Test that FastGPTError inherits from Exception."""
|
|
error = FastGPTError("Test")
|
|
|
|
assert isinstance(error, Exception)
|
|
assert isinstance(error, FastGPTError)
|
|
|
|
|
|
class TestAPIError:
|
|
"""Test suite for APIError exception."""
|
|
|
|
def test_api_error_inheritance(self):
|
|
"""Test that APIError inherits from FastGPTError."""
|
|
error = APIError("API error occurred")
|
|
|
|
assert isinstance(error, FastGPTError)
|
|
assert isinstance(error, APIError)
|
|
assert isinstance(error, Exception)
|
|
|
|
def test_api_error_basic(self):
|
|
"""Test basic APIError creation."""
|
|
error = APIError("API failed", status_code=500)
|
|
|
|
assert error.message == "API failed"
|
|
assert error.status_code == 500
|
|
|
|
def test_api_error_with_response_data(self):
|
|
"""Test APIError with response data."""
|
|
response_data = {"error": "Internal server error"}
|
|
error = APIError("Server error", status_code=500, response_data=response_data)
|
|
|
|
assert error.response_data == response_data
|
|
|
|
|
|
class TestAuthenticationError:
|
|
"""Test suite for AuthenticationError exception."""
|
|
|
|
def test_auth_error_inheritance(self):
|
|
"""Test that AuthenticationError inherits from FastGPTError."""
|
|
error = AuthenticationError("Invalid credentials")
|
|
|
|
assert isinstance(error, FastGPTError)
|
|
assert isinstance(error, AuthenticationError)
|
|
|
|
def test_auth_error_default(self):
|
|
"""Test AuthenticationError with default parameters."""
|
|
error = AuthenticationError("Invalid API key")
|
|
|
|
assert error.message == "Invalid API key"
|
|
assert error.status_code is None
|
|
assert error.response_data == {}
|
|
|
|
def test_auth_error_with_status_code(self):
|
|
"""Test AuthenticationError with status code."""
|
|
error = AuthenticationError(
|
|
"Authentication failed",
|
|
status_code=401
|
|
)
|
|
|
|
assert error.status_code == 401
|
|
|
|
def test_auth_error_full(self):
|
|
"""Test AuthenticationError with all parameters."""
|
|
response_data = {
|
|
"code": "invalid_api_key",
|
|
"message": "The provided API key is invalid"
|
|
}
|
|
error = AuthenticationError(
|
|
"Auth failed",
|
|
status_code=401,
|
|
response_data=response_data
|
|
)
|
|
|
|
assert error.message == "Auth failed"
|
|
assert error.status_code == 401
|
|
assert error.response_data == response_data
|
|
|
|
|
|
class TestRateLimitError:
|
|
"""Test suite for RateLimitError exception."""
|
|
|
|
def test_rate_limit_error_inheritance(self):
|
|
"""Test that RateLimitError inherits from FastGPTError."""
|
|
error = RateLimitError("Rate limit exceeded")
|
|
|
|
assert isinstance(error, FastGPTError)
|
|
assert isinstance(error, RateLimitError)
|
|
|
|
def test_rate_limit_error_basic(self):
|
|
"""Test RateLimitError with basic parameters."""
|
|
error = RateLimitError("Too many requests")
|
|
|
|
assert error.message == "Too many requests"
|
|
assert error.retry_after is None
|
|
assert error.status_code is None
|
|
|
|
def test_rate_limit_error_with_retry_after(self):
|
|
"""Test RateLimitError with retry_after parameter."""
|
|
error = RateLimitError(
|
|
"Rate limit exceeded",
|
|
retry_after="60"
|
|
)
|
|
|
|
assert error.message == "Rate limit exceeded"
|
|
assert error.retry_after == "60"
|
|
|
|
def test_rate_limit_error_full(self):
|
|
"""Test RateLimitError with all parameters."""
|
|
response_data = {
|
|
"code": "rate_limit_exceeded",
|
|
"limit": 100,
|
|
"remaining": 0
|
|
}
|
|
error = RateLimitError(
|
|
"Rate limit hit",
|
|
retry_after="120",
|
|
status_code=429,
|
|
response_data=response_data
|
|
)
|
|
|
|
assert error.message == "Rate limit hit"
|
|
assert error.retry_after == "120"
|
|
assert error.status_code == 429
|
|
assert error.response_data == response_data
|
|
|
|
def test_rate_limit_error_retry_after_integer(self):
|
|
"""Test RateLimitError with integer retry_after."""
|
|
error = RateLimitError(
|
|
"Rate limited",
|
|
retry_after=30
|
|
)
|
|
|
|
assert error.retry_after == 30
|
|
|
|
|
|
class TestValidationError:
|
|
"""Test suite for ValidationError exception."""
|
|
|
|
def test_validation_error_inheritance(self):
|
|
"""Test that ValidationError inherits from FastGPTError."""
|
|
error = ValidationError("Validation failed")
|
|
|
|
assert isinstance(error, FastGPTError)
|
|
assert isinstance(error, ValidationError)
|
|
|
|
def test_validation_error_basic(self):
|
|
"""Test ValidationError with basic parameters."""
|
|
error = ValidationError("Invalid parameter")
|
|
|
|
assert error.message == "Invalid parameter"
|
|
|
|
def test_validation_error_with_status(self):
|
|
"""Test ValidationError with status code."""
|
|
error = ValidationError(
|
|
"Invalid input",
|
|
status_code=422
|
|
)
|
|
|
|
assert error.status_code == 422
|
|
|
|
def test_validation_error_with_response_data(self):
|
|
"""Test ValidationError with response data."""
|
|
response_data = {
|
|
"code": "validation_error",
|
|
"field": "email",
|
|
"reason": "Invalid format"
|
|
}
|
|
error = ValidationError(
|
|
"Validation failed",
|
|
status_code=422,
|
|
response_data=response_data
|
|
)
|
|
|
|
assert error.response_data == response_data
|
|
|
|
|
|
class TestStreamParseError:
|
|
"""Test suite for StreamParseError exception."""
|
|
|
|
def test_stream_parse_error_inheritance(self):
|
|
"""Test that StreamParseError inherits from FastGPTError."""
|
|
error = StreamParseError("Stream parsing failed")
|
|
|
|
assert isinstance(error, FastGPTError)
|
|
assert isinstance(error, StreamParseError)
|
|
|
|
def test_stream_parse_error_basic(self):
|
|
"""Test StreamParseError with basic parameters."""
|
|
error = StreamParseError("Invalid stream format")
|
|
|
|
assert error.message == "Invalid stream format"
|
|
|
|
def test_stream_parse_error_with_status(self):
|
|
"""Test StreamParseError with status code."""
|
|
error = StreamParseError(
|
|
"Parse error",
|
|
status_code=500
|
|
)
|
|
|
|
assert error.status_code == 500
|
|
|
|
def test_stream_parse_error_with_response_data(self):
|
|
"""Test StreamParseError with response data."""
|
|
response_data = {
|
|
"error": "stream_error",
|
|
"details": "Unexpected EOF"
|
|
}
|
|
error = StreamParseError(
|
|
"Failed to parse stream",
|
|
status_code=502,
|
|
response_data=response_data
|
|
)
|
|
|
|
assert error.response_data == response_data
|
|
|
|
|
|
class TestExceptionUsagePatterns:
|
|
"""Test suite for common exception usage patterns."""
|
|
|
|
def test_catching_base_exception(self):
|
|
"""Test catching exceptions using base FastGPTError."""
|
|
with pytest.raises(FastGPTError) as exc_info:
|
|
raise APIError("API error")
|
|
|
|
assert isinstance(exc_info.value, FastGPTError)
|
|
assert str(exc_info.value) == "API error"
|
|
|
|
def test_catching_specific_exception(self):
|
|
"""Test catching specific exception types."""
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
raise AuthenticationError("Auth failed", status_code=401)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
|
|
def test_exception_chain(self):
|
|
"""Test exception chaining."""
|
|
try:
|
|
try:
|
|
raise ValueError("Original error")
|
|
except ValueError as e:
|
|
raise APIError("Wrapped error") from e
|
|
except APIError as caught:
|
|
assert caught.__cause__ is not None
|
|
assert isinstance(caught.__cause__, ValueError)
|
|
|
|
def test_raising_from_http_response(self):
|
|
"""Test raising exception from HTTP response data."""
|
|
response_data = {
|
|
"code": "invalid_request",
|
|
"message": "The request was invalid"
|
|
}
|
|
|
|
with pytest.raises(APIError) as exc_info:
|
|
raise APIError(
|
|
"Invalid request",
|
|
status_code=400,
|
|
response_data=response_data
|
|
)
|
|
|
|
assert exc_info.value.response_data == response_data
|
|
|
|
def test_exception_with_none_response_data(self):
|
|
"""Test exception when response_data is None."""
|
|
error = APIError("Error", status_code=500, response_data=None)
|
|
|
|
# None should default to empty dict
|
|
assert error.response_data == {}
|
|
|
|
def test_multiple_exception_types(self):
|
|
"""Test handling multiple exception types."""
|
|
def raise_specific_error(status_code):
|
|
if status_code == 401:
|
|
raise AuthenticationError("Unauthorized", status_code=status_code)
|
|
elif status_code == 429:
|
|
raise RateLimitError("Rate limited", status_code=status_code)
|
|
elif status_code == 422:
|
|
raise ValidationError("Invalid data", status_code=status_code)
|
|
else:
|
|
raise APIError("Generic error", status_code=status_code)
|
|
|
|
# Test each type
|
|
with pytest.raises(AuthenticationError):
|
|
raise_specific_error(401)
|
|
|
|
with pytest.raises(RateLimitError):
|
|
raise_specific_error(429)
|
|
|
|
with pytest.raises(ValidationError):
|
|
raise_specific_error(422)
|
|
|
|
with pytest.raises(APIError):
|
|
raise_specific_error(500)
|
|
|
|
def test_exception_message_formatting(self):
|
|
"""Test that exception messages are properly formatted."""
|
|
error1 = FastGPTError("Simple error")
|
|
assert str(error1) == "Simple error"
|
|
|
|
error2 = FastGPTError("Error with status", status_code=404)
|
|
assert str(error2) == "Error with status"
|
|
|
|
# Message should be accessible regardless of other parameters
|
|
error3 = APIError("Complex error", status_code=500, response_data={"key": "value"})
|
|
assert error3.message == "Complex error"
|
|
assert str(error3) == "Complex error"
|