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>
388 lines
14 KiB
Python
388 lines
14 KiB
Python
"""Tests for FastGPTClient."""
|
|
|
|
from unittest.mock import Mock, MagicMock, patch
|
|
import pytest
|
|
|
|
import httpx
|
|
|
|
from fastgpt_client.client import FastGPTClient
|
|
from fastgpt_client.exceptions import (
|
|
APIError,
|
|
AuthenticationError,
|
|
RateLimitError,
|
|
ValidationError,
|
|
)
|
|
|
|
|
|
class TestFastGPTClient:
|
|
"""Test suite for FastGPTClient."""
|
|
|
|
def test_init_default(self, api_key, base_url):
|
|
"""Test client initialization with defaults."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
assert client.api_key == api_key
|
|
assert client.base_url == base_url
|
|
assert client.timeout == 60.0
|
|
assert client.max_retries == 3
|
|
assert client.retry_delay == 1.0
|
|
assert client.enable_logging is False
|
|
assert isinstance(client._client, httpx.Client)
|
|
|
|
def test_init_custom_parameters(self, api_key):
|
|
"""Test client initialization with custom parameters."""
|
|
custom_url = "https://api.fastgpt.com"
|
|
client = FastGPTClient(
|
|
api_key,
|
|
base_url=custom_url,
|
|
timeout=120.0,
|
|
max_retries=5,
|
|
retry_delay=2.0,
|
|
enable_logging=True
|
|
)
|
|
|
|
assert client.api_key == api_key
|
|
assert client.base_url == custom_url
|
|
assert client.timeout == 120.0
|
|
assert client.max_retries == 5
|
|
assert client.retry_delay == 2.0
|
|
assert client.enable_logging is True
|
|
|
|
def test_context_manager(self, api_key):
|
|
"""Test using client as context manager."""
|
|
with FastGPTClient(api_key) as client:
|
|
assert client.api_key == api_key
|
|
assert not client._client.is_closed
|
|
|
|
# Client should be closed after exiting context
|
|
assert client._client.is_closed
|
|
|
|
def test_context_manager_with_exception(self, api_key):
|
|
"""Test context manager properly closes on exception."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
try:
|
|
with client:
|
|
raise ValueError("Test exception")
|
|
except ValueError:
|
|
pass
|
|
|
|
# Client should still be closed even with exception
|
|
assert client._client.is_closed
|
|
|
|
def test_close(self, api_key):
|
|
"""Test closing the client."""
|
|
client = FastGPTClient(api_key)
|
|
assert not client._client.is_closed
|
|
|
|
client.close()
|
|
assert client._client.is_closed
|
|
|
|
def test_close_idempotent(self, api_key):
|
|
"""Test that close can be called multiple times safely."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
client.close()
|
|
client.close() # Should not raise an exception
|
|
assert client._client.is_closed
|
|
|
|
|
|
class TestFastGPTClientSendRequest:
|
|
"""Test suite for FastGPTClient._send_request method."""
|
|
|
|
def test_send_request_get_success(self, api_key, mock_response):
|
|
"""Test successful GET request."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
response = client._send_request("GET", "/api/test")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_send_request_post_success(self, api_key, mock_response):
|
|
"""Test successful POST request with JSON body."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response) as mock_request:
|
|
response = client._send_request(
|
|
"POST",
|
|
"/api/test",
|
|
json={"key": "value"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
mock_request.assert_called_once()
|
|
call_kwargs = mock_request.call_args.kwargs
|
|
assert call_kwargs['json'] == {"key": "value"}
|
|
assert call_kwargs['headers']['Authorization'] == f"Bearer {api_key}"
|
|
|
|
def test_send_request_with_params(self, api_key, mock_response):
|
|
"""Test request with query parameters."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response) as mock_request:
|
|
response = client._send_request(
|
|
"GET",
|
|
"/api/test",
|
|
params={"page": 1, "limit": 10}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
call_kwargs = mock_request.call_args.kwargs
|
|
assert call_kwargs['params'] == {"page": 1, "limit": 10}
|
|
|
|
def test_send_request_with_streaming(self, api_key, mock_stream_response):
|
|
"""Test streaming request."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_stream_context = MagicMock()
|
|
mock_stream_context.__enter__ = Mock(return_value=mock_stream_response)
|
|
mock_stream_context.__exit__ = Mock(return_value=None)
|
|
|
|
with patch.object(client._client, 'stream', return_value=mock_stream_context):
|
|
response = client._send_request(
|
|
"POST",
|
|
"/api/test/stream",
|
|
stream=True
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
# Verify stream context was entered
|
|
mock_stream_context.__enter__.assert_called_once()
|
|
# Verify response has custom close method
|
|
assert hasattr(response, 'close')
|
|
|
|
def test_send_request_stream_cleanup(self, api_key):
|
|
"""Test that streaming response cleanup works correctly."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 200
|
|
|
|
# Track if close was called on the response
|
|
original_close_called = []
|
|
|
|
def original_close():
|
|
original_close_called.append(True)
|
|
|
|
mock_response.close = original_close
|
|
|
|
mock_stream_context = MagicMock()
|
|
mock_stream_context.__enter__ = Mock(return_value=mock_response)
|
|
mock_stream_context.__exit__ = Mock(return_value=None)
|
|
|
|
with patch.object(client._client, 'stream', return_value=mock_stream_context):
|
|
response = client._send_request("POST", "/api/stream", stream=True)
|
|
response.close()
|
|
|
|
# Verify stream context exit was called (this is the key behavior)
|
|
mock_stream_context.__exit__.assert_called_once_with(None, None, None)
|
|
# Verify the original close was called (through the wrapper)
|
|
assert len(original_close_called) == 1
|
|
|
|
def test_send_request_authentication_error(self, api_key, error_responses):
|
|
"""Test handling of 401 authentication error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 401
|
|
mock_response.json = Mock(return_value=error_responses['authentication_error'])
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
client._send_request("GET", "/api/test")
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert "Invalid API key" in exc_info.value.message
|
|
|
|
def test_send_request_rate_limit_error(self, api_key, error_responses):
|
|
"""Test handling of 429 rate limit error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 429
|
|
mock_response.headers = {"Retry-After": "60"}
|
|
mock_response.json = Mock(return_value=error_responses['rate_limit_error'])
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
with pytest.raises(RateLimitError) as exc_info:
|
|
client._send_request("GET", "/api/test")
|
|
|
|
assert exc_info.value.status_code == 429
|
|
assert exc_info.value.retry_after == "60"
|
|
|
|
def test_send_request_validation_error(self, api_key, error_responses):
|
|
"""Test handling of 422 validation error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 422
|
|
mock_response.json = Mock(return_value=error_responses['validation_error'])
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
client._send_request("GET", "/api/test")
|
|
|
|
assert exc_info.value.status_code == 422
|
|
assert "Invalid parameters" in exc_info.value.message
|
|
|
|
def test_send_request_generic_api_error(self, api_key, error_responses):
|
|
"""Test handling of generic 4xx API error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 404
|
|
mock_response.json = Mock(return_value={"message": "Not found"})
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._send_request("GET", "/api/test")
|
|
|
|
assert exc_info.value.status_code == 404
|
|
|
|
def test_send_request_with_invalid_json_error_response(self, api_key):
|
|
"""Test handling of error response with invalid JSON."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
# Use 4xx to avoid retry logic (5xx triggers retries)
|
|
mock_response.status_code = 400
|
|
mock_response.json = Mock(side_effect=ValueError("Invalid JSON"))
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
# Should handle the error and raise APIError with status code
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._send_request("GET", "/api/test")
|
|
|
|
assert exc_info.value.status_code == 400
|
|
assert "HTTP 400" in exc_info.value.message
|
|
|
|
def test_send_request_retry_on_server_error(self, api_key, mock_response):
|
|
"""Test that request is retried on 5xx errors."""
|
|
client = FastGPTClient(api_key, max_retries=2, retry_delay=0.01)
|
|
|
|
error_response = Mock(spec=httpx.Response)
|
|
error_response.status_code = 503
|
|
|
|
with patch.object(client._client, 'request', side_effect=[error_response, mock_response]):
|
|
response = client._send_request("GET", "/api/test")
|
|
|
|
assert response.status_code == 200
|
|
|
|
def test_send_request_validation_of_json_params(self, api_key, mock_response):
|
|
"""Test that JSON parameters are validated."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
with pytest.raises(ValidationError):
|
|
client._send_request(
|
|
"POST",
|
|
"/api/test",
|
|
json={"query": ""} # Empty string should fail validation
|
|
)
|
|
|
|
def test_send_request_validation_of_query_params(self, api_key, mock_response):
|
|
"""Test that query parameters are validated."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
with patch.object(client._client, 'request', return_value=mock_response):
|
|
with pytest.raises(ValidationError):
|
|
client._send_request(
|
|
"GET",
|
|
"/api/test",
|
|
params={"chatId": " "} # Whitespace-only should fail
|
|
)
|
|
|
|
|
|
class TestFastGPTClientHandleErrorResponse:
|
|
"""Test suite for FastGPTClient._handle_error_response method."""
|
|
|
|
def test_handle_success_response(self, api_key, mock_response):
|
|
"""Test handling of successful response (no error)."""
|
|
client = FastGPTClient(api_key)
|
|
mock_response.status_code = 200
|
|
|
|
# Should not raise any exception
|
|
client._handle_error_response(mock_response)
|
|
|
|
def test_handle_401_error(self, api_key, error_responses):
|
|
"""Test handling of 401 error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 401
|
|
mock_response.json = Mock(return_value=error_responses['authentication_error'])
|
|
|
|
with pytest.raises(AuthenticationError) as exc_info:
|
|
client._handle_error_response(mock_response)
|
|
|
|
assert exc_info.value.status_code == 401
|
|
assert exc_info.value.message == "Invalid API key"
|
|
|
|
def test_handle_429_error_without_retry_after(self, api_key, error_responses):
|
|
"""Test handling of 429 error without Retry-After header."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 429
|
|
mock_response.headers = {}
|
|
mock_response.json = Mock(return_value=error_responses['rate_limit_error'])
|
|
|
|
with pytest.raises(RateLimitError) as exc_info:
|
|
client._handle_error_response(mock_response)
|
|
|
|
assert exc_info.value.status_code == 429
|
|
assert exc_info.value.retry_after is None
|
|
|
|
def test_handle_422_error(self, api_key, error_responses):
|
|
"""Test handling of 422 validation error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 422
|
|
mock_response.json = Mock(return_value=error_responses['validation_error'])
|
|
|
|
with pytest.raises(ValidationError) as exc_info:
|
|
client._handle_error_response(mock_response)
|
|
|
|
assert exc_info.value.status_code == 422
|
|
|
|
def test_handle_500_error(self, api_key):
|
|
"""Test handling of 500 server error."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 500
|
|
mock_response.json = Mock(return_value={"message": "Internal error"})
|
|
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._handle_error_response(mock_response)
|
|
|
|
assert exc_info.value.status_code == 500
|
|
|
|
def test_handle_error_without_message(self, api_key):
|
|
"""Test handling of error response without message field."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 400
|
|
mock_response.json = Mock(return_value={})
|
|
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._handle_error_response(mock_response)
|
|
|
|
assert "HTTP 400" in exc_info.value.message
|
|
|
|
def test_handle_error_with_json_parse_error(self, api_key):
|
|
"""Test handling of error response that can't be parsed as JSON."""
|
|
client = FastGPTClient(api_key)
|
|
|
|
mock_response = Mock(spec=httpx.Response)
|
|
mock_response.status_code = 503
|
|
mock_response.json = Mock(side_effect=ValueError("Invalid JSON"))
|
|
|
|
with pytest.raises(APIError) as exc_info:
|
|
client._handle_error_response(mock_response)
|
|
|
|
assert "HTTP 503" in exc_info.value.message
|