sboulahtit/orion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adding integration tests for the middleware layer

Browse Source
This commit is contained in:
Samir Boulahtit
2025-11-19 22:52:43 +01:00
parent 23cf568c82
commit 96803c2708
6 changed files with 1683 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/integration/middleware/__init__.py Normal file
View File

@@ -0,0 +1,7 @@
# tests/integration/middleware/__init__.py
"""
Integration tests for middleware stack.
These tests verify the full middleware stack works correctly with real HTTP requests,
ensuring that vendor context, request context, and theme are properly detected and injected.
"""

92
tests/integration/middleware/conftest.py Normal file
View File

@@ -0,0 +1,92 @@
# tests/integration/middleware/conftest.py
"""
Fixtures specific to middleware integration tests.
"""
import pytest
from models.database.vendor import Vendor
from models.database.vendor_domain import VendorDomain
from models.database.vendor_theme import VendorTheme
@pytest.fixture
def vendor_with_subdomain(db):
"""Create a vendor with subdomain for testing."""
vendor = Vendor(
name="Test Vendor",
code="testvendor",
subdomain="testvendor",
is_active=True
)
db.add(vendor)
db.commit()
db.refresh(vendor)
return vendor
@pytest.fixture
def vendor_with_custom_domain(db):
"""Create a vendor with custom domain for testing."""
vendor = Vendor(
name="Custom Domain Vendor",
code="customvendor",
subdomain="customvendor",
is_active=True
)
db.add(vendor)
db.commit()
db.refresh(vendor)
# Add custom domain
domain = VendorDomain(
vendor_id=vendor.id,
domain="customdomain.com",
is_active=True,
is_primary=True
)
db.add(domain)
db.commit()
return vendor
@pytest.fixture
def vendor_with_theme(db):
"""Create a vendor with custom theme for testing."""
vendor = Vendor(
name="Themed Vendor",
code="themedvendor",
subdomain="themedvendor",
is_active=True
)
db.add(vendor)
db.commit()
db.refresh(vendor)
# Add custom theme
theme = VendorTheme(
vendor_id=vendor.id,
primary_color="#FF5733",
secondary_color="#33FF57",
logo_url="/static/vendors/themedvendor/logo.png",
favicon_url="/static/vendors/themedvendor/favicon.ico",
custom_css="body { background: #FF5733; }"
)
db.add(theme)
db.commit()
return vendor
@pytest.fixture
def inactive_vendor(db):
"""Create an inactive vendor for testing."""
vendor = Vendor(
name="Inactive Vendor",
code="inactive",
subdomain="inactive",
is_active=False
)
db.add(vendor)
db.commit()
db.refresh(vendor)
return vendor

462
tests/integration/middleware/test_context_detection_flow.py Normal file
View File

