Samir Boulahtit
6c5969e4e1
Completes middleware test coverage (11/11 files) with 19 cloudflare tests (dispatch, get_real_client_ip, get_client_country) and 23 language tests (admin/store/storefront/platform dispatch, helpers, private methods). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
298 lines
11 KiB
Python
298 lines
11 KiB
Python
# tests/unit/middleware/test_cloudflare.py
|
|
"""
|
|
Unit tests for CloudFlareMiddleware, get_real_client_ip, and get_client_country.
|
|
|
|
Tests cover:
|
|
- CloudFlare dispatch: CF disabled bypass, header extraction, missing header
|
|
fallbacks, CF-Visitor JSON parsing (valid/malformed), X-CF-Ray response
|
|
header, no-client fallback
|
|
- get_real_client_ip: state.real_ip, CF-Connecting-IP, X-Forwarded-For,
|
|
request.client fallback, no client → "unknown"
|
|
- get_client_country: state.client_country, CF-IPCountry header, neither
|
|
available, state takes priority over header
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
import pytest
|
|
from starlette.requests import Request
|
|
|
|
from middleware.cloudflare import (
|
|
CloudFlareMiddleware,
|
|
get_client_country,
|
|
get_real_client_ip,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestCloudFlareDispatch
|
|
# ---------------------------------------------------------------------------
|
|
@pytest.mark.unit
|
|
class TestCloudFlareDispatch:
|
|
"""Tests for CloudFlareMiddleware.dispatch()."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cf_disabled_bypasses_middleware(self):
|
|
"""When cloudflare_enabled is False, call_next is invoked directly."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = False
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
call_next.assert_awaited_once_with(request)
|
|
assert result is response
|
|
# No state attributes should have been set
|
|
assert not hasattr(request.state, "real_ip")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_cf_headers_extracted(self):
|
|
"""All CloudFlare headers are stored on request.state."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {
|
|
"CF-Connecting-IP": "1.2.3.4",
|
|
"CF-IPCountry": "LU",
|
|
"CF-Ray": "abc123",
|
|
"CF-Visitor": '{"scheme":"https"}',
|
|
}
|
|
request.state = Mock(spec=[])
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.real_ip == "1.2.3.4"
|
|
assert request.state.client_country == "LU"
|
|
assert request.state.cf_ray == "abc123"
|
|
assert request.state.original_scheme == "https"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_cf_connecting_ip_falls_back_to_client(self):
|
|
"""Without CF-Connecting-IP, real_ip falls back to request.client.host."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {}
|
|
request.state = Mock(spec=[])
|
|
request.client = Mock(host="10.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.real_ip == "10.0.0.1"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_cf_headers_no_state_attributes(self):
|
|
"""Missing optional headers do not set corresponding state attributes."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {}
|
|
request.state = Mock(spec=[])
|
|
request.client = Mock(host="10.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# client_country and cf_ray should NOT be set
|
|
assert not hasattr(request.state, "client_country")
|
|
assert not hasattr(request.state, "cf_ray")
|
|
assert not hasattr(request.state, "original_scheme")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cf_visitor_json_parsed(self):
|
|
"""CF-Visitor JSON body is parsed and scheme extracted."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {"CF-Visitor": '{"scheme":"http"}'}
|
|
request.state = Mock(spec=[])
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.original_scheme == "http"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cf_visitor_malformed_json_defaults_to_https(self):
|
|
"""Malformed CF-Visitor JSON defaults original_scheme to 'https'."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {"CF-Visitor": "not-json"}
|
|
request.state = Mock(spec=[])
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.original_scheme == "https"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cf_ray_added_to_response_header(self):
|
|
"""X-CF-Ray response header is set when CF-Ray is present."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {"CF-Ray": "ray-999"}
|
|
request.state = Mock(spec=[])
|
|
request.client = Mock(host="127.0.0.1")
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
result = await middleware.dispatch(request, call_next)
|
|
|
|
assert result.headers["X-CF-Ray"] == "ray-999"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_client_fallback_returns_unknown(self):
|
|
"""When request.client is None, real_ip is set to 'unknown'."""
|
|
middleware = CloudFlareMiddleware(app=None)
|
|
request = Mock(spec=Request)
|
|
request.headers = {}
|
|
request.state = Mock(spec=[])
|
|
request.client = None
|
|
|
|
response = Mock(status_code=200, headers={})
|
|
call_next = AsyncMock(return_value=response)
|
|
|
|
with patch("middleware.cloudflare.settings") as mock_settings:
|
|
mock_settings.cloudflare_enabled = True
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.real_ip == "unknown"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestGetRealClientIp
|
|
# ---------------------------------------------------------------------------
|
|
@pytest.mark.unit
|
|
class TestGetRealClientIp:
|
|
"""Tests for get_real_client_ip() standalone helper."""
|
|
|
|
def test_returns_state_real_ip_when_present(self):
|
|
"""state.real_ip takes priority over everything else."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock()
|
|
request.state.real_ip = "5.6.7.8"
|
|
request.headers = {"CF-Connecting-IP": "1.2.3.4"}
|
|
|
|
assert get_real_client_ip(request) == "5.6.7.8"
|
|
|
|
def test_returns_cf_connecting_ip_header(self):
|
|
"""Falls back to CF-Connecting-IP when state.real_ip absent."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[]) # no real_ip attribute
|
|
request.headers = {"CF-Connecting-IP": "1.2.3.4"}
|
|
|
|
assert get_real_client_ip(request) == "1.2.3.4"
|
|
|
|
def test_returns_first_x_forwarded_for_ip(self):
|
|
"""Falls back to first IP from X-Forwarded-For header."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
request.headers = {"X-Forwarded-For": "9.8.7.6, 10.0.0.1, 10.0.0.2"}
|
|
|
|
assert get_real_client_ip(request) == "9.8.7.6"
|
|
|
|
def test_returns_single_x_forwarded_for_ip(self):
|
|
"""X-Forwarded-For with single IP still works."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
request.headers = {"X-Forwarded-For": "9.8.7.6"}
|
|
|
|
assert get_real_client_ip(request) == "9.8.7.6"
|
|
|
|
def test_returns_client_host_as_fallback(self):
|
|
"""Falls back to request.client.host when no proxy headers."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
request.headers = {}
|
|
request.client = Mock(host="192.168.1.1")
|
|
|
|
assert get_real_client_ip(request) == "192.168.1.1"
|
|
|
|
def test_returns_unknown_when_no_client(self):
|
|
"""Returns 'unknown' when client is None and no proxy headers."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
request.headers = {}
|
|
request.client = None
|
|
|
|
assert get_real_client_ip(request) == "unknown"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# TestGetClientCountry
|
|
# ---------------------------------------------------------------------------
|
|
@pytest.mark.unit
|
|
class TestGetClientCountry:
|
|
"""Tests for get_client_country() standalone helper."""
|
|
|
|
def test_returns_state_client_country_when_present(self):
|
|
"""state.client_country takes priority."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock()
|
|
request.state.client_country = "LU"
|
|
request.headers = {"CF-IPCountry": "DE"}
|
|
|
|
assert get_client_country(request) == "LU"
|
|
|
|
def test_returns_cf_ipcountry_header_fallback(self):
|
|
"""Falls back to CF-IPCountry header when state not set."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
request.headers = {"CF-IPCountry": "FR"}
|
|
|
|
assert get_client_country(request) == "FR"
|
|
|
|
def test_returns_none_when_neither_available(self):
|
|
"""Returns None when neither state nor header is available."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[])
|
|
request.headers = {}
|
|
|
|
assert get_client_country(request) is None
|
|
|
|
def test_state_takes_priority_over_header(self):
|
|
"""state.client_country takes priority even when header differs."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock()
|
|
request.state.client_country = "BE"
|
|
request.headers = {"CF-IPCountry": "NL"}
|
|
|
|
assert get_client_country(request) == "BE"
|
|
|
|
def test_header_with_no_state_attribute(self):
|
|
"""CF-IPCountry header used when state has no client_country attr."""
|
|
request = Mock(spec=Request)
|
|
# Use spec=[] so hasattr returns False for all attributes
|
|
request.state = Mock(spec=[])
|
|
request.headers = {"CF-IPCountry": "DE"}
|
|
|
|
assert get_client_country(request) == "DE"
|