280 lines
10 KiB
Python
280 lines
10 KiB
Python
"""
|
|
Unit Tests - Tenant Middleware - ServiceManagerWeb
|
|
|
|
Tests para app.middleware.tenant: extracción de headers, rutas excluidas,
|
|
y comportamiento con tenants válidos/inválidos usando mocks.
|
|
No requieren base de datos real ni red.
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
|
|
# ============================================================
|
|
# EXCLUDED PATHS
|
|
# ============================================================
|
|
|
|
class TestExcludedPaths:
|
|
"""Tests para las rutas que no requieren validación de tenant."""
|
|
|
|
def test_excluded_paths_contains_health(self):
|
|
"""El health check debe estar en rutas excluidas."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
assert "/health" in TenantMiddleware.EXCLUDED_PATHS
|
|
|
|
def test_excluded_paths_contains_login(self):
|
|
"""El endpoint de login debe estar excluido."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
assert "/api/v1/auth/login" in TenantMiddleware.EXCLUDED_PATHS
|
|
assert "/v1/auth/login" in TenantMiddleware.EXCLUDED_PATHS
|
|
|
|
def test_excluded_paths_contains_refresh(self):
|
|
"""El endpoint de refresh token debe estar excluido."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
assert "/api/v1/auth/refresh" in TenantMiddleware.EXCLUDED_PATHS
|
|
assert "/v1/auth/refresh" in TenantMiddleware.EXCLUDED_PATHS
|
|
|
|
def test_excluded_paths_contains_docs(self):
|
|
"""Los endpoints de documentación deben estar excluidos."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
assert "/docs" in TenantMiddleware.EXCLUDED_PATHS
|
|
assert "/redoc" in TenantMiddleware.EXCLUDED_PATHS
|
|
|
|
def test_excluded_paths_contains_openapi(self):
|
|
"""El endpoint openapi.json debe estar excluido."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
assert "/openapi.json" in TenantMiddleware.EXCLUDED_PATHS
|
|
|
|
def test_root_path_is_excluded(self):
|
|
"""La ruta raíz debe estar excluida."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
assert "/" in TenantMiddleware.EXCLUDED_PATHS
|
|
|
|
|
|
# ============================================================
|
|
# MIDDLEWARE DISPATCH — RUTAS EXCLUIDAS
|
|
# ============================================================
|
|
|
|
class TestMiddlewareExcludedRoutes:
|
|
"""Tests que verifican que las rutas excluidas pasan sin validación."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_route_bypasses_tenant_validation(self):
|
|
"""La ruta /health pasa sin validación de tenant."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
# Simular request a /health sin headers de tenant
|
|
request = MagicMock()
|
|
request.url.path = "/health"
|
|
request.headers = {}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# call_next debe haberse llamado (pasó sin bloquear)
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_login_route_bypasses_tenant_validation(self):
|
|
"""La ruta /api/v1/auth/login pasa sin validación de tenant."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/api/v1/auth/login"
|
|
request.headers = {}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
await middleware.dispatch(request, call_next)
|
|
call_next.assert_called_once_with(request)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_docs_prefix_bypasses_tenant_validation(self):
|
|
"""Rutas que empiezan con /docs pasan sin validación."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/docs/swagger-ui"
|
|
request.headers = {}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
await middleware.dispatch(request, call_next)
|
|
call_next.assert_called_once_with(request)
|
|
|
|
|
|
# ============================================================
|
|
# MIDDLEWARE DISPATCH — SIN HEADERS DE TENANT
|
|
# ============================================================
|
|
|
|
class TestMiddlewareNoTenantHeaders:
|
|
"""Tests para requests sin headers de tenant."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_tenant_headers_returns_400(self):
|
|
"""Sin tenant headers debe retornar 400 (requerido para aislamiento multi-tenant)."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/v1/tickets/"
|
|
request.method = "GET"
|
|
request.headers = {}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
response = await middleware.dispatch(request, call_next)
|
|
|
|
assert response.status_code == 400
|
|
call_next.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_tenant_headers_does_not_call_next(self):
|
|
"""Sin tenant headers no debe llegar al handler (call_next)."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/v1/tickets/"
|
|
request.method = "GET"
|
|
request.headers = {}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
response = await middleware.dispatch(request, call_next)
|
|
|
|
assert response.status_code == 400
|
|
call_next.assert_not_called()
|
|
|
|
|
|
# ============================================================
|
|
# MIDDLEWARE DISPATCH — CON TENANT VÁLIDO
|
|
# ============================================================
|
|
|
|
class TestMiddlewareValidTenant:
|
|
"""Tests para requests con tenant válido."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_valid_tenant_id_sets_state(self):
|
|
"""Un tenant_id válido debe almacenarse en request.state."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
from app.models.tenant import TenantStatus
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
# Crear tenant mock
|
|
mock_tenant = MagicMock()
|
|
mock_tenant.id = "12345678-1234-5678-1234-567812345678"
|
|
mock_tenant.slug = "test-company"
|
|
mock_tenant.status = TenantStatus.ACTIVE
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/v1/tickets/"
|
|
request.method = "GET"
|
|
request.headers = {"X-Tenant-ID": str(mock_tenant.id)}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
# Mock de la sesión de BD
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.first.return_value = mock_tenant
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with patch("app.middleware.tenant.AsyncSessionLocal", return_value=mock_session):
|
|
await middleware.dispatch(request, call_next)
|
|
|
|
# El tenant debe haber sido asignado al state
|
|
assert request.state.tenant == mock_tenant
|
|
call_next.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_inactive_tenant_returns_403(self):
|
|
"""Un tenant suspendido debe retornar 403."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
from app.models.tenant import TenantStatus
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
mock_tenant = MagicMock()
|
|
mock_tenant.id = "12345678-1234-5678-1234-567812345678"
|
|
mock_tenant.slug = "suspended-company"
|
|
mock_tenant.status = TenantStatus.SUSPENDED
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/v1/tickets/"
|
|
request.method = "GET"
|
|
request.headers = {"X-Tenant-ID": str(mock_tenant.id)}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.first.return_value = mock_tenant
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with patch("app.middleware.tenant.AsyncSessionLocal", return_value=mock_session):
|
|
response = await middleware.dispatch(request, call_next)
|
|
|
|
assert response.status_code == 403
|
|
call_next.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_nonexistent_tenant_returns_404(self):
|
|
"""Un tenant_id que no existe en BD debe retornar 404."""
|
|
from app.middleware.tenant import TenantMiddleware
|
|
|
|
mock_app = AsyncMock()
|
|
middleware = TenantMiddleware(mock_app)
|
|
|
|
request = MagicMock()
|
|
request.url.path = "/v1/tickets/"
|
|
request.method = "GET"
|
|
request.headers = {"X-Tenant-ID": "00000000-0000-0000-0000-000000000000"}
|
|
request.state = MagicMock()
|
|
|
|
call_next = AsyncMock(return_value=MagicMock(status_code=200))
|
|
|
|
mock_result = MagicMock()
|
|
mock_result.scalars.return_value.first.return_value = None # No encontrado
|
|
|
|
mock_session = AsyncMock()
|
|
mock_session.execute = AsyncMock(return_value=mock_result)
|
|
mock_session.__aenter__ = AsyncMock(return_value=mock_session)
|
|
mock_session.__aexit__ = AsyncMock(return_value=False)
|
|
|
|
with patch("app.middleware.tenant.AsyncSessionLocal", return_value=mock_session):
|
|
response = await middleware.dispatch(request, call_next)
|
|
|
|
assert response.status_code == 404
|
|
call_next.assert_not_called()
|