@@ -0,0 +1,462 @@
# tests/integration/middleware/test_context_detection_flow.py
"""
Integration tests for request context detection end-to-end flow.
These tests verify that context type (API, ADMIN, VENDOR_DASHBOARD, SHOP, FALLBACK)
is correctly detected through real HTTP requests.
"""
import pytest
from unittest.mock import patch
from middleware.context import RequestContext
@pytest.mark.integration
@pytest.mark.middleware
@pytest.mark.context
class TestContextDetectionFlow:
"""Test context type detection through real HTTP requests."""
# ========================================================================
# API Context Detection Tests
# ========================================================================
def test_api_path_detected_as_api_context(self, client):
"""Test that /api/* paths are detected as API context."""
from fastapi import Request
from main import app
@app.get("/api/test-api-context")
async def test_api(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"context_enum": request.state.context_type.name if hasattr(request.state, 'context_type') else None
}
response = client.get("/api/test-api-context")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "api"
assert data["context_enum"] == "API"
def test_nested_api_path_detected_as_api_context(self, client):
"""Test that nested /api/ paths are detected as API context."""
from fastapi import Request
from main import app
@app.get("/api/v1/vendor/products")
async def test_nested_api(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
response = client.get("/api/v1/vendor/products")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "api"
# ========================================================================
# Admin Context Detection Tests
# ========================================================================
def test_admin_path_detected_as_admin_context(self, client):
"""Test that /admin/* paths are detected as ADMIN context."""
from fastapi import Request
from main import app
@app.get("/admin/test-admin-context")
async def test_admin(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"context_enum": request.state.context_type.name if hasattr(request.state, 'context_type') else None
}
response = client.get("/admin/test-admin-context")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "admin"
assert data["context_enum"] == "ADMIN"
def test_admin_subdomain_detected_as_admin_context(self, client):
"""Test that admin.* subdomain is detected as ADMIN context."""
from fastapi import Request
from main import app
@app.get("/test-admin-subdomain-context")
async def test_admin_subdomain(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-admin-subdomain-context",
headers={"host": "admin.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "admin"
def test_nested_admin_path_detected_as_admin_context(self, client):
"""Test that nested /admin/ paths are detected as ADMIN context."""
from fastapi import Request
from main import app
@app.get("/admin/vendors/123/edit")
async def test_nested_admin(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
response = client.get("/admin/vendors/123/edit")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "admin"
# ========================================================================
# Vendor Dashboard Context Detection Tests
# ========================================================================
def test_vendor_dashboard_path_detected(self, client, vendor_with_subdomain):
"""Test that /vendor/* paths are detected as VENDOR_DASHBOARD context."""
from fastapi import Request
from main import app
@app.get("/vendor/test-vendor-dashboard")
async def test_vendor_dashboard(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"context_enum": request.state.context_type.name if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/vendor/test-vendor-dashboard",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "vendor_dashboard"
assert data["context_enum"] == "VENDOR_DASHBOARD"
assert data["has_vendor"] is True
def test_nested_vendor_dashboard_path_detected(self, client, vendor_with_subdomain):
"""Test that nested /vendor/ paths are detected as VENDOR_DASHBOARD context."""
from fastapi import Request
from main import app
@app.get("/vendor/products/123/edit")
async def test_nested_vendor(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/vendor/products/123/edit",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "vendor_dashboard"
# ========================================================================
# Shop Context Detection Tests
# ========================================================================
def test_shop_path_with_vendor_detected_as_shop(self, client, vendor_with_subdomain):
"""Test that /shop/* paths with vendor are detected as SHOP context."""
from fastapi import Request
from main import app
@app.get("/shop/test-shop-context")
async def test_shop(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"context_enum": request.state.context_type.name if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/shop/test-shop-context",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "shop"
assert data["context_enum"] == "SHOP"
assert data["has_vendor"] is True
def test_root_path_with_vendor_detected_as_shop(self, client, vendor_with_subdomain):
"""Test that root path with vendor is detected as SHOP context."""
from fastapi import Request
from main import app
@app.get("/test-root-shop")
async def test_root_shop(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-root-shop",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# Root path with vendor should be SHOP context
assert data["context_type"] == "shop"
assert data["has_vendor"] is True
def test_custom_domain_shop_detected(self, client, vendor_with_custom_domain):
"""Test that custom domain shop is detected as SHOP context."""
from fastapi import Request
from main import app
@app.get("/products")
async def test_custom_domain_shop(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/products",
headers={"host": "customdomain.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "shop"
assert data["vendor_code"] == vendor_with_custom_domain.code
# ========================================================================
# Fallback Context Detection Tests
# ========================================================================
def test_unknown_path_without_vendor_fallback_context(self, client):
"""Test that unknown paths without vendor get FALLBACK context."""
from fastapi import Request
from main import app
@app.get("/test-fallback-context")
async def test_fallback(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"context_enum": request.state.context_type.name if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-fallback-context",
headers={"host": "platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "fallback"
assert data["context_enum"] == "FALLBACK"
assert data["has_vendor"] is False
# ========================================================================
# Context Priority Tests (Path takes precedence)
# ========================================================================
def test_api_path_overrides_vendor_context(self, client, vendor_with_subdomain):
"""Test that /api/* path sets API context even with vendor."""
from fastapi import Request
from main import app
@app.get("/api/test-api-priority")
async def test_api_priority(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/api/test-api-priority",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# API path should override vendor context
assert data["context_type"] == "api"
# But vendor should still be detected
assert data["has_vendor"] is True
def test_admin_path_overrides_vendor_context(self, client, vendor_with_subdomain):
"""Test that /admin/* path sets ADMIN context even with vendor."""
from fastapi import Request
from main import app
@app.get("/admin/test-admin-priority")
async def test_admin_priority(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/admin/test-admin-priority",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# Admin path should override vendor context
assert data["context_type"] == "admin"
def test_vendor_dashboard_overrides_shop_context(self, client, vendor_with_subdomain):
"""Test that /vendor/* path sets VENDOR_DASHBOARD, not SHOP."""
from fastapi import Request
from main import app
@app.get("/vendor/test-priority")
async def test_vendor_priority(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/vendor/test-priority",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# Should be VENDOR_DASHBOARD, not SHOP
assert data["context_type"] == "vendor_dashboard"
# ========================================================================
# Context Detection with Clean Path Tests
# ========================================================================
def test_context_uses_clean_path_for_detection(self, client, vendor_with_subdomain):
"""Test that context detection uses clean_path, not original path."""
from fastapi import Request
from main import app
@app.get("/vendors/{vendor_code}/shop/products")
async def test_clean_path_context(vendor_code: str, request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"clean_path": request.state.clean_path if hasattr(request.state, 'clean_path') else None,
"original_path": request.url.path
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
f"/vendors/{vendor_with_subdomain.code}/shop/products",
headers={"host": "localhost:8000"}
)
assert response.status_code == 200
data = response.json()
# Should detect SHOP context based on clean_path (/shop/products)
# not original path (/vendors/{code}/shop/products)
assert data["context_type"] == "shop"
assert "/shop/products" in data["clean_path"]
# ========================================================================
# Context Enum Value Tests
# ========================================================================
def test_context_type_is_enum_instance(self, client):
"""Test that context_type is a RequestContext enum instance."""
from fastapi import Request
from main import app
@app.get("/api/test-enum")
async def test_enum(request: Request):
context = request.state.context_type if hasattr(request.state, 'context_type') else None
return {
"is_enum": isinstance(context, RequestContext) if context else False,
"enum_name": context.name if context else None,
"enum_value": context.value if context else None
}
response = client.get("/api/test-enum")
assert response.status_code == 200
data = response.json()
assert data["is_enum"] is True
assert data["enum_name"] == "API"
assert data["enum_value"] == "api"
# ========================================================================
# Edge Cases
# ========================================================================
def test_empty_path_with_vendor_detected_as_shop(self, client, vendor_with_subdomain):
"""Test that empty/root path with vendor is detected as SHOP."""
from fastapi import Request
from main import app
@app.get("/")
async def test_root(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code in [200, 404] # Might not have root handler
if response.status_code == 200:
data = response.json()
assert data["context_type"] == "shop"
assert data["has_vendor"] is True
def test_case_insensitive_context_detection(self, client):
"""Test that context detection is case insensitive for paths."""
from fastapi import Request
from main import app
@app.get("/API/test-case")
@app.get("/api/test-case")
async def test_case(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
# Test uppercase
response = client.get("/API/test-case")
if response.status_code == 200:
data = response.json()
# Should still detect as API (depending on implementation)
assert data["context_type"] in ["api", "fallback"]

298
tests/integration/middleware/test_middleware_stack.py Normal file
View File

@@ -0,0 +1,298 @@
# tests/integration/middleware/test_middleware_stack.py
"""
Integration tests for the complete middleware stack.
These tests verify that all middleware components work together correctly
through real HTTP requests, ensuring proper execution order and state injection.
"""
import pytest
from unittest.mock import patch
from middleware.context import RequestContext
@pytest.mark.integration
@pytest.mark.middleware
class TestMiddlewareStackIntegration:
"""Test the full middleware stack with real HTTP requests."""
# ========================================================================
# Admin Context Tests
# ========================================================================
def test_admin_path_sets_admin_context(self, client):
"""Test that /admin/* paths set ADMIN context type."""
# Create a simple endpoint to inspect request state
from fastapi import Request
from main import app
@app.get("/admin/test-context")
async def test_admin_context(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"has_theme": hasattr(request.state, 'theme')
}
response = client.get("/admin/test-context")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "admin"
# Admin context typically doesn't require vendor, but might have one
# The key assertion is that context_type is correctly set to admin
def test_admin_subdomain_sets_admin_context(self, client):
"""Test that admin.* subdomain sets ADMIN context type."""
from fastapi import Request
from main import app
@app.get("/test-admin-subdomain")
async def test_admin_subdomain(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
# Simulate request with admin subdomain
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-admin-subdomain",
headers={"host": "admin.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "admin"
# ========================================================================
# API Context Tests
# ========================================================================
def test_api_path_sets_api_context(self, client):
"""Test that /api/* paths set API context type."""
from fastapi import Request
from main import app
@app.get("/api/test-context")
async def test_api_context(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
response = client.get("/api/test-context")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "api"
# ========================================================================
# Vendor Dashboard Context Tests
# ========================================================================
def test_vendor_dashboard_path_sets_vendor_context(self, client, vendor_with_subdomain):
"""Test that /vendor/* paths with vendor set VENDOR_DASHBOARD context."""
from fastapi import Request
from main import app
@app.get("/vendor/test-context")
async def test_vendor_context(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') else None
}
# Request with vendor subdomain
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/vendor/test-context",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "vendor_dashboard"
assert data["vendor_id"] == vendor_with_subdomain.id
assert data["vendor_code"] == vendor_with_subdomain.code
# ========================================================================
# Shop Context Tests
# ========================================================================
def test_shop_path_with_subdomain_sets_shop_context(self, client, vendor_with_subdomain):
"""Test that /shop/* paths with vendor subdomain set SHOP context."""
from fastapi import Request
from main import app
@app.get("/shop/test-context")
async def test_shop_context(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None,
"has_theme": hasattr(request.state, 'theme')
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/shop/test-context",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "shop"
assert data["vendor_id"] == vendor_with_subdomain.id
assert data["has_theme"] is True
def test_shop_path_with_custom_domain_sets_shop_context(self, client, vendor_with_custom_domain):
"""Test that /shop/* paths with custom domain set SHOP context."""
from fastapi import Request
from main import app
@app.get("/shop/test-custom-domain")
async def test_shop_custom_domain(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/shop/test-custom-domain",
headers={"host": "customdomain.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "shop"
assert data["vendor_id"] == vendor_with_custom_domain.id
# ========================================================================
# Middleware Execution Order Tests
# ========================================================================
def test_vendor_context_runs_before_context_detection(self, client, vendor_with_subdomain):
"""Test that VendorContextMiddleware runs before ContextDetectionMiddleware."""
from fastapi import Request
from main import app
@app.get("/test-execution-order")
async def test_execution_order(request: Request):
# If vendor context runs first, clean_path should be available
# before context detection uses it
return {
"has_vendor": hasattr(request.state, 'vendor'),
"has_clean_path": hasattr(request.state, 'clean_path'),
"has_context_type": hasattr(request.state, 'context_type')
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-execution-order",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# All middleware should have run and set their state
assert data["has_vendor"] is True
assert data["has_clean_path"] is True
assert data["has_context_type"] is True
def test_theme_context_runs_after_vendor_context(self, client, vendor_with_theme):
"""Test that ThemeContextMiddleware runs after VendorContextMiddleware."""
from fastapi import Request
from main import app
@app.get("/test-theme-loading")
async def test_theme_loading(request: Request):
return {
"has_vendor": hasattr(request.state, 'vendor'),
"has_theme": hasattr(request.state, 'theme'),
"theme_primary_color": request.state.theme.get('primary_color') if hasattr(request.state, 'theme') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-theme-loading",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_vendor"] is True
assert data["has_theme"] is True
assert data["theme_primary_color"] == "#FF5733"
# ========================================================================
# Static File Handling Tests
# ========================================================================
def test_static_files_skip_vendor_detection(self, client):
"""Test that static file requests skip vendor detection."""
# Static file requests should not trigger vendor detection
response = client.get("/static/css/style.css")
# We expect 404 (file doesn't exist) but middleware should have run
# The important thing is it doesn't crash trying to detect vendor
assert response.status_code in [404, 200] # Depending on if file exists
# ========================================================================
# Error Handling Tests
# ========================================================================
def test_missing_vendor_graceful_handling(self, client):
"""Test that missing vendor is handled gracefully."""
from fastapi import Request
from main import app
@app.get("/test-missing-vendor")
async def test_missing_vendor(request: Request):
return {
"has_vendor": hasattr(request.state, 'vendor'),
"vendor": request.state.vendor if hasattr(request.state, 'vendor') else None,
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-missing-vendor",
headers={"host": "nonexistent.platform.com"}
)
assert response.status_code == 200
data = response.json()
# Should handle missing vendor gracefully
assert data["has_vendor"] is False or data["vendor"] is None
# Should still set a context type (fallback)
assert data["context_type"] is not None
def test_inactive_vendor_not_loaded(self, client, inactive_vendor):
"""Test that inactive vendors are not loaded."""
from fastapi import Request
from main import app
@app.get("/test-inactive-vendor")
async def test_inactive_vendor_endpoint(request: Request):
return {
"has_vendor": hasattr(request.state, 'vendor'),
"vendor": request.state.vendor if hasattr(request.state, 'vendor') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-inactive-vendor",
headers={"host": f"{inactive_vendor.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# Inactive vendor should not be loaded
assert data["has_vendor"] is False or data["vendor"] is None

452
tests/integration/middleware/test_theme_loading_flow.py Normal file
View File

@@ -0,0 +1,452 @@
# tests/integration/middleware/test_theme_loading_flow.py
"""
Integration tests for theme loading end-to-end flow.
These tests verify that vendor themes are correctly loaded and injected
into request.state through real HTTP requests.
"""
import pytest
from unittest.mock import patch
@pytest.mark.integration
@pytest.mark.middleware
@pytest.mark.theme
class TestThemeLoadingFlow:
"""Test theme loading through real HTTP requests."""
# ========================================================================
# Basic Theme Loading Tests
# ========================================================================
def test_theme_loaded_for_vendor_with_custom_theme(self, client, vendor_with_theme):
"""Test that custom theme is loaded for vendor with theme."""
from fastapi import Request
from main import app
@app.get("/test-theme-loading")
async def test_theme(request: Request):
theme = request.state.theme if hasattr(request.state, 'theme') else None
return {
"has_theme": theme is not None,
"theme_data": theme if theme else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-theme-loading",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_theme"] is True
assert data["theme_data"] is not None
assert data["theme_data"]["primary_color"] == "#FF5733"
assert data["theme_data"]["secondary_color"] == "#33FF57"
def test_default_theme_loaded_for_vendor_without_theme(self, client, vendor_with_subdomain):
"""Test that default theme is loaded for vendor without custom theme."""
from fastapi import Request
from main import app
@app.get("/test-default-theme")
async def test_default_theme(request: Request):
theme = request.state.theme if hasattr(request.state, 'theme') else None
return {
"has_theme": theme is not None,
"theme_data": theme if theme else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-default-theme",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_theme"] is True
# Default theme should have basic structure
assert "primary_color" in data["theme_data"]
assert "secondary_color" in data["theme_data"]
def test_no_theme_loaded_without_vendor(self, client):
"""Test that no theme is loaded when there's no vendor."""
from fastapi import Request
from main import app
@app.get("/test-no-theme")
async def test_no_theme(request: Request):
return {
"has_theme": hasattr(request.state, 'theme'),
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-no-theme",
headers={"host": "platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_vendor"] is False
# No vendor means no theme should be loaded
assert data["has_theme"] is False
# ========================================================================
# Theme Structure Tests
# ========================================================================
def test_custom_theme_contains_all_fields(self, client, vendor_with_theme):
"""Test that custom theme contains all expected fields."""
from fastapi import Request
from main import app
@app.get("/test-theme-fields")
async def test_theme_fields(request: Request):
theme = request.state.theme if hasattr(request.state, 'theme') else {}
return {
"primary_color": theme.get("primary_color"),
"secondary_color": theme.get("secondary_color"),
"logo_url": theme.get("logo_url"),
"favicon_url": theme.get("favicon_url"),
"custom_css": theme.get("custom_css")
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-theme-fields",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["primary_color"] == "#FF5733"
assert data["secondary_color"] == "#33FF57"
assert data["logo_url"] == "/static/vendors/themedvendor/logo.png"
assert data["favicon_url"] == "/static/vendors/themedvendor/favicon.ico"
assert "background" in data["custom_css"]
def test_default_theme_structure(self, client, vendor_with_subdomain):
"""Test that default theme has expected structure."""
from fastapi import Request
from main import app
@app.get("/test-default-theme-structure")
async def test_default_structure(request: Request):
theme = request.state.theme if hasattr(request.state, 'theme') else {}
return {
"has_primary_color": "primary_color" in theme,
"has_secondary_color": "secondary_color" in theme,
"has_logo_url": "logo_url" in theme,
"has_favicon_url": "favicon_url" in theme,
"has_custom_css": "custom_css" in theme
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-default-theme-structure",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# Default theme should have basic structure
assert data["has_primary_color"] is True
assert data["has_secondary_color"] is True
# ========================================================================
# Theme Loading for Different Contexts Tests
# ========================================================================
def test_theme_loaded_in_shop_context(self, client, vendor_with_theme):
"""Test that theme is loaded in SHOP context."""
from fastapi import Request
from main import app
@app.get("/shop/test-shop-theme")
async def test_shop_theme(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_theme": hasattr(request.state, 'theme'),
"theme_primary": request.state.theme.get("primary_color") if hasattr(request.state, 'theme') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/shop/test-shop-theme",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "shop"
assert data["has_theme"] is True
assert data["theme_primary"] == "#FF5733"
def test_theme_loaded_in_vendor_dashboard_context(self, client, vendor_with_theme):
"""Test that theme is loaded in VENDOR_DASHBOARD context."""
from fastapi import Request
from main import app
@app.get("/vendor/test-dashboard-theme")
async def test_dashboard_theme(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_theme": hasattr(request.state, 'theme'),
"theme_secondary": request.state.theme.get("secondary_color") if hasattr(request.state, 'theme') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/vendor/test-dashboard-theme",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "vendor_dashboard"
assert data["has_theme"] is True
assert data["theme_secondary"] == "#33FF57"
def test_theme_loaded_in_api_context_with_vendor(self, client, vendor_with_theme):
"""Test that theme is loaded in API context when vendor is present."""
from fastapi import Request
from main import app
@app.get("/api/test-api-theme")
async def test_api_theme(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"has_theme": hasattr(request.state, 'theme')
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/api/test-api-theme",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "api"
assert data["has_vendor"] is True
# Theme should be loaded even for API context if vendor present
assert data["has_theme"] is True
def test_no_theme_in_admin_context(self, client):
"""Test that theme is not loaded in ADMIN context (no vendor)."""
from fastapi import Request
from main import app
@app.get("/admin/test-admin-no-theme")
async def test_admin_no_theme(request: Request):
return {
"context_type": request.state.context_type.value if hasattr(request.state, 'context_type') else None,
"has_theme": hasattr(request.state, 'theme')
}
response = client.get("/admin/test-admin-no-theme")
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "admin"
# Admin context has no vendor, so no theme
assert data["has_theme"] is False
# ========================================================================
# Theme Loading with Different Routing Modes Tests
# ========================================================================
def test_theme_loaded_with_subdomain_routing(self, client, vendor_with_theme):
"""Test theme loading with subdomain routing."""
from fastapi import Request
from main import app
@app.get("/test-subdomain-theme")
async def test_subdomain_theme(request: Request):
return {
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None,
"theme_logo": request.state.theme.get("logo_url") if hasattr(request.state, 'theme') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-subdomain-theme",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_code"] == vendor_with_theme.code
assert data["theme_logo"] == "/static/vendors/themedvendor/logo.png"
def test_theme_loaded_with_custom_domain_routing(self, client, vendor_with_custom_domain, db):
"""Test theme loading with custom domain routing."""
# Add theme to custom domain vendor
from models.database.vendor_theme import VendorTheme
theme = VendorTheme(
vendor_id=vendor_with_custom_domain.id,
primary_color="#123456",
secondary_color="#654321"
)
db.add(theme)
db.commit()
from fastapi import Request
from main import app
@app.get("/test-custom-domain-theme")
async def test_custom_domain_theme(request: Request):
return {
"has_theme": hasattr(request.state, 'theme'),
"theme_primary": request.state.theme.get("primary_color") if hasattr(request.state, 'theme') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-custom-domain-theme",
headers={"host": "customdomain.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_theme"] is True
assert data["theme_primary"] == "#123456"
# ========================================================================
# Theme Dependency on Vendor Context Tests
# ========================================================================
def test_theme_middleware_depends_on_vendor_middleware(self, client, vendor_with_theme):
"""Test that theme loading depends on vendor being detected first."""
from fastapi import Request
from main import app
@app.get("/test-theme-vendor-dependency")
async def test_dependency(request: Request):
return {
"has_vendor": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None,
"has_theme": hasattr(request.state, 'theme'),
"vendor_matches_theme": (
request.state.vendor_id == vendor_with_theme.id
if hasattr(request.state, 'vendor_id') else False
)
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-theme-vendor-dependency",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_vendor"] is True
assert data["vendor_id"] == vendor_with_theme.id
assert data["has_theme"] is True
assert data["vendor_matches_theme"] is True
# ========================================================================
# Theme Caching and Performance Tests
# ========================================================================
def test_theme_loaded_consistently_across_requests(self, client, vendor_with_theme):
"""Test that theme is loaded consistently across multiple requests."""
from fastapi import Request
from main import app
@app.get("/test-theme-consistency")
async def test_consistency(request: Request):
return {
"theme_primary": request.state.theme.get("primary_color") if hasattr(request.state, 'theme') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
# Make multiple requests
responses = []
for _ in range(3):
response = client.get(
"/test-theme-consistency",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
responses.append(response.json())
# All responses should have same theme
assert all(r["theme_primary"] == "#FF5733" for r in responses)
# ========================================================================
# Edge Cases and Error Handling Tests
# ========================================================================
def test_theme_gracefully_handles_missing_theme_fields(self, client, vendor_with_subdomain):
"""Test that missing theme fields are handled gracefully."""
from fastapi import Request
from main import app
@app.get("/test-partial-theme")
async def test_partial_theme(request: Request):
theme = request.state.theme if hasattr(request.state, 'theme') else {}
return {
"has_theme": bool(theme),
"primary_color": theme.get("primary_color", "default"),
"logo_url": theme.get("logo_url", "default")
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-partial-theme",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_theme"] is True
# Should have defaults for missing fields
assert data["primary_color"] is not None
assert data["logo_url"] is not None
def test_theme_dict_is_mutable(self, client, vendor_with_theme):
"""Test that theme dict can be accessed and read from."""
from fastapi import Request
from main import app
@app.get("/test-theme-mutable")
async def test_mutable(request: Request):
theme = request.state.theme if hasattr(request.state, 'theme') else {}
# Try to access theme values
primary = theme.get("primary_color")
return {
"can_read": primary is not None,
"value": primary
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-theme-mutable",
headers={"host": f"{vendor_with_theme.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["can_read"] is True
assert data["value"] == "#FF5733"

372
tests/integration/middleware/test_vendor_context_flow.py Normal file
View File

@@ -0,0 +1,372 @@
# tests/integration/middleware/test_vendor_context_flow.py
"""
Integration tests for vendor context detection end-to-end flow.
These tests verify that vendor detection works correctly through real HTTP requests
for all routing modes: subdomain, custom domain, and path-based.
"""
import pytest
from unittest.mock import patch
@pytest.mark.integration
@pytest.mark.middleware
@pytest.mark.vendor
class TestVendorContextFlow:
"""Test vendor context detection through real HTTP requests."""
# ========================================================================
# Subdomain Detection Tests
# ========================================================================
def test_subdomain_vendor_detection(self, client, vendor_with_subdomain):
"""Test vendor detection via subdomain routing."""
from fastapi import Request
from main import app
@app.get("/test-subdomain-detection")
async def test_subdomain(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None,
"vendor_name": request.state.vendor.name if hasattr(request.state, 'vendor') and request.state.vendor else None,
"detection_method": "subdomain"
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-subdomain-detection",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is True
assert data["vendor_id"] == vendor_with_subdomain.id
assert data["vendor_code"] == vendor_with_subdomain.code
assert data["vendor_name"] == vendor_with_subdomain.name
def test_subdomain_with_port_detection(self, client, vendor_with_subdomain):
"""Test vendor detection via subdomain with port number."""
from fastapi import Request
from main import app
@app.get("/test-subdomain-port")
async def test_subdomain_port(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-subdomain-port",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com:8000"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is True
assert data["vendor_code"] == vendor_with_subdomain.code
def test_nonexistent_subdomain_returns_no_vendor(self, client):
"""Test that nonexistent subdomain doesn't crash and returns no vendor."""
from fastapi import Request
from main import app
@app.get("/test-nonexistent-subdomain")
async def test_nonexistent(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor": request.state.vendor if hasattr(request.state, 'vendor') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-nonexistent-subdomain",
headers={"host": "nonexistent.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is False
# ========================================================================
# Custom Domain Detection Tests
# ========================================================================
def test_custom_domain_vendor_detection(self, client, vendor_with_custom_domain):
"""Test vendor detection via custom domain."""
from fastapi import Request
from main import app
@app.get("/test-custom-domain")
async def test_custom_domain(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None,
"detection_method": "custom_domain"
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-custom-domain",
headers={"host": "customdomain.com"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is True
assert data["vendor_id"] == vendor_with_custom_domain.id
assert data["vendor_code"] == vendor_with_custom_domain.code
def test_custom_domain_with_www_detection(self, client, vendor_with_custom_domain):
"""Test vendor detection via custom domain with www prefix."""
from fastapi import Request
from main import app
@app.get("/test-custom-domain-www")
async def test_custom_domain_www(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
# Test with www prefix - should still detect vendor
response = client.get(
"/test-custom-domain-www",
headers={"host": "www.customdomain.com"}
)
# This might fail if your implementation doesn't strip www
# Adjust assertion based on your actual behavior
assert response.status_code == 200
# ========================================================================
# Path-Based Detection Tests (Development Mode)
# ========================================================================
def test_path_based_vendor_detection_vendors_prefix(self, client, vendor_with_subdomain):
"""Test vendor detection via path-based routing with /vendors/ prefix."""
from fastapi import Request
from main import app
@app.get("/vendors/{vendor_code}/test-path")
async def test_path_based(vendor_code: str, request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_code_param": vendor_code,
"vendor_code_state": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None,
"clean_path": request.state.clean_path if hasattr(request.state, 'clean_path') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
f"/vendors/{vendor_with_subdomain.code}/test-path",
headers={"host": "localhost:8000"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is True
assert data["vendor_code_param"] == vendor_with_subdomain.code
assert data["vendor_code_state"] == vendor_with_subdomain.code
def test_path_based_vendor_detection_vendor_prefix(self, client, vendor_with_subdomain):
"""Test vendor detection via path-based routing with /vendor/ prefix."""
from fastapi import Request
from main import app
@app.get("/vendor/{vendor_code}/test")
async def test_vendor_path(vendor_code: str, request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None,
"vendor_code": request.state.vendor.code if hasattr(request.state, 'vendor') and request.state.vendor else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
f"/vendor/{vendor_with_subdomain.code}/test",
headers={"host": "localhost:8000"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is True
assert data["vendor_code"] == vendor_with_subdomain.code
# ========================================================================
# Clean Path Extraction Tests
# ========================================================================
def test_clean_path_extracted_from_vendor_prefix(self, client, vendor_with_subdomain):
"""Test that clean_path is correctly extracted from path-based routing."""
from fastapi import Request
from main import app
@app.get("/vendors/{vendor_code}/shop/products")
async def test_clean_path(vendor_code: str, request: Request):
return {
"clean_path": request.state.clean_path if hasattr(request.state, 'clean_path') else None,
"original_path": request.url.path
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
f"/vendors/{vendor_with_subdomain.code}/shop/products",
headers={"host": "localhost:8000"}
)
assert response.status_code == 200
data = response.json()
# Clean path should have vendor prefix removed
assert data["clean_path"] == "/shop/products"
assert f"/vendors/{vendor_with_subdomain.code}/shop/products" in data["original_path"]
def test_clean_path_unchanged_for_subdomain(self, client, vendor_with_subdomain):
"""Test that clean_path equals original path for subdomain routing."""
from fastapi import Request
from main import app
@app.get("/shop/test-clean-path")
async def test_subdomain_clean_path(request: Request):
return {
"clean_path": request.state.clean_path if hasattr(request.state, 'clean_path') else None,
"original_path": request.url.path
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/shop/test-clean-path",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
# For subdomain routing, clean path should equal original path
assert data["clean_path"] == data["original_path"]
assert data["clean_path"] == "/shop/test-clean-path"
# ========================================================================
# Vendor State Injection Tests
# ========================================================================
def test_vendor_id_injected_into_request_state(self, client, vendor_with_subdomain):
"""Test that vendor_id is correctly injected into request.state."""
from fastapi import Request
from main import app
@app.get("/test-vendor-id-injection")
async def test_vendor_id(request: Request):
return {
"has_vendor_id": hasattr(request.state, 'vendor_id'),
"vendor_id": request.state.vendor_id if hasattr(request.state, 'vendor_id') else None,
"vendor_id_type": type(request.state.vendor_id).__name__ if hasattr(request.state, 'vendor_id') else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-vendor-id-injection",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_vendor_id"] is True
assert data["vendor_id"] == vendor_with_subdomain.id
assert data["vendor_id_type"] == "int"
def test_vendor_object_injected_into_request_state(self, client, vendor_with_subdomain):
"""Test that full vendor object is injected into request.state."""
from fastapi import Request
from main import app
@app.get("/test-vendor-object-injection")
async def test_vendor_object(request: Request):
vendor = request.state.vendor if hasattr(request.state, 'vendor') and request.state.vendor else None
return {
"has_vendor": vendor is not None,
"vendor_attributes": {
"id": vendor.id if vendor else None,
"name": vendor.name if vendor else None,
"code": vendor.code if vendor else None,
"subdomain": vendor.subdomain if vendor else None,
"is_active": vendor.is_active if vendor else None
} if vendor else None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-vendor-object-injection",
headers={"host": f"{vendor_with_subdomain.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["has_vendor"] is True
assert data["vendor_attributes"]["id"] == vendor_with_subdomain.id
assert data["vendor_attributes"]["name"] == vendor_with_subdomain.name
assert data["vendor_attributes"]["code"] == vendor_with_subdomain.code
assert data["vendor_attributes"]["is_active"] is True
# ========================================================================
# Edge Cases and Error Handling
# ========================================================================
def test_inactive_vendor_not_detected(self, client, inactive_vendor):
"""Test that inactive vendors are not detected."""
from fastapi import Request
from main import app
@app.get("/test-inactive-vendor-detection")
async def test_inactive(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-inactive-vendor-detection",
headers={"host": f"{inactive_vendor.subdomain}.platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is False
def test_platform_domain_without_subdomain_no_vendor(self, client):
"""Test that platform domain without subdomain doesn't detect vendor."""
from fastapi import Request
from main import app
@app.get("/test-platform-domain")
async def test_platform(request: Request):
return {
"vendor_detected": hasattr(request.state, 'vendor') and request.state.vendor is not None
}
with patch('app.core.config.settings') as mock_settings:
mock_settings.platform_domain = "platform.com"
response = client.get(
"/test-platform-domain",
headers={"host": "platform.com"}
)
assert response.status_code == 200
data = response.json()
assert data["vendor_detected"] is False