Samir Boulahtit
238c1ec9b8
- Replace black, isort, and flake8 with Ruff (all-in-one linter and formatter) - Add comprehensive pyproject.toml configuration - Simplify Makefile code quality targets - Configure exclusions for venv/.venv in pyproject.toml - Auto-fix 1,359 linting issues across codebase Benefits: - Much faster builds (Ruff is written in Rust) - Single tool replaces multiple tools - More comprehensive rule set (UP, B, C4, SIM, PIE, RET, Q) - All configuration centralized in pyproject.toml - Better import sorting and formatting consistency 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
593 lines
21 KiB
Python
593 lines
21 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.
|
|
"""
|
|
|
|
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/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"]
|