Samir Boulahtit
f47c680cb8
- Extract store/platform context from Referer header for storefront API requests
(StoreContextMiddleware and PlatformContextMiddleware) so login POST works in
dev mode where API paths lack /platforms/{code}/ prefix
- Set customer token cookie path to "/" for cross-route compatibility
- Fix double storefront in URLs: replace {{ base_url }}storefront/ with {{ base_url }}
across all 24 storefront templates
- Fix auth error redirect to include platform prefix and use store_code
- Update seed script to output correct storefront login URLs
- Add 20 new unit tests covering all fixes; fix 9 pre-existing test failures
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1030 lines
38 KiB
Python
1030 lines
38 KiB
Python
# tests/unit/middleware/test_store_context.py
|
|
"""
|
|
Comprehensive unit tests for StoreContextMiddleware and StoreContextManager.
|
|
|
|
Tests cover:
|
|
- Store detection from custom domains, subdomains, and path-based routing
|
|
- Database lookup and store validation
|
|
- Path extraction and cleanup
|
|
- Admin and API request detection
|
|
- Static file request detection
|
|
- Edge cases and error handling
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock, Mock, patch
|
|
|
|
import pytest
|
|
from fastapi import Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.frontend_detector import FrontendDetector
|
|
from app.modules.tenancy.exceptions import StoreNotFoundException
|
|
from middleware.store_context import (
|
|
StoreContextManager,
|
|
StoreContextMiddleware,
|
|
get_current_store,
|
|
require_store_context,
|
|
)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.stores
|
|
class TestStoreContextManager:
|
|
"""Test suite for StoreContextManager static methods."""
|
|
|
|
# ========================================================================
|
|
# Store Context Detection Tests
|
|
# ========================================================================
|
|
|
|
def test_detect_custom_domain(self):
|
|
"""Test custom domain detection."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "customdomain1.com"}
|
|
request.url = Mock(path="/")
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "custom_domain"
|
|
assert context["domain"] == "customdomain1.com"
|
|
assert context["host"] == "customdomain1.com"
|
|
|
|
def test_detect_custom_domain_with_port(self):
|
|
"""Test custom domain detection with port number."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "customdomain1.com:8000"}
|
|
request.url = Mock(path="/")
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "custom_domain"
|
|
assert context["domain"] == "customdomain1.com"
|
|
assert context["host"] == "customdomain1.com"
|
|
|
|
def test_detect_subdomain(self):
|
|
"""Test subdomain detection."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "store1.platform.com"}
|
|
request.url = Mock(path="/")
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "subdomain"
|
|
assert context["subdomain"] == "store1"
|
|
assert context["host"] == "store1.platform.com"
|
|
|
|
def test_detect_subdomain_with_port(self):
|
|
"""Test subdomain detection with port number."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "store1.platform.com:8000"}
|
|
request.url = Mock(path="/")
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "subdomain"
|
|
assert context["subdomain"] == "store1"
|
|
|
|
def test_detect_path_store_singular(self):
|
|
"""Test path-based detection with /store/ prefix."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/store/store1/storefront")
|
|
# Set platform_clean_path to simulate PlatformContextMiddleware output
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = "/store/store1/storefront"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "path"
|
|
assert context["subdomain"] == "store1"
|
|
assert context["path_prefix"] == "/store/store1"
|
|
assert context["full_prefix"] == "/store/"
|
|
|
|
def test_detect_path_stores_plural(self):
|
|
"""Test path-based detection with /stores/ prefix."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/stores/store1/storefront")
|
|
# Set platform_clean_path to simulate PlatformContextMiddleware output
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = "/stores/store1/storefront"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "path"
|
|
assert context["subdomain"] == "store1"
|
|
assert context["path_prefix"] == "/stores/store1"
|
|
assert context["full_prefix"] == "/stores/"
|
|
|
|
def test_detect_no_store_context(self):
|
|
"""Test when no store context can be detected."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/random/path")
|
|
# Set platform_clean_path to None to use url.path
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
def test_ignore_admin_subdomain(self):
|
|
"""Test that admin subdomain is not detected as store."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "admin.platform.com"}
|
|
request.url = Mock(path="/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
def test_ignore_www_subdomain(self):
|
|
"""Test that www subdomain is not detected as store."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "www.platform.com"}
|
|
request.url = Mock(path="/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
def test_ignore_api_subdomain(self):
|
|
"""Test that api subdomain is not detected as store."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "api.platform.com"}
|
|
request.url = Mock(path="/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
def test_ignore_localhost(self):
|
|
"""Test that localhost is not detected as custom domain."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
# ========================================================================
|
|
# Store Database Lookup Tests
|
|
# ========================================================================
|
|
|
|
def test_get_store_from_custom_domain_context(self):
|
|
"""Test getting store from custom domain context."""
|
|
mock_db = Mock(spec=Session)
|
|
mock_store_domain = Mock()
|
|
mock_store = Mock()
|
|
mock_store.is_active = True
|
|
mock_store_domain.store = mock_store
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = mock_store_domain
|
|
|
|
context = {"detection_method": "custom_domain", "domain": "customdomain1.com"}
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is mock_store
|
|
assert store.is_active is True
|
|
|
|
def test_get_store_from_custom_domain_inactive_store(self):
|
|
"""Test getting inactive store from custom domain context."""
|
|
mock_db = Mock(spec=Session)
|
|
mock_store_domain = Mock()
|
|
mock_store = Mock()
|
|
mock_store.is_active = False
|
|
mock_store_domain.store = mock_store
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = mock_store_domain
|
|
|
|
context = {"detection_method": "custom_domain", "domain": "customdomain1.com"}
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is None
|
|
|
|
def test_get_store_from_custom_domain_not_found(self):
|
|
"""Test custom domain not found in database."""
|
|
mock_db = Mock(spec=Session)
|
|
# Ensure all query chain variants return None for .first()
|
|
# (primary StoreDomain lookup and MerchantDomain fallback)
|
|
query_mock = mock_db.query.return_value
|
|
query_mock.filter.return_value.first.return_value = None
|
|
query_mock.filter.return_value.filter.return_value.first.return_value = None
|
|
query_mock.filter.return_value.filter.return_value.filter.return_value.first.return_value = None
|
|
query_mock.filter.return_value.order_by.return_value.first.return_value = None
|
|
|
|
context = {"detection_method": "custom_domain", "domain": "nonexistent.com"}
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is None
|
|
|
|
def test_get_store_from_subdomain_context(self):
|
|
"""Test getting store from subdomain context."""
|
|
mock_db = Mock(spec=Session)
|
|
mock_store = Mock()
|
|
mock_store.is_active = True
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_store
|
|
|
|
context = {"detection_method": "subdomain", "subdomain": "store1"}
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is mock_store
|
|
|
|
def test_get_store_from_path_context(self):
|
|
"""Test getting store from path context."""
|
|
mock_db = Mock(spec=Session)
|
|
mock_store = Mock()
|
|
mock_store.is_active = True
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_store
|
|
|
|
context = {"detection_method": "path", "subdomain": "store1"}
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is mock_store
|
|
|
|
def test_get_store_with_no_context(self):
|
|
"""Test getting store with no context."""
|
|
mock_db = Mock(spec=Session)
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, None)
|
|
|
|
assert store is None
|
|
|
|
def test_get_store_subdomain_case_insensitive(self):
|
|
"""Test subdomain lookup is case-insensitive."""
|
|
mock_db = Mock(spec=Session)
|
|
mock_store = Mock()
|
|
mock_store.is_active = True
|
|
|
|
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_store
|
|
|
|
context = {"detection_method": "subdomain", "subdomain": "STORE1"} # Uppercase
|
|
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is mock_store
|
|
|
|
# ========================================================================
|
|
# Path Extraction Tests
|
|
# ========================================================================
|
|
|
|
def test_extract_clean_path_from_store_path(self):
|
|
"""Test extracting clean path from /store/ prefix."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/store/store1/storefront/products")
|
|
|
|
store_context = {"detection_method": "path", "path_prefix": "/store/store1"}
|
|
|
|
clean_path = StoreContextManager.extract_clean_path(request, store_context)
|
|
|
|
assert clean_path == "/storefront/products"
|
|
|
|
def test_extract_clean_path_from_stores_path(self):
|
|
"""Test extracting clean path from /stores/ prefix."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/stores/store1/storefront/products")
|
|
|
|
store_context = {"detection_method": "path", "path_prefix": "/stores/store1"}
|
|
|
|
clean_path = StoreContextManager.extract_clean_path(request, store_context)
|
|
|
|
assert clean_path == "/storefront/products"
|
|
|
|
def test_extract_clean_path_root(self):
|
|
"""Test extracting clean path when result is empty (should return /)."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/store/store1")
|
|
|
|
store_context = {"detection_method": "path", "path_prefix": "/store/store1"}
|
|
|
|
clean_path = StoreContextManager.extract_clean_path(request, store_context)
|
|
|
|
assert clean_path == "/"
|
|
|
|
def test_extract_clean_path_no_path_context(self):
|
|
"""Test extracting clean path for non-path detection methods."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/storefront/products")
|
|
|
|
store_context = {"detection_method": "subdomain", "subdomain": "store1"}
|
|
|
|
clean_path = StoreContextManager.extract_clean_path(request, store_context)
|
|
|
|
assert clean_path == "/storefront/products"
|
|
|
|
def test_extract_clean_path_no_context(self):
|
|
"""Test extracting clean path with no store context."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/storefront/products")
|
|
|
|
clean_path = StoreContextManager.extract_clean_path(request, None)
|
|
|
|
assert clean_path == "/storefront/products"
|
|
|
|
# ========================================================================
|
|
# Request Type Detection Tests
|
|
# ========================================================================
|
|
|
|
def test_is_admin_request_admin_subdomain(self):
|
|
"""Test admin request detection from subdomain."""
|
|
assert FrontendDetector.is_admin("admin.platform.com", "/dashboard") is True
|
|
|
|
def test_is_admin_request_admin_path(self):
|
|
"""Test admin request detection from path."""
|
|
assert FrontendDetector.is_admin("localhost", "/admin/dashboard") is True
|
|
|
|
def test_is_admin_request_with_port(self):
|
|
"""Test admin request detection with port number."""
|
|
assert FrontendDetector.is_admin("admin.localhost:8000", "/dashboard") is True
|
|
|
|
def test_is_not_admin_request(self):
|
|
"""Test non-admin request."""
|
|
assert FrontendDetector.is_admin("store1.platform.com", "/storefront") is False
|
|
|
|
def test_is_api_request(self):
|
|
"""Test API request detection."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/api/v1/stores")
|
|
|
|
assert StoreContextManager.is_api_request(request) is True
|
|
|
|
def test_is_not_api_request(self):
|
|
"""Test non-API request."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path="/storefront/products")
|
|
|
|
assert StoreContextManager.is_api_request(request) is False
|
|
|
|
# ========================================================================
|
|
# Extract Store From Referer Tests
|
|
# ========================================================================
|
|
|
|
def test_extract_store_from_referer_path_stores(self):
|
|
"""Test extracting store from referer with /stores/ path."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {
|
|
"referer": "http://localhost:8000/stores/orion/storefront/products"
|
|
}
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is not None
|
|
assert context["subdomain"] == "orion"
|
|
assert context["detection_method"] == "path"
|
|
assert context["path_prefix"] == "/stores/orion"
|
|
assert context["full_prefix"] == "/stores/"
|
|
|
|
def test_extract_store_from_referer_path_store(self):
|
|
"""Test extracting store from referer with /store/ path."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {
|
|
"referer": "http://localhost:8000/store/myshop/storefront/products"
|
|
}
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is not None
|
|
assert context["subdomain"] == "myshop"
|
|
assert context["detection_method"] == "path"
|
|
assert context["path_prefix"] == "/store/myshop"
|
|
assert context["full_prefix"] == "/store/"
|
|
|
|
def test_extract_store_from_referer_subdomain(self):
|
|
"""Test extracting store from referer with subdomain."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"referer": "http://orion.platform.com/storefront/products"} # noqa: SEC034
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is not None
|
|
assert context["subdomain"] == "orion"
|
|
assert context["detection_method"] == "referer_subdomain"
|
|
assert context["host"] == "orion.platform.com"
|
|
|
|
def test_extract_store_from_referer_custom_domain(self):
|
|
"""Test extracting store from referer with custom domain."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"referer": "http://my-custom-shop.com/storefront/products"} # noqa: SEC034
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is not None
|
|
assert context["domain"] == "my-custom-shop.com"
|
|
assert context["detection_method"] == "referer_custom_domain"
|
|
assert context["host"] == "my-custom-shop.com"
|
|
|
|
def test_extract_store_from_referer_no_header(self):
|
|
"""Test extracting store when no referer header present."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {}
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is None
|
|
|
|
def test_extract_store_from_referer_origin_header(self):
|
|
"""Test extracting store from origin header when referer is missing."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"origin": "http://localhost:8000/stores/testshop/storefront"}
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is not None
|
|
assert context["subdomain"] == "testshop"
|
|
|
|
def test_extract_store_from_referer_ignores_admin_subdomain(self):
|
|
"""Test that admin subdomain is not extracted from referer."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"referer": "http://admin.platform.com/dashboard"} # noqa: SEC034
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
# admin subdomain should not be detected as store
|
|
assert context is None
|
|
|
|
def test_extract_store_from_referer_ignores_www_subdomain(self):
|
|
"""Test that www subdomain is not extracted from referer."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"referer": "http://www.platform.com/storefront"} # noqa: SEC034
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is None
|
|
|
|
def test_extract_store_from_referer_localhost_not_custom_domain(self):
|
|
"""Test that localhost is not treated as custom domain."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"referer": "http://localhost:8000/storefront"}
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.extract_store_from_referer(request)
|
|
|
|
assert context is None
|
|
|
|
# ========================================================================
|
|
# Static File Detection Tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/static/css/style.css",
|
|
"/static/js/app.js",
|
|
"/media/images/product.png",
|
|
"/assets/logo.svg",
|
|
"/.well-known/security.txt",
|
|
"/favicon.ico",
|
|
"/image.jpg",
|
|
"/style.css",
|
|
"/app.webmanifest",
|
|
"/static/", # Path starting with /static/ but no extension
|
|
"/media/uploads", # Path starting with /media/ but no extension
|
|
"/subfolder/favicon.ico", # favicon.ico in subfolder
|
|
"/favicon.ico.bak", # Contains favicon.ico but doesn't end with static extension (hits line 226)
|
|
],
|
|
)
|
|
def test_is_static_file_request(self, path):
|
|
"""Test static file detection for various paths and extensions."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path=path)
|
|
|
|
assert StoreContextManager.is_static_file_request(request) is True
|
|
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/storefront/products",
|
|
"/admin/dashboard",
|
|
"/api/stores",
|
|
"/about",
|
|
],
|
|
)
|
|
def test_is_not_static_file_request(self, path):
|
|
"""Test non-static file paths."""
|
|
request = Mock(spec=Request)
|
|
request.url = Mock(path=path)
|
|
|
|
assert StoreContextManager.is_static_file_request(request) is False
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.stores
|
|
class TestStoreContextMiddleware:
|
|
"""Test suite for StoreContextMiddleware."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_skips_admin_request(self):
|
|
"""Test middleware skips store detection for admin requests."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "admin.platform.com"}
|
|
request.url = Mock(path="/admin/dashboard")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with patch.object(FrontendDetector, "is_admin", return_value=True):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
assert request.state.store_context is None
|
|
assert request.state.clean_path == "/admin/dashboard"
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_skips_api_request(self):
|
|
"""Test middleware skips store detection for API requests."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/api/v1/stores")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with patch.object(StoreContextManager, "is_api_request", return_value=True):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
assert request.state.store_context is None
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_skips_static_file_request(self):
|
|
"""Test middleware skips store detection for static files."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/static/css/style.css")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with patch.object(
|
|
StoreContextManager, "is_static_file_request", return_value=True
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_detects_and_sets_store(self):
|
|
"""Test middleware successfully detects and sets store."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "store1.platform.com"}
|
|
request.url = Mock(path="/storefront/products")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
mock_store = Mock()
|
|
mock_store.id = 1
|
|
mock_store.name = "Test Store"
|
|
mock_store.subdomain = "store1"
|
|
|
|
store_context = {"detection_method": "subdomain", "subdomain": "store1"}
|
|
|
|
mock_db = MagicMock()
|
|
|
|
with (
|
|
patch.object(
|
|
StoreContextManager,
|
|
"detect_store_context",
|
|
return_value=store_context,
|
|
),
|
|
patch.object(
|
|
StoreContextManager,
|
|
"get_store_from_context",
|
|
return_value=mock_store,
|
|
),
|
|
patch.object(
|
|
StoreContextManager,
|
|
"extract_clean_path",
|
|
return_value="/storefront/products",
|
|
),
|
|
patch("middleware.store_context.get_db", return_value=iter([mock_db])),
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is mock_store
|
|
assert request.state.store_context == store_context
|
|
assert request.state.clean_path == "/storefront/products"
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_store_not_found(self):
|
|
"""Test middleware when store context detected but store not in database."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "nonexistent.platform.com"}
|
|
request.url = Mock(path="/storefront")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
store_context = {"detection_method": "subdomain", "subdomain": "nonexistent"}
|
|
|
|
mock_db = MagicMock()
|
|
|
|
with (
|
|
patch.object(
|
|
StoreContextManager,
|
|
"detect_store_context",
|
|
return_value=store_context,
|
|
),
|
|
patch.object(
|
|
StoreContextManager, "get_store_from_context", return_value=None
|
|
),
|
|
patch("middleware.store_context.get_db", return_value=iter([mock_db])),
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
assert request.state.store_context == store_context
|
|
assert request.state.clean_path == "/storefront"
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_no_store_context(self):
|
|
"""Test middleware when no store context detected."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/random/path")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with patch.object(
|
|
StoreContextManager, "detect_store_context", return_value=None
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
assert request.state.store_context is None
|
|
assert request.state.clean_path == "/random/path"
|
|
call_next.assert_called_once_with(request)
|
|
|
|
# ========================================================================
|
|
# Storefront API Referer Tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_storefront_api_uses_referer(self):
|
|
"""Test storefront API requests get store context from Referer header."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {
|
|
"host": "localhost",
|
|
"referer": "http://localhost:8000/platforms/loyalty/storefront/FASHIONHUB/account/login",
|
|
}
|
|
request.url = Mock(path="/api/v1/storefront/auth/login")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
mock_store = Mock()
|
|
mock_store.id = 4
|
|
mock_store.name = "Fashion Hub"
|
|
mock_store.subdomain = "fashionhub"
|
|
|
|
referer_context = {
|
|
"subdomain": "FASHIONHUB",
|
|
"detection_method": "path",
|
|
"path_prefix": "/storefront/FASHIONHUB",
|
|
"full_prefix": "/storefront/",
|
|
"host": "localhost",
|
|
"referer": "http://localhost:8000/platforms/loyalty/storefront/FASHIONHUB/account/login",
|
|
}
|
|
|
|
mock_db = MagicMock()
|
|
|
|
with (
|
|
patch.object(
|
|
StoreContextManager,
|
|
"is_api_request",
|
|
return_value=True,
|
|
),
|
|
patch.object(
|
|
StoreContextManager,
|
|
"extract_store_from_referer",
|
|
return_value=referer_context,
|
|
),
|
|
patch.object(
|
|
StoreContextManager,
|
|
"get_store_from_context",
|
|
return_value=mock_store,
|
|
),
|
|
patch("middleware.store_context.get_db", return_value=iter([mock_db])),
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is mock_store
|
|
assert request.state.store_context is referer_context
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_storefront_api_no_referer(self):
|
|
"""Test storefront API request without Referer header sets store to None."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/api/v1/storefront/auth/login")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with (
|
|
patch.object(
|
|
StoreContextManager,
|
|
"is_api_request",
|
|
return_value=True,
|
|
),
|
|
patch.object(
|
|
StoreContextManager,
|
|
"extract_store_from_referer",
|
|
return_value=None,
|
|
),
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
assert request.state.store_context is None
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_middleware_non_storefront_api_still_skips(self):
|
|
"""Test non-storefront API requests still skip store detection."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/api/v1/admin/stores")
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with patch.object(
|
|
StoreContextManager, "is_api_request", return_value=True
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
call_next.assert_called_once_with(request)
|
|
|
|
# ========================================================================
|
|
# System Path Skipping Tests
|
|
# ========================================================================
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"path",
|
|
[
|
|
"/",
|
|
"/health",
|
|
"/docs",
|
|
"/redoc",
|
|
"/openapi.json",
|
|
],
|
|
)
|
|
async def test_middleware_skips_system_paths(self, path):
|
|
"""Test middleware skips store detection for system paths."""
|
|
middleware = StoreContextMiddleware(app=None)
|
|
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path=path)
|
|
request.state = Mock()
|
|
|
|
call_next = AsyncMock(return_value=Mock())
|
|
|
|
with (
|
|
patch.object(FrontendDetector, "is_admin", return_value=False),
|
|
patch.object(
|
|
StoreContextManager, "is_static_file_request", return_value=False
|
|
),
|
|
):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
assert request.state.store is None
|
|
assert request.state.store_context is None
|
|
assert request.state.clean_path == path
|
|
call_next.assert_called_once_with(request)
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.stores
|
|
class TestHelperFunctions:
|
|
"""Test suite for helper functions."""
|
|
|
|
def test_get_current_store_exists(self):
|
|
"""Test getting current store when it exists."""
|
|
request = Mock(spec=Request)
|
|
mock_store = Mock()
|
|
request.state.store = mock_store
|
|
|
|
store = get_current_store(request)
|
|
|
|
assert store is mock_store
|
|
|
|
def test_get_current_store_not_exists(self):
|
|
"""Test getting current store when it doesn't exist."""
|
|
request = Mock(spec=Request)
|
|
request.state = Mock(spec=[]) # store attribute doesn't exist
|
|
|
|
store = get_current_store(request)
|
|
|
|
assert store is None
|
|
|
|
def test_require_store_context_success(self):
|
|
"""Test require_store_context dependency with store present."""
|
|
request = Mock(spec=Request)
|
|
mock_store = Mock()
|
|
request.state.store = mock_store
|
|
|
|
dependency = require_store_context()
|
|
result = dependency(request)
|
|
|
|
assert result is mock_store
|
|
|
|
def test_require_store_context_failure(self):
|
|
"""Test require_store_context dependency raises StoreNotFoundException when no store."""
|
|
request = Mock(spec=Request)
|
|
request.state.store = None
|
|
|
|
dependency = require_store_context()
|
|
|
|
with pytest.raises(StoreNotFoundException) as exc_info:
|
|
dependency(request)
|
|
|
|
assert exc_info.value.status_code == 404
|
|
assert "Store" in exc_info.value.message
|
|
assert "not found" in exc_info.value.message
|
|
|
|
|
|
@pytest.mark.unit
|
|
@pytest.mark.stores
|
|
class TestEdgeCases:
|
|
"""Test suite for edge cases and error scenarios."""
|
|
|
|
def test_detect_store_context_empty_host(self):
|
|
"""Test store detection with empty host header."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": ""}
|
|
request.url = Mock(path="/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
def test_detect_store_context_missing_host(self):
|
|
"""Test store detection with missing host header."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {}
|
|
request.url = Mock(path="/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = None
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is None
|
|
|
|
def test_detect_store_path_with_trailing_slash(self):
|
|
"""Test path detection with trailing slash."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/store/store1/")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = "/store/store1/"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "path"
|
|
assert context["subdomain"] == "store1"
|
|
|
|
def test_detect_store_path_without_trailing_slash(self):
|
|
"""Test path detection without trailing slash."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "localhost"}
|
|
request.url = Mock(path="/store/store1")
|
|
request.state = Mock()
|
|
request.state.platform_clean_path = "/store/store1"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
assert context["detection_method"] == "path"
|
|
assert context["subdomain"] == "store1"
|
|
|
|
def test_detect_store_complex_subdomain(self):
|
|
"""Test detection with multiple subdomain levels."""
|
|
request = Mock(spec=Request)
|
|
request.headers = {"host": "shop.store1.platform.com"}
|
|
request.url = Mock(path="/")
|
|
|
|
with patch("middleware.store_context.settings") as mock_settings:
|
|
mock_settings.platform_domain = "platform.com"
|
|
|
|
context = StoreContextManager.detect_store_context(request)
|
|
|
|
assert context is not None
|
|
# Should detect 'shop' as subdomain since it's the first part
|
|
assert context["detection_method"] == "subdomain"
|
|
assert context["subdomain"] == "shop"
|
|
|
|
def test_get_store_logs_warning_when_not_found_subdomain(self):
|
|
"""Test that warning is logged when store is not found by subdomain."""
|
|
mock_db = Mock()
|
|
# Mock the complete query chain properly - need to mock filter twice and then first
|
|
mock_query = mock_db.query.return_value
|
|
mock_filter1 = mock_query.filter.return_value
|
|
mock_filter2 = mock_filter1.filter.return_value
|
|
mock_filter2.first.return_value = None
|
|
|
|
context = {"subdomain": "nonexistent", "detection_method": "subdomain"}
|
|
|
|
with patch("middleware.store_context.logger") as mock_logger:
|
|
store = StoreContextManager.get_store_from_context(mock_db, context)
|
|
|
|
assert store is None
|
|
# Verify warning was logged
|
|
mock_logger.warning.assert_called()
|
|
warning_message = str(mock_logger.warning.call_args)
|
|
assert (
|
|
"No active store found for subdomain" in warning_message
|
|
and "nonexistent" in warning_message
|
|
)
|