Add comprehensive unit test suite
- 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>
This commit is contained in:
Xin Wang
220
tests/test_base_client.py
Normal file
220
tests/test_base_client.py
Normal file
@@ -0,0 +1,220 @@
|
||||
"""Tests for BaseClientMixin."""
|
||||
|
||||
import time
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import pytest
|
||||
import httpx
|
||||
|
||||
from fastgpt_client.base_client import BaseClientMixin
|
||||
from fastgpt_client.exceptions import APIError, ValidationError
|
||||
|
||||
|
||||
class TestBaseClientMixin:
|
||||
"""Test suite for BaseClientMixin."""
|
||||
|
||||
def test_init_default_parameters(self, api_key, base_url):
|
||||
"""Test initialization with default parameters."""
|
||||
mixin = BaseClientMixin(api_key, base_url)
|
||||
|
||||
assert mixin.api_key == api_key
|
||||
assert mixin.base_url == base_url
|
||||
assert mixin.timeout == 60.0
|
||||
assert mixin.max_retries == 3
|
||||
assert mixin.retry_delay == 1.0
|
||||
assert mixin.enable_logging is False
|
||||
|
||||
def test_init_custom_parameters(self, api_key, base_url):
|
||||
"""Test initialization with custom parameters."""
|
||||
mixin = BaseClientMixin(
|
||||
api_key,
|
||||
base_url,
|
||||
timeout=120.0,
|
||||
max_retries=5,
|
||||
retry_delay=2.0,
|
||||
enable_logging=True
|
||||
)
|
||||
|
||||
assert mixin.timeout == 120.0
|
||||
assert mixin.max_retries == 5
|
||||
assert mixin.retry_delay == 2.0
|
||||
assert mixin.enable_logging is True
|
||||
|
||||
def test_validate_params_with_valid_params(self, api_key, base_url):
|
||||
"""Test parameter validation with valid parameters."""
|
||||
mixin = BaseClientMixin(api_key, base_url)
|
||||
|
||||
# Should not raise any exception
|
||||
mixin._validate_params(
|
||||
query="test query",
|
||||
chatId="chat-123",
|
||||
appId="app-123",
|
||||
dataId="data-123",
|
||||
content="some content"
|
||||
)
|
||||
|
||||
def test_validate_params_with_empty_strings(self, api_key, base_url):
|
||||
"""Test parameter validation with empty strings."""
|
||||
mixin = BaseClientMixin(api_key, base_url)
|
||||
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
mixin._validate_params(query="")
|
||||
|
||||
assert "query must be a non-empty string" in str(exc_info.value)
|
||||
|
||||
def test_validate_params_with_whitespace_only(self, api_key, base_url):
|
||||
"""Test parameter validation with whitespace-only strings."""
|
||||
mixin = BaseClientMixin(api_key, base_url)
|
||||
|
||||
with pytest.raises(ValidationError) as exc_info:
|
||||
mixin._validate_params(chatId=" ")
|
||||
|
||||
assert "chatId must be a non-empty string" in str(exc_info.value)
|
||||
|
||||
def test_validate_params_with_none_values(self, api_key, base_url):
|
||||
"""Test parameter validation ignores None values."""
|
||||
mixin = BaseClientMixin(api_key, base_url)
|
||||
|
||||
# Should not raise any exception for None values
|
||||
mixin._validate_params(
|
||||
query=None,
|
||||
chatId=None,
|
||||
appId="app-123" # Only one non-None value
|
||||
)
|
||||
|
||||
def test_validate_params_with_empty_collections(self, api_key, base_url):
|
||||
"""Test parameter validation with empty collections."""
|
||||
mixin = BaseClientMixin(
|
||||
api_key,
|
||||
base_url,
|
||||
enable_logging=True
|
||||
)
|
||||
|
||||
# Should not raise exception, just log debug message
|
||||
mixin._validate_params(
|
||||
messages=[],
|
||||
variables={}
|
||||
)
|
||||
|
||||
def test_retry_request_success_on_first_attempt(self, api_key, base_url, mock_response):
|
||||
"""Test retry request succeeds on first attempt."""
|
||||
mixin = BaseClientMixin(api_key, base_url)
|
||||
|
||||
request_func = Mock(return_value=mock_response)
|
||||
result = mixin._retry_request(request_func, "GET /api/test")
|
||||
|
||||
assert result == mock_response
|
||||
assert request_func.call_count == 1
|
||||
|
||||
def test_retry_request_success_on_second_attempt(self, api_key, base_url, mock_response):
|
||||
"""Test retry request succeeds on second attempt after 500 error."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=3, retry_delay=0.01)
|
||||
|
||||
# First call returns 500, second succeeds
|
||||
error_response = Mock(spec=httpx.Response)
|
||||
error_response.status_code = 500
|
||||
|
||||
request_func = Mock(side_effect=[error_response, mock_response])
|
||||
|
||||
start_time = time.time()
|
||||
result = mixin._retry_request(request_func, "GET /api/test")
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
assert result == mock_response
|
||||
assert request_func.call_count == 2
|
||||
# Should have slept at least retry_delay seconds (with some tolerance for test execution)
|
||||
assert elapsed_time >= 0.005 # Allow some tolerance
|
||||
|
||||
def test_retry_request_exhausted_with_server_errors(self, api_key, base_url):
|
||||
"""Test retry request exhausted by server errors."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=2, retry_delay=0.01)
|
||||
|
||||
error_response = Mock(spec=httpx.Response)
|
||||
error_response.status_code = 500
|
||||
request_func = Mock(return_value=error_response)
|
||||
|
||||
with pytest.raises(APIError) as exc_info:
|
||||
mixin._retry_request(request_func, "GET /api/test")
|
||||
|
||||
assert "failed after 2 attempts" in str(exc_info.value)
|
||||
assert request_func.call_count == 2
|
||||
|
||||
def test_retry_request_with_network_exception(self, api_key, base_url):
|
||||
"""Test retry request with network exception."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=2, retry_delay=0.01)
|
||||
|
||||
request_func = Mock(side_effect=httpx.ConnectError("Connection failed"))
|
||||
|
||||
with pytest.raises(APIError) as exc_info:
|
||||
mixin._retry_request(request_func, "GET /api/test")
|
||||
|
||||
assert "failed after 2 attempts" in str(exc_info.value)
|
||||
assert "Connection failed" in str(exc_info.value)
|
||||
assert request_func.call_count == 2
|
||||
|
||||
def test_retry_request_exponential_backoff(self, api_key, base_url, mock_response):
|
||||
"""Test exponential backoff in retry logic."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=4, retry_delay=0.05)
|
||||
|
||||
error_response = Mock(spec=httpx.Response)
|
||||
error_response.status_code = 503
|
||||
|
||||
# First 3 attempts fail, last succeeds
|
||||
request_func = Mock(side_effect=[
|
||||
error_response,
|
||||
error_response,
|
||||
error_response,
|
||||
mock_response
|
||||
])
|
||||
|
||||
start_time = time.time()
|
||||
result = mixin._retry_request(request_func, "GET /api/test")
|
||||
elapsed_time = time.time() - start_time
|
||||
|
||||
assert result == mock_response
|
||||
assert request_func.call_count == 4
|
||||
|
||||
# Exponential backoff: 0.05 + 0.10 + 0.20 = 0.35 seconds minimum
|
||||
# Allow some tolerance for test execution time
|
||||
assert elapsed_time >= 0.25
|
||||
|
||||
def test_retry_request_4xx_no_retry(self, api_key, base_url):
|
||||
"""Test that 4xx errors don't trigger retries."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=3, retry_delay=0.01)
|
||||
|
||||
error_response = Mock(spec=httpx.Response)
|
||||
error_response.status_code = 404
|
||||
request_func = Mock(return_value=error_response)
|
||||
|
||||
result = mixin._retry_request(request_func, "GET /api/test")
|
||||
|
||||
# Should return the 4xx response without retrying
|
||||
assert result == error_response
|
||||
assert request_func.call_count == 1
|
||||
|
||||
def test_retry_request_mixed_exception_and_success(self, api_key, base_url, mock_response):
|
||||
"""Test retry request with exception followed by success."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=3, retry_delay=0.01)
|
||||
|
||||
request_func = Mock(side_effect=[
|
||||
httpx.TimeoutException("Timeout"),
|
||||
mock_response
|
||||
])
|
||||
|
||||
result = mixin._retry_request(request_func, "POST /api/test")
|
||||
|
||||
assert result == mock_response
|
||||
assert request_func.call_count == 2
|
||||
|
||||
def test_retry_request_all_attempts_with_exceptions(self, api_key, base_url):
|
||||
"""Test retry request when all attempts raise exceptions."""
|
||||
mixin = BaseClientMixin(api_key, base_url, max_retries=2, retry_delay=0.01)
|
||||
|
||||
request_func = Mock(side_effect=httpx.TimeoutException("Timeout"))
|
||||
|
||||
with pytest.raises(APIError) as exc_info:
|
||||
mixin._retry_request(request_func, "POST /api/test")
|
||||
|
||||
assert "failed after 2 attempts" in str(exc_info.value)
|
||||
assert "Timeout" in str(exc_info.value)
|
||||
assert request_func.call_count == 2
|
||||
Reference in New Issue
Block a user