Added more tests for strategies and rate_limiter

Author: arsh-anwr

backend/tests/rate_limiter/test_ip_strategy.py

@@ -1,11 +1,18 @@
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
+from app.core.rate_limiter.rate_limiting_algorithm.registry import get_rate_limiter
 from app.core.rate_limiter.rate_limiting_algorithm.sliding_window import (
     SlidingWindowRateLimiter,
 )
 
+@pytest.fixture
+def mock_redis_from_url():
+    with patch("redis.asyncio.from_url") as mocker:
+        mock_instance = MagicMock()
+        mocker.return_value = mock_instance
+        yield mocker
 
 @pytest.mark.asyncio
 async def test_allow_first_request():
@@ -40,3 +47,50 @@ async def test_fail_open_allows_requests():
 
     assert allowed is True
     assert retry_after is None
+
+def test_get_rate_limiter_none_strategy(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="none", redis_url="redis://localhost:6379/0", fail_open=True)
+    assert rl is None
+
+
+def test_get_rate_limiter_empty_strategy(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="", redis_url="redis://localhost:6379/0", fail_open=True)
+    assert rl is None
+
+
+def test_get_rate_limiter_null_strategy(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="null", redis_url="redis://localhost:6379/0", fail_open=True)
+    assert rl is None
+
+
+def test_get_rate_limiter_no_strategy(mock_redis_from_url):
+    rl = get_rate_limiter(strategy=None, redis_url="redis://localhost:6379/0", fail_open=True)
+    assert rl is None
+
+
+def test_get_rate_limiter_no_redis_url(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="sliding_window", redis_url="", fail_open=True)
+    assert rl is None
+
+
+def test_get_rate_limiter_sliding_window(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="sliding_window", redis_url="redis://x", fail_open=True)
+    assert isinstance(rl, SlidingWindowRateLimiter)
+
+
+def test_get_rate_limiter_sliding_window_dash(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="sliding-window", redis_url="redis://x", fail_open=False)
+    assert isinstance(rl, SlidingWindowRateLimiter)
+    assert rl.get_fail_open() is False
+
+
+def test_get_rate_limiter_fail_open_coercion(mock_redis_from_url):
+    rl = get_rate_limiter(strategy="sliding_window", redis_url="redis://x", fail_open=None)
+    assert isinstance(rl, SlidingWindowRateLimiter)
+    assert rl.get_fail_open() is False  # default fallback
+
+
+def test_rate_limiter_unknown_strategy(mock_redis_from_url):
+    with pytest.raises(ValueError) as e:
+        get_rate_limiter(strategy="weird_strategy", redis_url="redis://x", fail_open=True)
+    assert "Unknown rate limiter strategy" in str(e.value)

backend/tests/rate_limiter/test_sliding_window_logic.py

@@ -0,0 +1,49 @@
+import pytest
+from starlette.requests import Request
+
+from app.core.rate_limiter.key_strategy.header_key_strategy import HeaderKeyStrategy
+from app.core.rate_limiter.key_strategy.ip_key_strategy import IPKeyStrategy
+from app.core.rate_limiter.key_strategy.key_strategy_enum import KeyStrategyName
+from app.core.rate_limiter.key_strategy.key_strategy_registry import get_key_strategy
+
+
+def test_ip_strategy_build_key():
+    strat = IPKeyStrategy()
+
+    scope = {
+        "client": ("127.0.0.1", 5050),
+        "method": "GET",
+        "path": "/api/test",
+        "headers": [],
+        "type": "http"
+    }
+    request = Request(scope)
+
+    key = strat.get_key(request, scope.get('path'))
+
+    assert key.startswith("ip:127.0.0.1")
+    assert ":/api/test" in key
+
+def test_get_key_strategy_header_default():
+    """
+    Should return HeaderKeyStrategy with default header name
+    when header_name is not provided.
+    """
+    ks = get_key_strategy(KeyStrategyName.HEADER)
+    assert isinstance(ks, HeaderKeyStrategy)
+    assert ks.header_name == "X-Client-ID"
+
+
+def test_get_key_strategy_header_custom():
+    """Should use user-supplied custom header name."""
+    ks = get_key_strategy(KeyStrategyName.HEADER, header_name="X-Auth-ID")
+    assert isinstance(ks, HeaderKeyStrategy)
+    assert ks.header_name == "X-Auth-ID"
+
+
+def test_get_key_strategy_invalid():
+    """Should throw ValueError for unsupported key strategies."""
+    with pytest.raises(ValueError) as e:
+        get_key_strategy("UNKNOWN")  # type: ignore[arg-type]
+
+    assert "Unsupported key strategy" in str(e.value)