zanewalker
dfaa0aaec4
Covers all algorithms, backends, decorators, middleware, and integration scenarios. Added conftest.py with shared fixtures and pytest-asyncio configuration.
270 lines
9.4 KiB
Python
270 lines
9.4 KiB
Python
"""Tests for exceptions and error handling.
|
|
|
|
Comprehensive tests covering:
|
|
- Exception classes and their attributes
|
|
- Exception inheritance hierarchy
|
|
- Error message formatting
|
|
- Rate limit info in exceptions
|
|
- Configuration validation errors
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from fastapi_traffic import BackendError, ConfigurationError, RateLimitExceeded
|
|
from fastapi_traffic.core.config import RateLimitConfig
|
|
from fastapi_traffic.core.models import RateLimitInfo
|
|
from fastapi_traffic.exceptions import FastAPITrafficError
|
|
|
|
|
|
class TestExceptionHierarchy:
|
|
"""Tests for exception class hierarchy."""
|
|
|
|
def test_all_exceptions_inherit_from_base(self) -> None:
|
|
"""Test that all exceptions inherit from FastAPITrafficError."""
|
|
assert issubclass(RateLimitExceeded, FastAPITrafficError)
|
|
assert issubclass(BackendError, FastAPITrafficError)
|
|
assert issubclass(ConfigurationError, FastAPITrafficError)
|
|
|
|
def test_base_exception_inherits_from_exception(self) -> None:
|
|
"""Test that base exception inherits from Exception."""
|
|
assert issubclass(FastAPITrafficError, Exception)
|
|
|
|
def test_exceptions_are_catchable_as_base(self) -> None:
|
|
"""Test that all exceptions can be caught as base type."""
|
|
try:
|
|
raise RateLimitExceeded("test")
|
|
except FastAPITrafficError:
|
|
pass
|
|
|
|
try:
|
|
raise BackendError("test")
|
|
except FastAPITrafficError:
|
|
pass
|
|
|
|
try:
|
|
raise ConfigurationError("test")
|
|
except FastAPITrafficError:
|
|
pass
|
|
|
|
|
|
class TestRateLimitExceeded:
|
|
"""Tests for RateLimitExceeded exception."""
|
|
|
|
def test_default_message(self) -> None:
|
|
"""Test default error message."""
|
|
exc = RateLimitExceeded()
|
|
assert exc.message == "Rate limit exceeded"
|
|
assert str(exc) == "Rate limit exceeded"
|
|
|
|
def test_custom_message(self) -> None:
|
|
"""Test custom error message."""
|
|
exc = RateLimitExceeded("Custom rate limit message")
|
|
assert exc.message == "Custom rate limit message"
|
|
assert str(exc) == "Custom rate limit message"
|
|
|
|
def test_retry_after_attribute(self) -> None:
|
|
"""Test retry_after attribute."""
|
|
exc = RateLimitExceeded("test", retry_after=30.5)
|
|
assert exc.retry_after == 30.5
|
|
|
|
def test_retry_after_none_by_default(self) -> None:
|
|
"""Test retry_after is None by default."""
|
|
exc = RateLimitExceeded("test")
|
|
assert exc.retry_after is None
|
|
|
|
def test_limit_info_attribute(self) -> None:
|
|
"""Test limit_info attribute."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=0,
|
|
reset_at=1234567890.0,
|
|
retry_after=30.0,
|
|
)
|
|
exc = RateLimitExceeded("test", limit_info=info)
|
|
assert exc.limit_info is not None
|
|
assert exc.limit_info.limit == 100
|
|
assert exc.limit_info.remaining == 0
|
|
|
|
def test_limit_info_none_by_default(self) -> None:
|
|
"""Test limit_info is None by default."""
|
|
exc = RateLimitExceeded("test")
|
|
assert exc.limit_info is None
|
|
|
|
def test_full_exception_construction(self) -> None:
|
|
"""Test constructing exception with all attributes."""
|
|
info = RateLimitInfo(
|
|
limit=50,
|
|
remaining=0,
|
|
reset_at=1234567890.0,
|
|
retry_after=15.0,
|
|
window_size=60.0,
|
|
)
|
|
exc = RateLimitExceeded(
|
|
"API rate limit exceeded",
|
|
retry_after=15.0,
|
|
limit_info=info,
|
|
)
|
|
assert exc.message == "API rate limit exceeded"
|
|
assert exc.retry_after == 15.0
|
|
assert exc.limit_info is not None
|
|
assert exc.limit_info.window_size == 60.0
|
|
|
|
|
|
class TestBackendError:
|
|
"""Tests for BackendError exception."""
|
|
|
|
def test_default_message(self) -> None:
|
|
"""Test default error message."""
|
|
exc = BackendError()
|
|
assert exc.message == "Backend operation failed"
|
|
|
|
def test_custom_message(self) -> None:
|
|
"""Test custom error message."""
|
|
exc = BackendError("Redis connection failed")
|
|
assert exc.message == "Redis connection failed"
|
|
|
|
def test_original_error_attribute(self) -> None:
|
|
"""Test original_error attribute."""
|
|
original = ValueError("Connection refused")
|
|
exc = BackendError("Failed to connect", original_error=original)
|
|
assert exc.original_error is original
|
|
assert isinstance(exc.original_error, ValueError)
|
|
|
|
def test_original_error_none_by_default(self) -> None:
|
|
"""Test original_error is None by default."""
|
|
exc = BackendError("test")
|
|
assert exc.original_error is None
|
|
|
|
def test_chained_exception_handling(self) -> None:
|
|
"""Test that original error can be used for chaining."""
|
|
original = ConnectionError("Network unreachable")
|
|
exc = BackendError("Backend unavailable", original_error=original)
|
|
|
|
assert exc.original_error is not None
|
|
assert str(exc.original_error) == "Network unreachable"
|
|
|
|
|
|
class TestConfigurationError:
|
|
"""Tests for ConfigurationError exception."""
|
|
|
|
def test_basic_construction(self) -> None:
|
|
"""Test basic exception construction."""
|
|
exc = ConfigurationError("Invalid configuration")
|
|
assert str(exc) == "Invalid configuration"
|
|
|
|
def test_inherits_from_base(self) -> None:
|
|
"""Test inheritance from base exception."""
|
|
exc = ConfigurationError("test")
|
|
assert isinstance(exc, FastAPITrafficError)
|
|
assert isinstance(exc, Exception)
|
|
|
|
|
|
class TestRateLimitConfigValidation:
|
|
"""Tests for RateLimitConfig validation errors."""
|
|
|
|
def test_negative_limit_raises_error(self) -> None:
|
|
"""Test that negative limit raises ValueError."""
|
|
with pytest.raises(ValueError, match="limit must be positive"):
|
|
RateLimitConfig(limit=-1, window_size=60.0)
|
|
|
|
def test_zero_limit_raises_error(self) -> None:
|
|
"""Test that zero limit raises ValueError."""
|
|
with pytest.raises(ValueError, match="limit must be positive"):
|
|
RateLimitConfig(limit=0, window_size=60.0)
|
|
|
|
def test_negative_window_size_raises_error(self) -> None:
|
|
"""Test that negative window_size raises ValueError."""
|
|
with pytest.raises(ValueError, match="window_size must be positive"):
|
|
RateLimitConfig(limit=100, window_size=-1.0)
|
|
|
|
def test_zero_window_size_raises_error(self) -> None:
|
|
"""Test that zero window_size raises ValueError."""
|
|
with pytest.raises(ValueError, match="window_size must be positive"):
|
|
RateLimitConfig(limit=100, window_size=0.0)
|
|
|
|
def test_negative_cost_raises_error(self) -> None:
|
|
"""Test that negative cost raises ValueError."""
|
|
with pytest.raises(ValueError, match="cost must be positive"):
|
|
RateLimitConfig(limit=100, window_size=60.0, cost=-1)
|
|
|
|
def test_zero_cost_raises_error(self) -> None:
|
|
"""Test that zero cost raises ValueError."""
|
|
with pytest.raises(ValueError, match="cost must be positive"):
|
|
RateLimitConfig(limit=100, window_size=60.0, cost=0)
|
|
|
|
def test_valid_config_does_not_raise(self) -> None:
|
|
"""Test that valid configuration does not raise."""
|
|
config = RateLimitConfig(limit=100, window_size=60.0, cost=1)
|
|
assert config.limit == 100
|
|
assert config.window_size == 60.0
|
|
assert config.cost == 1
|
|
|
|
|
|
class TestRateLimitInfo:
|
|
"""Tests for RateLimitInfo model."""
|
|
|
|
def test_to_headers_basic(self) -> None:
|
|
"""Test basic header generation."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=50,
|
|
reset_at=1234567890.0,
|
|
)
|
|
headers = info.to_headers()
|
|
assert headers["X-RateLimit-Limit"] == "100"
|
|
assert headers["X-RateLimit-Remaining"] == "50"
|
|
assert headers["X-RateLimit-Reset"] == "1234567890"
|
|
|
|
def test_to_headers_with_retry_after(self) -> None:
|
|
"""Test header generation with retry_after."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=0,
|
|
reset_at=1234567890.0,
|
|
retry_after=30.0,
|
|
)
|
|
headers = info.to_headers()
|
|
assert "Retry-After" in headers
|
|
assert headers["Retry-After"] == "30"
|
|
|
|
def test_to_headers_without_retry_after(self) -> None:
|
|
"""Test header generation without retry_after."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=50,
|
|
reset_at=1234567890.0,
|
|
)
|
|
headers = info.to_headers()
|
|
assert "Retry-After" not in headers
|
|
|
|
def test_remaining_cannot_be_negative_in_headers(self) -> None:
|
|
"""Test that remaining is clamped to 0 in headers."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=-5,
|
|
reset_at=1234567890.0,
|
|
)
|
|
headers = info.to_headers()
|
|
assert headers["X-RateLimit-Remaining"] == "0"
|
|
|
|
def test_frozen_dataclass(self) -> None:
|
|
"""Test that RateLimitInfo is immutable."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=50,
|
|
reset_at=1234567890.0,
|
|
)
|
|
with pytest.raises(AttributeError):
|
|
info.limit = 200 # type: ignore[misc]
|
|
|
|
def test_default_window_size(self) -> None:
|
|
"""Test default window_size value."""
|
|
info = RateLimitInfo(
|
|
limit=100,
|
|
remaining=50,
|
|
reset_at=1234567890.0,
|
|
)
|
|
assert info.window_size == 60.0
|