Files
orion/tests/unit/middleware/test_vendor_context.py
Samir Boulahtit 3d3b8cae22 feat: add platform detail/edit admin UI and service enhancements
- Add platform detail and edit admin pages with templates and JS
- Add ContentPageService methods: list_all_platform_pages, list_all_vendor_defaults
- Deprecate /admin/platform-homepage route (redirects to /admin/platforms)
- Add migration to fix content_page nullable columns
- Refine platform and vendor context middleware
- Add platform context middleware unit tests
- Update platforms.js with improved functionality
- Add section-based homepage plan documentation

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-23 14:08:02 +01:00

1109 lines
41 KiB
Python

# tests/unit/middleware/test_vendor_context.py
"""
Comprehensive unit tests for VendorContextMiddleware and VendorContextManager.
Tests cover:
- Vendor detection from custom domains, subdomains, and path-based routing
- Database lookup and vendor 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.exceptions.vendor import VendorNotFoundException
from middleware.vendor_context import (
VendorContextManager,
VendorContextMiddleware,
get_current_vendor,
require_vendor_context,
)
@pytest.mark.unit
@pytest.mark.vendors
class TestVendorContextManager:
"""Test suite for VendorContextManager static methods."""
# ========================================================================
# Vendor 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.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_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.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_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": "vendor1.platform.com"}
request.url = Mock(path="/")
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_context(request)
assert context is not None
assert context["detection_method"] == "subdomain"
assert context["subdomain"] == "vendor1"
assert context["host"] == "vendor1.platform.com"
def test_detect_subdomain_with_port(self):
"""Test subdomain detection with port number."""
request = Mock(spec=Request)
request.headers = {"host": "vendor1.platform.com:8000"}
request.url = Mock(path="/")
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_context(request)
assert context is not None
assert context["detection_method"] == "subdomain"
assert context["subdomain"] == "vendor1"
def test_detect_path_vendor_singular(self):
"""Test path-based detection with /vendor/ prefix."""
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/vendor/vendor1/shop")
# Set platform_clean_path to simulate PlatformContextMiddleware output
request.state = Mock()
request.state.platform_clean_path = "/vendor/vendor1/shop"
context = VendorContextManager.detect_vendor_context(request)
assert context is not None
assert context["detection_method"] == "path"
assert context["subdomain"] == "vendor1"
assert context["path_prefix"] == "/vendor/vendor1"
assert context["full_prefix"] == "/vendor/"
def test_detect_path_vendors_plural(self):
"""Test path-based detection with /vendors/ prefix."""
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/vendors/vendor1/shop")
# Set platform_clean_path to simulate PlatformContextMiddleware output
request.state = Mock()
request.state.platform_clean_path = "/vendors/vendor1/shop"
context = VendorContextManager.detect_vendor_context(request)
assert context is not None
assert context["detection_method"] == "path"
assert context["subdomain"] == "vendor1"
assert context["path_prefix"] == "/vendors/vendor1"
assert context["full_prefix"] == "/vendors/"
def test_detect_no_vendor_context(self):
"""Test when no vendor 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 = VendorContextManager.detect_vendor_context(request)
assert context is None
def test_ignore_admin_subdomain(self):
"""Test that admin subdomain is not detected as vendor."""
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.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_context(request)
assert context is None
def test_ignore_www_subdomain(self):
"""Test that www subdomain is not detected as vendor."""
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.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_context(request)
assert context is None
def test_ignore_api_subdomain(self):
"""Test that api subdomain is not detected as vendor."""
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.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_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.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_context(request)
assert context is None
# ========================================================================
# Vendor Database Lookup Tests
# ========================================================================
def test_get_vendor_from_custom_domain_context(self):
"""Test getting vendor from custom domain context."""
mock_db = Mock(spec=Session)
mock_vendor_domain = Mock()
mock_vendor = Mock()
mock_vendor.is_active = True
mock_vendor_domain.vendor = mock_vendor
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = mock_vendor_domain
context = {"detection_method": "custom_domain", "domain": "customdomain1.com"}
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is mock_vendor
assert vendor.is_active is True
def test_get_vendor_from_custom_domain_inactive_vendor(self):
"""Test getting inactive vendor from custom domain context."""
mock_db = Mock(spec=Session)
mock_vendor_domain = Mock()
mock_vendor = Mock()
mock_vendor.is_active = False
mock_vendor_domain.vendor = mock_vendor
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = mock_vendor_domain
context = {"detection_method": "custom_domain", "domain": "customdomain1.com"}
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is None
def test_get_vendor_from_custom_domain_not_found(self):
"""Test custom domain not found in database."""
mock_db = Mock(spec=Session)
mock_db.query.return_value.filter.return_value.filter.return_value.filter.return_value.first.return_value = None
context = {"detection_method": "custom_domain", "domain": "nonexistent.com"}
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is None
def test_get_vendor_from_subdomain_context(self):
"""Test getting vendor from subdomain context."""
mock_db = Mock(spec=Session)
mock_vendor = Mock()
mock_vendor.is_active = True
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_vendor
context = {"detection_method": "subdomain", "subdomain": "vendor1"}
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is mock_vendor
def test_get_vendor_from_path_context(self):
"""Test getting vendor from path context."""
mock_db = Mock(spec=Session)
mock_vendor = Mock()
mock_vendor.is_active = True
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_vendor
context = {"detection_method": "path", "subdomain": "vendor1"}
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is mock_vendor
def test_get_vendor_with_no_context(self):
"""Test getting vendor with no context."""
mock_db = Mock(spec=Session)
vendor = VendorContextManager.get_vendor_from_context(mock_db, None)
assert vendor is None
def test_get_vendor_subdomain_case_insensitive(self):
"""Test subdomain lookup is case-insensitive."""
mock_db = Mock(spec=Session)
mock_vendor = Mock()
mock_vendor.is_active = True
mock_db.query.return_value.filter.return_value.filter.return_value.first.return_value = mock_vendor
context = {"detection_method": "subdomain", "subdomain": "VENDOR1"} # Uppercase
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is mock_vendor
# ========================================================================
# Path Extraction Tests
# ========================================================================
def test_extract_clean_path_from_vendor_path(self):
"""Test extracting clean path from /vendor/ prefix."""
request = Mock(spec=Request)
request.url = Mock(path="/vendor/vendor1/shop/products")
vendor_context = {"detection_method": "path", "path_prefix": "/vendor/vendor1"}
clean_path = VendorContextManager.extract_clean_path(request, vendor_context)
assert clean_path == "/shop/products"
def test_extract_clean_path_from_vendors_path(self):
"""Test extracting clean path from /vendors/ prefix."""
request = Mock(spec=Request)
request.url = Mock(path="/vendors/vendor1/shop/products")
vendor_context = {"detection_method": "path", "path_prefix": "/vendors/vendor1"}
clean_path = VendorContextManager.extract_clean_path(request, vendor_context)
assert clean_path == "/shop/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="/vendor/vendor1")
vendor_context = {"detection_method": "path", "path_prefix": "/vendor/vendor1"}
clean_path = VendorContextManager.extract_clean_path(request, vendor_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="/shop/products")
vendor_context = {"detection_method": "subdomain", "subdomain": "vendor1"}
clean_path = VendorContextManager.extract_clean_path(request, vendor_context)
assert clean_path == "/shop/products"
def test_extract_clean_path_no_context(self):
"""Test extracting clean path with no vendor context."""
request = Mock(spec=Request)
request.url = Mock(path="/shop/products")
clean_path = VendorContextManager.extract_clean_path(request, None)
assert clean_path == "/shop/products"
# ========================================================================
# Request Type Detection Tests
# ========================================================================
def test_is_admin_request_admin_subdomain(self):
"""Test admin request detection from subdomain."""
request = Mock(spec=Request)
request.headers = {"host": "admin.platform.com"}
request.url = Mock(path="/dashboard")
assert VendorContextManager.is_admin_request(request) is True
def test_is_admin_request_admin_path(self):
"""Test admin request detection from path."""
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/admin/dashboard")
assert VendorContextManager.is_admin_request(request) is True
def test_is_admin_request_with_port(self):
"""Test admin request detection with port number."""
request = Mock(spec=Request)
request.headers = {"host": "admin.localhost:8000"}
request.url = Mock(path="/dashboard")
assert VendorContextManager.is_admin_request(request) is True
def test_is_not_admin_request(self):
"""Test non-admin request."""
request = Mock(spec=Request)
request.headers = {"host": "vendor1.platform.com"}
request.url = Mock(path="/shop")
assert VendorContextManager.is_admin_request(request) is False
def test_is_api_request(self):
"""Test API request detection."""
request = Mock(spec=Request)
request.url = Mock(path="/api/v1/vendors")
assert VendorContextManager.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="/shop/products")
assert VendorContextManager.is_api_request(request) is False
# ========================================================================
# Shop API Request Detection Tests
# ========================================================================
def test_is_shop_api_request(self):
"""Test shop API request detection."""
request = Mock(spec=Request)
request.url = Mock(path="/api/v1/shop/products")
assert VendorContextManager.is_shop_api_request(request) is True
def test_is_shop_api_request_cart(self):
"""Test shop API request detection for cart endpoint."""
request = Mock(spec=Request)
request.url = Mock(path="/api/v1/shop/cart")
assert VendorContextManager.is_shop_api_request(request) is True
def test_is_not_shop_api_request_admin(self):
"""Test non-shop API request (admin API)."""
request = Mock(spec=Request)
request.url = Mock(path="/api/v1/admin/vendors")
assert VendorContextManager.is_shop_api_request(request) is False
def test_is_not_shop_api_request_vendor(self):
"""Test non-shop API request (vendor API)."""
request = Mock(spec=Request)
request.url = Mock(path="/api/v1/vendor/products")
assert VendorContextManager.is_shop_api_request(request) is False
def test_is_not_shop_api_request_non_api(self):
"""Test non-shop API request (non-API path)."""
request = Mock(spec=Request)
request.url = Mock(path="/shop/products")
assert VendorContextManager.is_shop_api_request(request) is False
# ========================================================================
# Extract Vendor From Referer Tests
# ========================================================================
def test_extract_vendor_from_referer_path_vendors(self):
"""Test extracting vendor from referer with /vendors/ path."""
request = Mock(spec=Request)
request.headers = {
"referer": "http://localhost:8000/vendors/wizamart/shop/products"
}
context = VendorContextManager.extract_vendor_from_referer(request)
assert context is not None
assert context["subdomain"] == "wizamart"
assert context["detection_method"] == "path"
assert context["path_prefix"] == "/vendors/wizamart"
assert context["full_prefix"] == "/vendors/"
def test_extract_vendor_from_referer_path_vendor(self):
"""Test extracting vendor from referer with /vendor/ path."""
request = Mock(spec=Request)
request.headers = {
"referer": "http://localhost:8000/vendor/myshop/shop/products"
}
context = VendorContextManager.extract_vendor_from_referer(request)
assert context is not None
assert context["subdomain"] == "myshop"
assert context["detection_method"] == "path"
assert context["path_prefix"] == "/vendor/myshop"
assert context["full_prefix"] == "/vendor/"
def test_extract_vendor_from_referer_subdomain(self):
"""Test extracting vendor from referer with subdomain."""
request = Mock(spec=Request)
request.headers = {"referer": "http://wizamart.platform.com/shop/products"}
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.extract_vendor_from_referer(request)
assert context is not None
assert context["subdomain"] == "wizamart"
assert context["detection_method"] == "referer_subdomain"
assert context["host"] == "wizamart.platform.com"
def test_extract_vendor_from_referer_custom_domain(self):
"""Test extracting vendor from referer with custom domain."""
request = Mock(spec=Request)
request.headers = {"referer": "http://my-custom-shop.com/shop/products"}
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.extract_vendor_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_vendor_from_referer_no_header(self):
"""Test extracting vendor when no referer header present."""
request = Mock(spec=Request)
request.headers = {}
context = VendorContextManager.extract_vendor_from_referer(request)
assert context is None
def test_extract_vendor_from_referer_origin_header(self):
"""Test extracting vendor from origin header when referer is missing."""
request = Mock(spec=Request)
request.headers = {"origin": "http://localhost:8000/vendors/testshop/shop"}
context = VendorContextManager.extract_vendor_from_referer(request)
assert context is not None
assert context["subdomain"] == "testshop"
def test_extract_vendor_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"}
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.extract_vendor_from_referer(request)
# admin subdomain should not be detected as vendor
assert context is None
def test_extract_vendor_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/shop"}
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.extract_vendor_from_referer(request)
assert context is None
def test_extract_vendor_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/shop"}
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.extract_vendor_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 VendorContextManager.is_static_file_request(request) is True
@pytest.mark.parametrize(
"path",
[
"/shop/products",
"/admin/dashboard",
"/api/vendors",
"/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 VendorContextManager.is_static_file_request(request) is False
@pytest.mark.unit
@pytest.mark.vendors
class TestVendorContextMiddleware:
"""Test suite for VendorContextMiddleware."""
@pytest.mark.asyncio
async def test_middleware_skips_admin_request(self):
"""Test middleware skips vendor detection for admin requests."""
middleware = VendorContextMiddleware(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(VendorContextManager, "is_admin_request", return_value=True):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_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 vendor detection for API requests."""
middleware = VendorContextMiddleware(app=None)
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/api/v1/vendors")
request.state = Mock()
call_next = AsyncMock(return_value=Mock())
with patch.object(VendorContextManager, "is_api_request", return_value=True):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_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 vendor detection for static files."""
middleware = VendorContextMiddleware(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(
VendorContextManager, "is_static_file_request", return_value=True
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_middleware_detects_and_sets_vendor(self):
"""Test middleware successfully detects and sets vendor."""
middleware = VendorContextMiddleware(app=None)
request = Mock(spec=Request)
request.headers = {"host": "vendor1.platform.com"}
request.url = Mock(path="/shop/products")
request.state = Mock()
call_next = AsyncMock(return_value=Mock())
mock_vendor = Mock()
mock_vendor.id = 1
mock_vendor.name = "Test Vendor"
mock_vendor.subdomain = "vendor1"
vendor_context = {"detection_method": "subdomain", "subdomain": "vendor1"}
mock_db = MagicMock()
with (
patch.object(
VendorContextManager,
"detect_vendor_context",
return_value=vendor_context,
),
patch.object(
VendorContextManager,
"get_vendor_from_context",
return_value=mock_vendor,
),
patch.object(
VendorContextManager,
"extract_clean_path",
return_value="/shop/products",
),
patch("middleware.vendor_context.get_db", return_value=iter([mock_db])),
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is mock_vendor
assert request.state.vendor_context == vendor_context
assert request.state.clean_path == "/shop/products"
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_middleware_vendor_not_found(self):
"""Test middleware when vendor context detected but vendor not in database."""
middleware = VendorContextMiddleware(app=None)
request = Mock(spec=Request)
request.headers = {"host": "nonexistent.platform.com"}
request.url = Mock(path="/shop")
request.state = Mock()
call_next = AsyncMock(return_value=Mock())
vendor_context = {"detection_method": "subdomain", "subdomain": "nonexistent"}
mock_db = MagicMock()
with (
patch.object(
VendorContextManager,
"detect_vendor_context",
return_value=vendor_context,
),
patch.object(
VendorContextManager, "get_vendor_from_context", return_value=None
),
patch("middleware.vendor_context.get_db", return_value=iter([mock_db])),
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_context == vendor_context
assert request.state.clean_path == "/shop"
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_middleware_no_vendor_context(self):
"""Test middleware when no vendor context detected."""
middleware = VendorContextMiddleware(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(
VendorContextManager, "detect_vendor_context", return_value=None
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_context is None
assert request.state.clean_path == "/random/path"
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 vendor detection for system paths."""
middleware = VendorContextMiddleware(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(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_context is None
assert request.state.clean_path == path
call_next.assert_called_once_with(request)
# ========================================================================
# Shop API Request Handling Tests
# ========================================================================
@pytest.mark.asyncio
async def test_middleware_shop_api_with_referer_vendor_found(self):
"""Test middleware handles shop API request with vendor from Referer."""
middleware = VendorContextMiddleware(app=None)
request = Mock(spec=Request)
request.headers = {
"host": "localhost",
"referer": "http://localhost:8000/vendors/wizamart/shop/products",
}
request.url = Mock(path="/api/v1/shop/cart")
request.state = Mock()
call_next = AsyncMock(return_value=Mock())
mock_vendor = Mock()
mock_vendor.id = 1
mock_vendor.name = "Wizamart"
mock_vendor.subdomain = "wizamart"
vendor_context = {
"subdomain": "wizamart",
"detection_method": "path",
"path_prefix": "/vendors/wizamart",
"full_prefix": "/vendors/",
}
mock_db = MagicMock()
with (
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
patch.object(
VendorContextManager, "is_shop_api_request", return_value=True
),
patch.object(
VendorContextManager,
"extract_vendor_from_referer",
return_value=vendor_context,
),
patch.object(
VendorContextManager,
"get_vendor_from_context",
return_value=mock_vendor,
),
patch("middleware.vendor_context.get_db", return_value=iter([mock_db])),
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is mock_vendor
assert request.state.vendor_context == vendor_context
assert request.state.clean_path == "/api/v1/shop/cart"
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_middleware_shop_api_with_referer_vendor_not_found(self):
"""Test middleware handles shop API when vendor from Referer not in database."""
middleware = VendorContextMiddleware(app=None)
request = Mock(spec=Request)
request.headers = {
"host": "localhost",
"referer": "http://localhost:8000/vendors/nonexistent/shop/products",
}
request.url = Mock(path="/api/v1/shop/cart")
request.state = Mock()
call_next = AsyncMock(return_value=Mock())
vendor_context = {
"subdomain": "nonexistent",
"detection_method": "path",
"path_prefix": "/vendors/nonexistent",
"full_prefix": "/vendors/",
}
mock_db = MagicMock()
with (
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
patch.object(
VendorContextManager, "is_shop_api_request", return_value=True
),
patch.object(
VendorContextManager,
"extract_vendor_from_referer",
return_value=vendor_context,
),
patch.object(
VendorContextManager, "get_vendor_from_context", return_value=None
),
patch("middleware.vendor_context.get_db", return_value=iter([mock_db])),
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_context == vendor_context
assert request.state.clean_path == "/api/v1/shop/cart"
call_next.assert_called_once_with(request)
@pytest.mark.asyncio
async def test_middleware_shop_api_without_referer(self):
"""Test middleware handles shop API request without Referer header."""
middleware = VendorContextMiddleware(app=None)
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/api/v1/shop/products")
request.state = Mock()
call_next = AsyncMock(return_value=Mock())
with (
patch.object(VendorContextManager, "is_admin_request", return_value=False),
patch.object(
VendorContextManager, "is_static_file_request", return_value=False
),
patch.object(
VendorContextManager, "is_shop_api_request", return_value=True
),
patch.object(
VendorContextManager, "extract_vendor_from_referer", return_value=None
),
):
await middleware.dispatch(request, call_next)
assert request.state.vendor is None
assert request.state.vendor_context is None
assert request.state.clean_path == "/api/v1/shop/products"
call_next.assert_called_once_with(request)
@pytest.mark.unit
@pytest.mark.vendors
class TestHelperFunctions:
"""Test suite for helper functions."""
def test_get_current_vendor_exists(self):
"""Test getting current vendor when it exists."""
request = Mock(spec=Request)
mock_vendor = Mock()
request.state.vendor = mock_vendor
vendor = get_current_vendor(request)
assert vendor is mock_vendor
def test_get_current_vendor_not_exists(self):
"""Test getting current vendor when it doesn't exist."""
request = Mock(spec=Request)
request.state = Mock(spec=[]) # vendor attribute doesn't exist
vendor = get_current_vendor(request)
assert vendor is None
def test_require_vendor_context_success(self):
"""Test require_vendor_context dependency with vendor present."""
request = Mock(spec=Request)
mock_vendor = Mock()
request.state.vendor = mock_vendor
dependency = require_vendor_context()
result = dependency(request)
assert result is mock_vendor
def test_require_vendor_context_failure(self):
"""Test require_vendor_context dependency raises VendorNotFoundException when no vendor."""
request = Mock(spec=Request)
request.state.vendor = None
dependency = require_vendor_context()
with pytest.raises(VendorNotFoundException) as exc_info:
dependency(request)
assert exc_info.value.status_code == 404
assert "Vendor" in exc_info.value.message
assert "not found" in exc_info.value.message
@pytest.mark.unit
@pytest.mark.vendors
class TestEdgeCases:
"""Test suite for edge cases and error scenarios."""
def test_detect_vendor_context_empty_host(self):
"""Test vendor 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 = VendorContextManager.detect_vendor_context(request)
assert context is None
def test_detect_vendor_context_missing_host(self):
"""Test vendor 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 = VendorContextManager.detect_vendor_context(request)
assert context is None
def test_detect_vendor_path_with_trailing_slash(self):
"""Test path detection with trailing slash."""
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/vendor/vendor1/")
request.state = Mock()
request.state.platform_clean_path = "/vendor/vendor1/"
context = VendorContextManager.detect_vendor_context(request)
assert context is not None
assert context["detection_method"] == "path"
assert context["subdomain"] == "vendor1"
def test_detect_vendor_path_without_trailing_slash(self):
"""Test path detection without trailing slash."""
request = Mock(spec=Request)
request.headers = {"host": "localhost"}
request.url = Mock(path="/vendor/vendor1")
request.state = Mock()
request.state.platform_clean_path = "/vendor/vendor1"
context = VendorContextManager.detect_vendor_context(request)
assert context is not None
assert context["detection_method"] == "path"
assert context["subdomain"] == "vendor1"
def test_detect_vendor_complex_subdomain(self):
"""Test detection with multiple subdomain levels."""
request = Mock(spec=Request)
request.headers = {"host": "shop.vendor1.platform.com"}
request.url = Mock(path="/")
with patch("middleware.vendor_context.settings") as mock_settings:
mock_settings.platform_domain = "platform.com"
context = VendorContextManager.detect_vendor_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_vendor_logs_warning_when_not_found_subdomain(self):
"""Test that warning is logged when vendor 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.vendor_context.logger") as mock_logger:
vendor = VendorContextManager.get_vendor_from_context(mock_db, context)
assert vendor is None
# Verify warning was logged
mock_logger.warning.assert_called()
warning_message = str(mock_logger.warning.call_args)
assert (
"No active vendor found for subdomain" in warning_message
and "nonexistent" in warning_message
)