zanewalker
fb23e3c7cf
- Add ConfigLoader class for loading RateLimitConfig and GlobalConfig - Support .env files with FASTAPI_TRAFFIC_* prefixed variables - Support JSON configuration files with type validation - Add convenience functions: load_rate_limit_config, load_global_config - Add load_rate_limit_config_from_env, load_global_config_from_env - Support custom environment variable prefixes - Add comprehensive error handling with ConfigurationError - Add 47 tests for configuration loading - Add example 11_config_loader.py with 9 usage patterns - Update examples/README.md with config loader documentation - Update CHANGELOG.md with new feature - Fix typo in limiter.py (errant 'fi' on line 4)
600 lines
22 KiB
Python
600 lines
22 KiB
Python
"""Tests for configuration loader."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from typing import TYPE_CHECKING
|
|
|
|
import pytest
|
|
|
|
from fastapi_traffic.core.algorithms import Algorithm
|
|
from fastapi_traffic.core.config import GlobalConfig
|
|
from fastapi_traffic.core.config_loader import (
|
|
ConfigLoader,
|
|
load_global_config,
|
|
load_global_config_from_env,
|
|
load_rate_limit_config,
|
|
load_rate_limit_config_from_env,
|
|
)
|
|
from fastapi_traffic.exceptions import ConfigurationError
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Generator
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir() -> Generator[Path, None, None]:
|
|
"""Create a temporary directory for test files."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
|
|
@pytest.fixture
|
|
def loader() -> ConfigLoader:
|
|
"""Create a ConfigLoader instance."""
|
|
return ConfigLoader()
|
|
|
|
|
|
class TestConfigLoaderEnv:
|
|
"""Tests for loading configuration from environment variables."""
|
|
|
|
def test_load_rate_limit_config_from_env(self, loader: ConfigLoader) -> None:
|
|
"""Test loading RateLimitConfig from environment variables."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "100",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_WINDOW_SIZE": "60.0",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_ALGORITHM": "token_bucket",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_KEY_PREFIX": "test",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_INCLUDE_HEADERS": "true",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_STATUS_CODE": "429",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_SKIP_ON_ERROR": "false",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_COST": "1",
|
|
}
|
|
|
|
config = loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
assert config.limit == 100
|
|
assert config.window_size == 60.0
|
|
assert config.algorithm == Algorithm.TOKEN_BUCKET
|
|
assert config.key_prefix == "test"
|
|
assert config.include_headers is True
|
|
assert config.status_code == 429
|
|
assert config.skip_on_error is False
|
|
assert config.cost == 1
|
|
|
|
def test_load_rate_limit_config_from_env_with_burst_size(
|
|
self, loader: ConfigLoader
|
|
) -> None:
|
|
"""Test loading RateLimitConfig with burst_size."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "50",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_BURST_SIZE": "100",
|
|
}
|
|
|
|
config = loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
assert config.limit == 50
|
|
assert config.burst_size == 100
|
|
|
|
def test_load_rate_limit_config_from_env_missing_limit(
|
|
self, loader: ConfigLoader
|
|
) -> None:
|
|
"""Test that missing 'limit' field raises ConfigurationError."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_WINDOW_SIZE": "60.0",
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError, match="Required field 'limit'"):
|
|
loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
def test_load_rate_limit_config_from_env_with_overrides(
|
|
self, loader: ConfigLoader
|
|
) -> None:
|
|
"""Test loading with overrides."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "100",
|
|
}
|
|
|
|
config = loader.load_rate_limit_config_from_env(
|
|
env_vars, window_size=120.0, error_message="Custom error"
|
|
)
|
|
|
|
assert config.limit == 100
|
|
assert config.window_size == 120.0
|
|
assert config.error_message == "Custom error"
|
|
|
|
def test_load_global_config_from_env(self, loader: ConfigLoader) -> None:
|
|
"""Test loading GlobalConfig from environment variables."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_GLOBAL_ENABLED": "true",
|
|
"FASTAPI_TRAFFIC_GLOBAL_DEFAULT_LIMIT": "200",
|
|
"FASTAPI_TRAFFIC_GLOBAL_DEFAULT_WINDOW_SIZE": "120.0",
|
|
"FASTAPI_TRAFFIC_GLOBAL_DEFAULT_ALGORITHM": "fixed_window",
|
|
"FASTAPI_TRAFFIC_GLOBAL_KEY_PREFIX": "global_test",
|
|
"FASTAPI_TRAFFIC_GLOBAL_INCLUDE_HEADERS": "false",
|
|
"FASTAPI_TRAFFIC_GLOBAL_STATUS_CODE": "503",
|
|
"FASTAPI_TRAFFIC_GLOBAL_SKIP_ON_ERROR": "true",
|
|
"FASTAPI_TRAFFIC_GLOBAL_HEADERS_PREFIX": "X-Custom",
|
|
}
|
|
|
|
config = loader.load_global_config_from_env(env_vars)
|
|
|
|
assert config.enabled is True
|
|
assert config.default_limit == 200
|
|
assert config.default_window_size == 120.0
|
|
assert config.default_algorithm == Algorithm.FIXED_WINDOW
|
|
assert config.key_prefix == "global_test"
|
|
assert config.include_headers is False
|
|
assert config.status_code == 503
|
|
assert config.skip_on_error is True
|
|
assert config.headers_prefix == "X-Custom"
|
|
|
|
def test_load_global_config_from_env_with_sets(self, loader: ConfigLoader) -> None:
|
|
"""Test loading GlobalConfig with set fields."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_GLOBAL_EXEMPT_IPS": "127.0.0.1, 192.168.1.1, 10.0.0.1",
|
|
"FASTAPI_TRAFFIC_GLOBAL_EXEMPT_PATHS": "/health, /metrics",
|
|
}
|
|
|
|
config = loader.load_global_config_from_env(env_vars)
|
|
|
|
assert config.exempt_ips == {"127.0.0.1", "192.168.1.1", "10.0.0.1"}
|
|
assert config.exempt_paths == {"/health", "/metrics"}
|
|
|
|
def test_load_global_config_from_env_empty_sets(
|
|
self, loader: ConfigLoader
|
|
) -> None:
|
|
"""Test loading GlobalConfig with empty set fields."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_GLOBAL_EXEMPT_IPS": "",
|
|
"FASTAPI_TRAFFIC_GLOBAL_EXEMPT_PATHS": "",
|
|
}
|
|
|
|
config = loader.load_global_config_from_env(env_vars)
|
|
|
|
assert config.exempt_ips == set()
|
|
assert config.exempt_paths == set()
|
|
|
|
def test_load_global_config_defaults(self, loader: ConfigLoader) -> None:
|
|
"""Test that GlobalConfig uses defaults when no env vars set."""
|
|
config = loader.load_global_config_from_env({})
|
|
|
|
assert config.enabled is True
|
|
assert config.default_limit == 100
|
|
assert config.default_window_size == 60.0
|
|
|
|
|
|
class TestConfigLoaderDotenv:
|
|
"""Tests for loading configuration from .env files."""
|
|
|
|
def test_load_rate_limit_config_from_dotenv(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test loading RateLimitConfig from .env file."""
|
|
env_file = temp_dir / ".env"
|
|
env_file.write_text(
|
|
"""
|
|
# Rate limit configuration
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT=150
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_WINDOW_SIZE=30.0
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_ALGORITHM=sliding_window
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_KEY_PREFIX=api
|
|
"""
|
|
)
|
|
|
|
config = loader.load_rate_limit_config_from_dotenv(env_file)
|
|
|
|
assert config.limit == 150
|
|
assert config.window_size == 30.0
|
|
assert config.algorithm == Algorithm.SLIDING_WINDOW
|
|
assert config.key_prefix == "api"
|
|
|
|
def test_load_rate_limit_config_from_dotenv_with_quotes(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test loading config with quoted values."""
|
|
env_file = temp_dir / ".env"
|
|
env_file.write_text(
|
|
"""
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT=100
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_ERROR_MESSAGE="Custom error message"
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_KEY_PREFIX='quoted_prefix'
|
|
"""
|
|
)
|
|
|
|
config = loader.load_rate_limit_config_from_dotenv(env_file)
|
|
|
|
assert config.limit == 100
|
|
assert config.error_message == "Custom error message"
|
|
assert config.key_prefix == "quoted_prefix"
|
|
|
|
def test_load_global_config_from_dotenv(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test loading GlobalConfig from .env file."""
|
|
env_file = temp_dir / ".env"
|
|
env_file.write_text(
|
|
"""
|
|
FASTAPI_TRAFFIC_GLOBAL_ENABLED=false
|
|
FASTAPI_TRAFFIC_GLOBAL_DEFAULT_LIMIT=500
|
|
FASTAPI_TRAFFIC_GLOBAL_EXEMPT_IPS=10.0.0.1,10.0.0.2
|
|
"""
|
|
)
|
|
|
|
config = loader.load_global_config_from_dotenv(env_file)
|
|
|
|
assert config.enabled is False
|
|
assert config.default_limit == 500
|
|
assert config.exempt_ips == {"10.0.0.1", "10.0.0.2"}
|
|
|
|
def test_load_from_dotenv_file_not_found(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that missing file raises ConfigurationError."""
|
|
with pytest.raises(ConfigurationError, match="not found"):
|
|
loader.load_rate_limit_config_from_dotenv(temp_dir / "nonexistent.env")
|
|
|
|
def test_load_from_dotenv_invalid_line(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that invalid line format raises ConfigurationError."""
|
|
env_file = temp_dir / ".env"
|
|
env_file.write_text(
|
|
"""
|
|
FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT=100
|
|
invalid line without equals
|
|
"""
|
|
)
|
|
|
|
with pytest.raises(ConfigurationError, match="missing '='"):
|
|
loader.load_rate_limit_config_from_dotenv(env_file)
|
|
|
|
|
|
class TestConfigLoaderJson:
|
|
"""Tests for loading configuration from JSON files."""
|
|
|
|
def test_load_rate_limit_config_from_json(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test loading RateLimitConfig from JSON file."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {
|
|
"limit": 200,
|
|
"window_size": 45.0,
|
|
"algorithm": "leaky_bucket",
|
|
"key_prefix": "json_test",
|
|
"include_headers": False,
|
|
"status_code": 429,
|
|
"cost": 2,
|
|
}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
config = loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
assert config.limit == 200
|
|
assert config.window_size == 45.0
|
|
assert config.algorithm == Algorithm.LEAKY_BUCKET
|
|
assert config.key_prefix == "json_test"
|
|
assert config.include_headers is False
|
|
assert config.status_code == 429
|
|
assert config.cost == 2
|
|
|
|
def test_load_rate_limit_config_from_json_with_int_window(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that integer window_size is converted to float."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 100, "window_size": 60}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
config = loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
assert config.window_size == 60.0
|
|
assert isinstance(config.window_size, float)
|
|
|
|
def test_load_global_config_from_json(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test loading GlobalConfig from JSON file."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {
|
|
"enabled": True,
|
|
"default_limit": 1000,
|
|
"default_window_size": 300.0,
|
|
"default_algorithm": "sliding_window_counter",
|
|
"exempt_ips": ["127.0.0.1", "::1"],
|
|
"exempt_paths": ["/health", "/ready", "/metrics"],
|
|
}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
config = loader.load_global_config_from_json(json_file)
|
|
|
|
assert config.enabled is True
|
|
assert config.default_limit == 1000
|
|
assert config.default_window_size == 300.0
|
|
assert config.default_algorithm == Algorithm.SLIDING_WINDOW_COUNTER
|
|
assert config.exempt_ips == {"127.0.0.1", "::1"}
|
|
assert config.exempt_paths == {"/health", "/ready", "/metrics"}
|
|
|
|
def test_load_from_json_file_not_found(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that missing file raises ConfigurationError."""
|
|
with pytest.raises(ConfigurationError, match="not found"):
|
|
loader.load_rate_limit_config_from_json(temp_dir / "nonexistent.json")
|
|
|
|
def test_load_from_json_invalid_json(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that invalid JSON raises ConfigurationError."""
|
|
json_file = temp_dir / "config.json"
|
|
json_file.write_text("{ invalid json }")
|
|
|
|
with pytest.raises(ConfigurationError, match="Invalid JSON"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_load_from_json_non_object_root(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that non-object JSON root raises ConfigurationError."""
|
|
json_file = temp_dir / "config.json"
|
|
json_file.write_text("[1, 2, 3]")
|
|
|
|
with pytest.raises(ConfigurationError, match="JSON root must be an object"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_load_from_json_missing_limit(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that missing 'limit' field raises ConfigurationError."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"window_size": 60.0}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ConfigurationError, match="Required field 'limit'"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
|
|
class TestConfigLoaderValidation:
|
|
"""Tests for configuration validation."""
|
|
|
|
def test_invalid_algorithm_value(self, loader: ConfigLoader) -> None:
|
|
"""Test that invalid algorithm raises ConfigurationError."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "100",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_ALGORITHM": "invalid_algorithm",
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError, match="Cannot parse value"):
|
|
loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
def test_invalid_int_value(self, loader: ConfigLoader) -> None:
|
|
"""Test that invalid integer raises ConfigurationError."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "not_a_number",
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError, match="Cannot parse value"):
|
|
loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
def test_invalid_float_value(self, loader: ConfigLoader) -> None:
|
|
"""Test that invalid float raises ConfigurationError."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "100",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_WINDOW_SIZE": "not_a_float",
|
|
}
|
|
|
|
with pytest.raises(ConfigurationError, match="Cannot parse value"):
|
|
loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
def test_unknown_field(self, loader: ConfigLoader, temp_dir: Path) -> None:
|
|
"""Test that unknown field raises ConfigurationError."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 100, "unknown_field": "value"}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ConfigurationError, match="Unknown configuration field"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_non_loadable_field(self, loader: ConfigLoader, temp_dir: Path) -> None:
|
|
"""Test that non-loadable field raises ConfigurationError."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 100, "key_extractor": "some_function"}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ConfigurationError, match="cannot be loaded"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_invalid_type_in_json(self, loader: ConfigLoader, temp_dir: Path) -> None:
|
|
"""Test that invalid type in JSON raises ConfigurationError."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": "not_an_int"}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ConfigurationError, match="Cannot parse value"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_bool_parsing_variations(self, loader: ConfigLoader) -> None:
|
|
"""Test various boolean string representations."""
|
|
for true_val in ["true", "True", "TRUE", "1", "yes", "Yes", "on", "ON"]:
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_GLOBAL_ENABLED": true_val,
|
|
}
|
|
config = loader.load_global_config_from_env(env_vars)
|
|
assert config.enabled is True, f"Failed for value: {true_val}"
|
|
|
|
for false_val in ["false", "False", "FALSE", "0", "no", "No", "off", "OFF"]:
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_GLOBAL_ENABLED": false_val,
|
|
}
|
|
config = loader.load_global_config_from_env(env_vars)
|
|
assert config.enabled is False, f"Failed for value: {false_val}"
|
|
|
|
|
|
class TestConvenienceFunctions:
|
|
"""Tests for convenience functions."""
|
|
|
|
def test_load_rate_limit_config_json(self, temp_dir: Path) -> None:
|
|
"""Test load_rate_limit_config with JSON file."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 100}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
config = load_rate_limit_config(json_file)
|
|
|
|
assert config.limit == 100
|
|
|
|
def test_load_rate_limit_config_dotenv(self, temp_dir: Path) -> None:
|
|
"""Test load_rate_limit_config with .env file."""
|
|
env_file = temp_dir / ".env"
|
|
env_file.write_text("FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT=200")
|
|
|
|
config = load_rate_limit_config(env_file)
|
|
|
|
assert config.limit == 200
|
|
|
|
def test_load_rate_limit_config_env_suffix(self, temp_dir: Path) -> None:
|
|
"""Test load_rate_limit_config with .env suffix."""
|
|
env_file = temp_dir / "custom.env"
|
|
env_file.write_text("FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT=300")
|
|
|
|
config = load_rate_limit_config(env_file)
|
|
|
|
assert config.limit == 300
|
|
|
|
def test_load_global_config_json(self, temp_dir: Path) -> None:
|
|
"""Test load_global_config with JSON file."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"default_limit": 500}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
config = load_global_config(json_file)
|
|
|
|
assert config.default_limit == 500
|
|
|
|
def test_load_global_config_dotenv(self, temp_dir: Path) -> None:
|
|
"""Test load_global_config with .env file."""
|
|
env_file = temp_dir / ".env"
|
|
env_file.write_text("FASTAPI_TRAFFIC_GLOBAL_DEFAULT_LIMIT=600")
|
|
|
|
config = load_global_config(env_file)
|
|
|
|
assert config.default_limit == 600
|
|
|
|
def test_load_config_unknown_format(self, temp_dir: Path) -> None:
|
|
"""Test that unknown file format raises ConfigurationError."""
|
|
unknown_file = temp_dir / "config.yaml"
|
|
unknown_file.write_text("limit: 100")
|
|
|
|
with pytest.raises(ConfigurationError, match="Unknown configuration file"):
|
|
load_rate_limit_config(unknown_file)
|
|
|
|
def test_load_rate_limit_config_from_env_function(self) -> None:
|
|
"""Test load_rate_limit_config_from_env convenience function."""
|
|
# This will use defaults since no env vars are set
|
|
# We need to provide the limit as an override
|
|
config = load_rate_limit_config_from_env(limit=100)
|
|
|
|
assert config.limit == 100
|
|
|
|
def test_load_global_config_from_env_function(self) -> None:
|
|
"""Test load_global_config_from_env convenience function."""
|
|
config = load_global_config_from_env()
|
|
|
|
assert isinstance(config, GlobalConfig)
|
|
assert config.enabled is True
|
|
|
|
|
|
class TestCustomEnvPrefix:
|
|
"""Tests for custom environment variable prefix."""
|
|
|
|
def test_custom_prefix(self) -> None:
|
|
"""Test loading with custom environment prefix."""
|
|
loader = ConfigLoader(env_prefix="CUSTOM_")
|
|
env_vars = {
|
|
"CUSTOM_RATE_LIMIT_LIMIT": "100",
|
|
"CUSTOM_RATE_LIMIT_WINDOW_SIZE": "30.0",
|
|
}
|
|
|
|
config = loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
assert config.limit == 100
|
|
assert config.window_size == 30.0
|
|
|
|
def test_custom_prefix_global(self) -> None:
|
|
"""Test loading GlobalConfig with custom prefix."""
|
|
loader = ConfigLoader(env_prefix="MY_APP_")
|
|
env_vars = {
|
|
"MY_APP_GLOBAL_ENABLED": "false",
|
|
"MY_APP_GLOBAL_DEFAULT_LIMIT": "250",
|
|
}
|
|
|
|
config = loader.load_global_config_from_env(env_vars)
|
|
|
|
assert config.enabled is False
|
|
assert config.default_limit == 250
|
|
|
|
|
|
class TestAllAlgorithms:
|
|
"""Tests for all algorithm types."""
|
|
|
|
@pytest.mark.parametrize(
|
|
"algorithm_str,expected",
|
|
[
|
|
("token_bucket", Algorithm.TOKEN_BUCKET),
|
|
("sliding_window", Algorithm.SLIDING_WINDOW),
|
|
("fixed_window", Algorithm.FIXED_WINDOW),
|
|
("leaky_bucket", Algorithm.LEAKY_BUCKET),
|
|
("sliding_window_counter", Algorithm.SLIDING_WINDOW_COUNTER),
|
|
("TOKEN_BUCKET", Algorithm.TOKEN_BUCKET),
|
|
("SLIDING_WINDOW", Algorithm.SLIDING_WINDOW),
|
|
],
|
|
)
|
|
def test_algorithm_parsing(
|
|
self, loader: ConfigLoader, algorithm_str: str, expected: Algorithm
|
|
) -> None:
|
|
"""Test that all algorithm values are parsed correctly."""
|
|
env_vars = {
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_LIMIT": "100",
|
|
"FASTAPI_TRAFFIC_RATE_LIMIT_ALGORITHM": algorithm_str,
|
|
}
|
|
|
|
config = loader.load_rate_limit_config_from_env(env_vars)
|
|
|
|
assert config.algorithm == expected
|
|
|
|
|
|
class TestDataclassValidation:
|
|
"""Tests that dataclass validation still works after loading."""
|
|
|
|
def test_invalid_limit_value(self, loader: ConfigLoader, temp_dir: Path) -> None:
|
|
"""Test that invalid limit value is caught by dataclass validation."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 0}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ValueError, match="limit must be positive"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_invalid_window_size_value(
|
|
self, loader: ConfigLoader, temp_dir: Path
|
|
) -> None:
|
|
"""Test that invalid window_size is caught by dataclass validation."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 100, "window_size": -1.0}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ValueError, match="window_size must be positive"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|
|
|
|
def test_invalid_cost_value(self, loader: ConfigLoader, temp_dir: Path) -> None:
|
|
"""Test that invalid cost is caught by dataclass validation."""
|
|
json_file = temp_dir / "config.json"
|
|
config_data = {"limit": 100, "cost": 0}
|
|
json_file.write_text(json.dumps(config_data))
|
|
|
|
with pytest.raises(ValueError, match="cost must be positive"):
|
|
loader.load_rate_limit_config_from_json(json_file)
|