Files
orion/tests/integration/middleware/test_context_detection_flow.py
Samir Boulahtit 9920430b9e fix: correct tojson|safe usage in templates and update validator
- Remove |safe from |tojson in HTML attributes (x-data) - quotes must
  become " for browsers to parse correctly
- Update LANG-002 and LANG-003 architecture rules to document correct
  |tojson usage patterns:
  - HTML attributes: |tojson (no |safe)
  - Script blocks: |tojson|safe
- Fix validator to warn when |tojson|safe is used in x-data (breaks
  HTML attribute parsing)
- Improve code quality across services, APIs, and tests

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2025-12-13 22:59:51 +01:00

532 lines
19 KiB
Python

# 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.
Note: Test routes use /api/test-* prefix to avoid being caught by the
platform's /{slug} catch-all route for content pages.
"""
from unittest.mock import patch
import pytest
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/test-nested-api-context")
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/test-nested-api-context")
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("/api/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(
"/api/test-admin-subdomain-context",
headers={"host": "admin.platform.com"},
)
assert response.status_code == 200
data = response.json()
# Note: API path overrides subdomain, so still API context
assert data["context_type"] == "api"
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/test-vendors-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/test-vendors-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/test-products-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/test-products-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_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("/shop/test-custom-domain-shop")
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.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(
"/shop/test-custom-domain-shop", 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.vendor_code
# ========================================================================
# Fallback Context Detection Tests
# ========================================================================
def test_unknown_path_without_vendor_fallback_context(self, client):
"""Test that API paths without vendor get API context (fallback via API)."""
from fastapi import Request
from main import app
@app.get("/api/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(
"/api/test-fallback-context", headers={"host": "platform.com"}
)
assert response.status_code == 200
data = response.json()
# API path triggers API context
assert data["context_type"] == "api"
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("/api/test-clean-path-context")
async def test_clean_path_context(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(
"/api/test-clean-path-context",
headers={"host": "localhost:8000"},
)
assert response.status_code == 200
data = response.json()
assert data["context_type"] == "api"
# ========================================================================
# 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_case_insensitive_context_detection(self, client):
"""Test that context detection handles different cases 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"]