test: add 42 unit tests for middleware/cloudflare.py and middleware/language.py
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>
This commit is contained in:
297
tests/unit/middleware/test_cloudflare.py
Normal file
297
tests/unit/middleware/test_cloudflare.py
Normal file
@@ -0,0 +1,297 @@
|
||||
# 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"
|
||||
491
tests/unit/middleware/test_language.py
Normal file
491
tests/unit/middleware/test_language.py
Normal file
@@ -0,0 +1,491 @@
|
||||
# tests/unit/middleware/test_language.py
|
||||
"""
|
||||
Unit tests for LanguageMiddleware, set_language_cookie, and delete_language_cookie.
|
||||
|
||||
Tests cover:
|
||||
- Admin frontend: user pref used, falls back to "en"
|
||||
- Store dashboard: calls resolve_store_dashboard_language, no-store fallback
|
||||
- Storefront: calls resolve_storefront_language, no-store fallback
|
||||
- General: platform frontend, fallback/unknown type, unsupported language
|
||||
validation, request.state.language set, request.state.language_info populated,
|
||||
cookie language read
|
||||
- Helpers: _get_user_language_from_token, _get_customer_language_from_token,
|
||||
set_language_cookie (valid/invalid), delete_language_cookie
|
||||
- Private methods: user with pref, user without pref, no user, customer with pref
|
||||
"""
|
||||
|
||||
from unittest.mock import AsyncMock, Mock, patch
|
||||
|
||||
import pytest
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import Response
|
||||
|
||||
from app.modules.enums import FrontendType
|
||||
from middleware.language import (
|
||||
LanguageMiddleware,
|
||||
delete_language_cookie,
|
||||
set_language_cookie,
|
||||
)
|
||||
|
||||
|
||||
def _make_request(
|
||||
frontend_type=None,
|
||||
store=None,
|
||||
cookies=None,
|
||||
headers=None,
|
||||
current_user=None,
|
||||
current_customer=None,
|
||||
):
|
||||
"""Build a mock Request with the given state attributes."""
|
||||
request = Mock(spec=Request)
|
||||
request.state = Mock(spec=[])
|
||||
request.cookies = cookies or {}
|
||||
request.headers = headers or {}
|
||||
|
||||
if frontend_type is not None:
|
||||
request.state.frontend_type = frontend_type
|
||||
if store is not None:
|
||||
request.state.store = store
|
||||
if current_user is not None:
|
||||
request.state.current_user = current_user
|
||||
if current_customer is not None:
|
||||
request.state.current_customer = current_customer
|
||||
|
||||
return request
|
||||
|
||||
|
||||
def _make_call_next():
|
||||
"""Return (call_next, response) pair."""
|
||||
response = Mock(status_code=200, headers={})
|
||||
call_next = AsyncMock(return_value=response)
|
||||
return call_next, response
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestLanguageDispatchAdmin
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.unit
|
||||
class TestLanguageDispatchAdmin:
|
||||
"""Admin frontend language resolution."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_user_preferred_language_used(self):
|
||||
"""Admin: user's preferred_language is used when set."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
user = Mock()
|
||||
user.preferred_language = "de"
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.ADMIN,
|
||||
current_user=user,
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "de"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_falls_back_to_en(self):
|
||||
"""Admin: falls back to 'en' when no user preference."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(frontend_type=FrontendType.ADMIN)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "en"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestLanguageDispatchStore
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.unit
|
||||
class TestLanguageDispatchStore:
|
||||
"""Store dashboard language resolution."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_store_calls_resolve_store_dashboard_language(self):
|
||||
"""Store: resolve_store_dashboard_language is called with correct args."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
user = Mock()
|
||||
user.preferred_language = "de"
|
||||
store = Mock()
|
||||
store.dashboard_language = "fr"
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.STORE,
|
||||
store=store,
|
||||
current_user=user,
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with (
|
||||
patch("middleware.language.parse_accept_language", return_value=None),
|
||||
patch(
|
||||
"middleware.language.resolve_store_dashboard_language",
|
||||
return_value="de",
|
||||
) as mock_resolve,
|
||||
):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
mock_resolve.assert_called_once_with(
|
||||
user_preferred="de",
|
||||
store_dashboard="fr",
|
||||
)
|
||||
assert request.state.language == "de"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_store_no_store_passes_none_dashboard(self):
|
||||
"""Store: store_dashboard is None when request.state has no store."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(frontend_type=FrontendType.STORE)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with (
|
||||
patch("middleware.language.parse_accept_language", return_value=None),
|
||||
patch(
|
||||
"middleware.language.resolve_store_dashboard_language",
|
||||
return_value="fr",
|
||||
) as mock_resolve,
|
||||
):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
mock_resolve.assert_called_once_with(
|
||||
user_preferred=None,
|
||||
store_dashboard=None,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestLanguageDispatchStorefront
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.unit
|
||||
class TestLanguageDispatchStorefront:
|
||||
"""Storefront language resolution."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_storefront_calls_resolve_storefront_language(self):
|
||||
"""Storefront: resolve_storefront_language called with correct args."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
customer = Mock()
|
||||
customer.preferred_language = "lb"
|
||||
store = Mock()
|
||||
store.storefront_language = "fr"
|
||||
store.storefront_languages = ["fr", "de", "lb"]
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.STOREFRONT,
|
||||
store=store,
|
||||
current_customer=customer,
|
||||
cookies={"lang": "de"},
|
||||
headers={"accept-language": "en-US,en;q=0.9"},
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"middleware.language.parse_accept_language", return_value="en"
|
||||
),
|
||||
patch(
|
||||
"middleware.language.resolve_storefront_language",
|
||||
return_value="lb",
|
||||
) as mock_resolve,
|
||||
):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
mock_resolve.assert_called_once_with(
|
||||
customer_preferred="lb",
|
||||
session_language="de",
|
||||
store_storefront="fr",
|
||||
browser_language="en",
|
||||
enabled_languages=["fr", "de", "lb"],
|
||||
)
|
||||
assert request.state.language == "lb"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_storefront_no_store_passes_none(self):
|
||||
"""Storefront: store fields are None when no store in state."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(frontend_type=FrontendType.STOREFRONT)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with (
|
||||
patch("middleware.language.parse_accept_language", return_value=None),
|
||||
patch(
|
||||
"middleware.language.resolve_storefront_language",
|
||||
return_value="fr",
|
||||
) as mock_resolve,
|
||||
):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
mock_resolve.assert_called_once_with(
|
||||
customer_preferred=None,
|
||||
session_language=None,
|
||||
store_storefront=None,
|
||||
browser_language=None,
|
||||
enabled_languages=None,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestLanguageDispatchGeneral
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.unit
|
||||
class TestLanguageDispatchGeneral:
|
||||
"""Platform, fallback/unknown, validation, state, and cookie tests."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_platform_uses_cookie_language(self):
|
||||
"""Platform: cookie language is used first."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.PLATFORM,
|
||||
cookies={"lang": "de"},
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value="en"):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "de"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_platform_falls_back_to_browser_language(self):
|
||||
"""Platform: browser language used when no cookie."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(frontend_type=FrontendType.PLATFORM)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value="de"):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "de"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_fallback_unknown_frontend_type(self):
|
||||
"""Unknown/None frontend type uses cookie → browser → default."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request() # no frontend_type
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
# DEFAULT_LANGUAGE is "fr"
|
||||
assert request.state.language == "fr"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_unsupported_language_falls_back_to_default(self):
|
||||
"""Unsupported language code is replaced by DEFAULT_LANGUAGE."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.PLATFORM,
|
||||
cookies={"lang": "xx"}, # unsupported
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "fr"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_state_language_is_set(self):
|
||||
"""request.state.language is always set after dispatch."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(frontend_type=FrontendType.PLATFORM)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value="en"):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert hasattr(request.state, "language")
|
||||
assert request.state.language == "en"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_request_state_language_info_populated(self):
|
||||
"""request.state.language_info contains code, cookie, browser, frontend_type."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.PLATFORM,
|
||||
cookies={"lang": "en"},
|
||||
headers={"accept-language": "de"},
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value="de"):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
info = request.state.language_info
|
||||
assert info["code"] == "en"
|
||||
assert info["cookie"] == "en"
|
||||
assert info["browser"] == "de"
|
||||
assert info["frontend_type"] == "platform"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_cookie_language_read_from_request(self):
|
||||
"""The 'lang' cookie is read and considered in resolution."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.PLATFORM,
|
||||
cookies={"lang": "lb"},
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "lb"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestLanguageHelpers
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.unit
|
||||
class TestLanguageHelpers:
|
||||
"""Tests for set_language_cookie and delete_language_cookie."""
|
||||
|
||||
def test_set_language_cookie_valid(self):
|
||||
"""set_language_cookie sets cookie for supported language."""
|
||||
response = Mock(spec=Response)
|
||||
result = set_language_cookie(response, "de")
|
||||
|
||||
response.set_cookie.assert_called_once_with(
|
||||
key="lang",
|
||||
value="de",
|
||||
max_age=60 * 60 * 24 * 365,
|
||||
httponly=False,
|
||||
samesite="lax",
|
||||
)
|
||||
assert result is response
|
||||
|
||||
def test_set_language_cookie_invalid_language(self):
|
||||
"""set_language_cookie does NOT call set_cookie for unsupported lang."""
|
||||
response = Mock(spec=Response)
|
||||
result = set_language_cookie(response, "xx")
|
||||
|
||||
response.set_cookie.assert_not_called()
|
||||
assert result is response
|
||||
|
||||
def test_delete_language_cookie(self):
|
||||
"""delete_language_cookie calls delete_cookie with correct key."""
|
||||
response = Mock(spec=Response)
|
||||
result = delete_language_cookie(response)
|
||||
|
||||
response.delete_cookie.assert_called_once_with(key="lang")
|
||||
assert result is response
|
||||
|
||||
def test_get_user_language_from_token_with_pref(self):
|
||||
"""_get_user_language_from_token returns preferred_language."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = Mock(spec=Request)
|
||||
request.state = Mock()
|
||||
request.state.current_user = Mock()
|
||||
request.state.current_user.preferred_language = "de"
|
||||
|
||||
assert middleware._get_user_language_from_token(request) == "de"
|
||||
|
||||
def test_get_user_language_from_token_no_user(self):
|
||||
"""_get_user_language_from_token returns None with no user."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = Mock(spec=Request)
|
||||
request.state = Mock(spec=[]) # no current_user
|
||||
|
||||
assert middleware._get_user_language_from_token(request) is None
|
||||
|
||||
def test_get_customer_language_from_token_with_pref(self):
|
||||
"""_get_customer_language_from_token returns preferred_language."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = Mock(spec=Request)
|
||||
request.state = Mock()
|
||||
request.state.current_customer = Mock()
|
||||
request.state.current_customer.preferred_language = "lb"
|
||||
|
||||
assert middleware._get_customer_language_from_token(request) == "lb"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestLanguageDispatchPrivateMethods
|
||||
# ---------------------------------------------------------------------------
|
||||
@pytest.mark.unit
|
||||
class TestLanguageDispatchPrivateMethods:
|
||||
"""Integration-style tests exercising private helpers through dispatch."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_user_with_preferred_language(self):
|
||||
"""Admin + user with pref → uses that preference."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
user = Mock()
|
||||
user.preferred_language = "lb"
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.ADMIN,
|
||||
current_user=user,
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "lb"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_user_without_preferred_language(self):
|
||||
"""Admin + user without preferred_language → falls back to 'en'."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
user = Mock(spec=[]) # no preferred_language attr
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.ADMIN,
|
||||
current_user=user,
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "en"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_admin_no_user_at_all(self):
|
||||
"""Admin + no user at all → falls back to 'en'."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
request = _make_request(frontend_type=FrontendType.ADMIN)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with patch("middleware.language.parse_accept_language", return_value=None):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
assert request.state.language == "en"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_storefront_customer_with_preferred_language(self):
|
||||
"""Storefront + customer with pref → passed to resolve function."""
|
||||
middleware = LanguageMiddleware(app=None)
|
||||
customer = Mock()
|
||||
customer.preferred_language = "de"
|
||||
store = Mock()
|
||||
store.storefront_language = "fr"
|
||||
store.storefront_languages = ["fr", "de"]
|
||||
request = _make_request(
|
||||
frontend_type=FrontendType.STOREFRONT,
|
||||
store=store,
|
||||
current_customer=customer,
|
||||
)
|
||||
call_next, _ = _make_call_next()
|
||||
|
||||
with (
|
||||
patch("middleware.language.parse_accept_language", return_value=None),
|
||||
patch(
|
||||
"middleware.language.resolve_storefront_language",
|
||||
return_value="de",
|
||||
) as mock_resolve,
|
||||
):
|
||||
await middleware.dispatch(request, call_next)
|
||||
|
||||
mock_resolve.assert_called_once_with(
|
||||
customer_preferred="de",
|
||||
session_language=None,
|
||||
store_storefront="fr",
|
||||
browser_language=None,
|
||||
enabled_languages=["fr", "de"],
|
||||
)
|
||||
assert request.state.language == "de"
|
||||
Reference in New Issue
Block a user