busya
54a0317e98
- executors: 13 tests, 100% coverage (init/shutdown/run_blocking/run_cpu_blocking) - fileio_utils: 33 tests, 45% coverage (sanitize_filename, get_filename_from_headers, calculate_crc32, create_temp_file, remove_empty_directories, create_dashboard_export, consolidate_archive_folders) - client_registry: 14 tests, 92% coverage (get_client, get_superset_client, get_semaphore, get_auth_lock, shutdown) - cot_logger: 24 tests, 100% coverage (seed/set/get trace_id, push/pop span, structured log with markers, MarkerLogger proxy) - encryption: 12 tests, 100% coverage (key validation, encrypt/decrypt cycle) - rate_limiter: 9 tests, 100% coverage (ban logic, window pruning, per-IP isolation) - auth/logger: 10 tests, 100% coverage (_mask_details, log_security_event)
93 lines
3.8 KiB
Python
93 lines
3.8 KiB
Python
# #region Test.RateLimiter [C:3] [TYPE Module] [SEMANTICS test,rate_limit,auth,throttle]
|
|
# @BRIEF Tests for core/rate_limiter.py — in-memory rate limiter, ban/window logic.
|
|
# @RELATION BINDS_TO -> [RateLimiterModule]
|
|
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
import pytest
|
|
|
|
|
|
class TestRateLimiter:
|
|
"""RateLimiter — in-memory sliding-window rate limiter."""
|
|
|
|
@pytest.fixture
|
|
def limiter(self):
|
|
from src.core.rate_limiter import RateLimiter
|
|
return RateLimiter()
|
|
|
|
def test_is_banned_initial_false(self, limiter):
|
|
assert limiter.is_banned("192.168.1.1") is False
|
|
|
|
def test_record_attempt_under_limit(self, limiter):
|
|
with patch("src.core.rate_limiter.time") as mock_time:
|
|
mock_time.monotonic.return_value = 1000.0
|
|
for i in range(5): # 5 attempts, well under MAX_ATTEMPTS=10
|
|
limiter.record_attempt("192.168.1.1")
|
|
assert limiter.is_banned("192.168.1.1") is False
|
|
|
|
def test_record_attempt_exceeds_limit(self, limiter):
|
|
from src.core.rate_limiter import MAX_ATTEMPTS
|
|
with patch("src.core.rate_limiter.time") as mock_time:
|
|
mock_time.monotonic.return_value = 1000.0
|
|
for i in range(MAX_ATTEMPTS + 1):
|
|
limiter.record_attempt("192.168.1.1")
|
|
assert limiter.is_banned("192.168.1.1") is True
|
|
|
|
def test_ban_expires(self, limiter):
|
|
from src.core.rate_limiter import MAX_ATTEMPTS, BAN_DURATION
|
|
with patch("src.core.rate_limiter.time") as mock_time:
|
|
mock_time.monotonic.return_value = 1000.0
|
|
for i in range(MAX_ATTEMPTS + 1):
|
|
limiter.record_attempt("192.168.1.1")
|
|
assert limiter.is_banned("192.168.1.1") is True
|
|
|
|
# Advance time past ban duration
|
|
mock_time.monotonic.return_value = 1000.0 + BAN_DURATION + 1
|
|
assert limiter.is_banned("192.168.1.1") is False
|
|
|
|
def test_old_attempts_pruned(self, limiter):
|
|
from src.core.rate_limiter import MAX_ATTEMPTS
|
|
with patch("src.core.rate_limiter.time") as mock_time:
|
|
mock_time.monotonic.return_value = 1000.0
|
|
for i in range(MAX_ATTEMPTS):
|
|
limiter.record_attempt("192.168.1.1")
|
|
|
|
# Advance time past window but not to ban expiry
|
|
mock_time.monotonic.return_value = 1500.0 # 500s later, past 300s window
|
|
# Old attempts should be pruned. One new attempt should not trigger ban.
|
|
limiter.record_attempt("192.168.1.1")
|
|
assert limiter.is_banned("192.168.1.1") is False
|
|
|
|
def test_record_success_clears_attempts(self, limiter):
|
|
with patch("src.core.rate_limiter.time") as mock_time:
|
|
mock_time.monotonic.return_value = 1000.0
|
|
for i in range(5):
|
|
limiter.record_attempt("192.168.1.1")
|
|
limiter.record_success("192.168.1.1")
|
|
assert limiter.is_banned("192.168.1.1") is False
|
|
|
|
def test_different_ips_independent(self, limiter):
|
|
with patch("src.core.rate_limiter.time") as mock_time:
|
|
mock_time.monotonic.return_value = 1000.0
|
|
for i in range(15): # exceeds limit for ip1
|
|
limiter.record_attempt("192.168.1.1")
|
|
assert limiter.is_banned("192.168.1.1") is True
|
|
# Different IP should not be affected
|
|
assert limiter.is_banned("10.0.0.1") is False
|
|
|
|
|
|
class TestRateLimiterSingleton:
|
|
"""rate_limiter singleton is pre-created."""
|
|
|
|
def test_singleton_exists(self):
|
|
from src.core.rate_limiter import rate_limiter
|
|
assert rate_limiter is not None
|
|
assert hasattr(rate_limiter, "is_banned")
|
|
assert hasattr(rate_limiter, "record_attempt")
|
|
assert hasattr(rate_limiter, "record_success")
|
|
# #endregion Test.RateLimiter
|