422 lines
12 KiB
Python
422 lines
12 KiB
Python
"""
|
|
Authentication Integration Tests - ServiceManagerWeb
|
|
|
|
Tests completos del flujo de autenticación incluyendo:
|
|
- Login
|
|
- Refresh tokens
|
|
- Logout
|
|
- Permisos y roles
|
|
"""
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.user import User, UserRole
|
|
from app.models.tenant import Tenant
|
|
|
|
# Importar fixtures desde conftest_integration
|
|
@pytest.mark.integration
|
|
@pytest.mark.auth
|
|
class TestAuthentication:
|
|
"""Tests de autenticación básica."""
|
|
|
|
async def test_login_success(
|
|
self,
|
|
client: AsyncClient,
|
|
test_admin_user: User,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test login exitoso con credenciales válidas."""
|
|
response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "AdminPass123!",
|
|
"tenant_slug": test_tenant.slug
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
assert "access_token" in data
|
|
assert "refresh_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["expires_in"] > 0
|
|
assert data["user"]["email"] == "admin@test.com"
|
|
assert data["user"]["role"] == "ADMIN"
|
|
|
|
async def test_login_invalid_password(
|
|
self,
|
|
client: AsyncClient,
|
|
test_admin_user: User,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test login con contraseña incorrecta."""
|
|
response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "WrongPassword123!",
|
|
"tenant_slug": test_tenant.slug
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
assert "Invalid credentials" in response.json()["detail"]
|
|
|
|
async def test_login_invalid_tenant_slug(
|
|
self,
|
|
client: AsyncClient,
|
|
test_admin_user: User
|
|
):
|
|
"""Test login con tenant slug inexistente."""
|
|
response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "AdminPass123!",
|
|
"tenant_slug": "nonexistent-tenant"
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
async def test_login_user_not_found(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test login con email inexistente."""
|
|
response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "notfound@test.com",
|
|
"password": "SomePassword123!",
|
|
"tenant_slug": test_tenant.slug
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
async def test_login_inactive_user(
|
|
self,
|
|
client: AsyncClient,
|
|
db_session: AsyncSession,
|
|
test_admin_user: User,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test login con usuario desactivado."""
|
|
# Desactivar usuario
|
|
test_admin_user.is_active = False
|
|
await db_session.commit()
|
|
|
|
response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "AdminPass123!",
|
|
"tenant_slug": test_tenant.slug
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
async def test_login_rate_limited_after_too_many_attempts(
|
|
self,
|
|
client: AsyncClient,
|
|
test_admin_user: User,
|
|
test_tenant: Tenant,
|
|
monkeypatch,
|
|
):
|
|
"""Debe devolver 429 después de demasiados intentos de login (rate limit)."""
|
|
|
|
import app.api.v1.endpoints.auth as auth_endpoint
|
|
|
|
class _FakeCache:
|
|
def __init__(self):
|
|
self._counts = {}
|
|
self._expires = {}
|
|
|
|
async def incr(self, key: str, amount: int = 1):
|
|
self._counts[key] = self._counts.get(key, 0) + amount
|
|
return self._counts[key]
|
|
|
|
async def expire(self, key: str, ttl: int):
|
|
self._expires[key] = ttl
|
|
return True
|
|
|
|
async def delete(self, key: str):
|
|
self._counts.pop(key, None)
|
|
return True
|
|
|
|
fake_cache = _FakeCache()
|
|
monkeypatch.setattr(auth_endpoint, "cache", fake_cache)
|
|
monkeypatch.setattr(auth_endpoint.settings, "RATE_LIMIT_ENABLED", True, raising=False)
|
|
monkeypatch.setattr(auth_endpoint.settings, "TESTING", False, raising=False)
|
|
monkeypatch.setattr(auth_endpoint.settings, "LOGIN_RATE_LIMIT_WINDOW_SECONDS", 60, raising=False)
|
|
monkeypatch.setattr(auth_endpoint.settings, "LOGIN_RATE_LIMIT_IP_MAX_ATTEMPTS", 10_000, raising=False)
|
|
monkeypatch.setattr(auth_endpoint.settings, "LOGIN_RATE_LIMIT_ID_MAX_ATTEMPTS", 2, raising=False)
|
|
|
|
payload = {
|
|
"email": test_admin_user.email,
|
|
"password": "WrongPassword123!",
|
|
"tenant_slug": test_tenant.slug,
|
|
}
|
|
|
|
r1 = await client.post("/v1/auth/login", json=payload)
|
|
assert r1.status_code == 401
|
|
|
|
r2 = await client.post("/v1/auth/login", json=payload)
|
|
assert r2.status_code == 401
|
|
|
|
r3 = await client.post("/v1/auth/login", json=payload)
|
|
assert r3.status_code == 429
|
|
assert "Retry-After" in r3.headers
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.auth
|
|
class TestRefreshToken:
|
|
"""Tests de refresh tokens."""
|
|
|
|
async def test_refresh_token_success(
|
|
self,
|
|
client: AsyncClient,
|
|
test_admin_user: User,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test refresh token exitoso."""
|
|
# Login para obtener tokens
|
|
login_response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "AdminPass123!",
|
|
"tenant_slug": test_tenant.slug
|
|
}
|
|
)
|
|
|
|
assert login_response.status_code == 200
|
|
refresh_token = login_response.json()["refresh_token"]
|
|
|
|
# Usar refresh token
|
|
refresh_response = await client.post(
|
|
"/v1/auth/refresh",
|
|
json={"refresh_token": refresh_token}
|
|
)
|
|
|
|
assert refresh_response.status_code == 200
|
|
data = refresh_response.json()
|
|
|
|
assert "access_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
assert data["expires_in"] > 0
|
|
|
|
async def test_refresh_token_invalid(self, client: AsyncClient):
|
|
"""Test refresh con token inválido."""
|
|
response = await client.post(
|
|
"/v1/auth/refresh",
|
|
json={"refresh_token": "invalid-token"}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
async def test_refresh_token_after_logout(
|
|
self,
|
|
client: AsyncClient,
|
|
test_admin_user: User,
|
|
test_tenant: Tenant,
|
|
admin_token: str
|
|
):
|
|
"""Test que refresh token no funciona después de logout."""
|
|
# Login
|
|
login_response = await client.post(
|
|
"/v1/auth/login",
|
|
json={
|
|
"email": "admin@test.com",
|
|
"password": "AdminPass123!",
|
|
"tenant_slug": test_tenant.slug
|
|
}
|
|
)
|
|
|
|
refresh_token = login_response.json()["refresh_token"]
|
|
|
|
# Logout
|
|
logout_response = await client.post(
|
|
"/v1/auth/logout",
|
|
headers={"Authorization": f"Bearer {admin_token}"}
|
|
)
|
|
|
|
assert logout_response.status_code == 200
|
|
|
|
# Intentar usar refresh token después de logout
|
|
refresh_response = await client.post(
|
|
"/v1/auth/refresh",
|
|
json={"refresh_token": refresh_token}
|
|
)
|
|
|
|
assert refresh_response.status_code == 401
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.auth
|
|
class TestAuthorization:
|
|
"""Tests de autorización y permisos."""
|
|
|
|
async def test_admin_can_access_admin_endpoint(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test que admin puede acceder a endpoints de admin."""
|
|
response = await client.get(
|
|
"/v1/tenants/",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
|
|
async def test_agent_cannot_access_admin_endpoint(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
auth_headers_agent: dict
|
|
):
|
|
"""Test que agent no puede acceder a endpoints de admin."""
|
|
response = await client.get(
|
|
"/v1/tenants/",
|
|
headers={
|
|
**auth_headers_agent,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
async def test_client_cannot_access_admin_endpoint(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
auth_headers_client: dict
|
|
):
|
|
"""Test que client no puede acceder a endpoints de admin."""
|
|
response = await client.get(
|
|
"/v1/tenants/",
|
|
headers={
|
|
**auth_headers_client,
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
async def test_protected_endpoint_without_token(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test que endpoints protegidos requieren token."""
|
|
response = await client.get(
|
|
"/v1/tickets/",
|
|
headers={"X-Tenant-ID": str(test_tenant.id)}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
async def test_protected_endpoint_with_invalid_token(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant
|
|
):
|
|
"""Test con token inválido."""
|
|
response = await client.get(
|
|
"/v1/tickets/",
|
|
headers={
|
|
"Authorization": "Bearer invalid-token",
|
|
"X-Tenant-ID": str(test_tenant.id)
|
|
}
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.auth
|
|
class TestUserProfile:
|
|
"""Tests del perfil de usuario."""
|
|
|
|
async def test_get_current_user_profile(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test obtener perfil del usuario actual."""
|
|
response = await client.get(
|
|
"/v1/users/me",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
assert data["email"] == "admin@test.com"
|
|
assert data["role"] == "ADMIN"
|
|
assert data["first_name"] == "Admin"
|
|
assert data["last_name"] == "User"
|
|
assert "password_hash" not in data # No debe exponer password
|
|
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.auth
|
|
class TestPasswordSecurity:
|
|
"""Tests de seguridad de contraseñas."""
|
|
|
|
async def test_password_hashing(self):
|
|
"""Test que las contraseñas se hashean correctamente."""
|
|
from app.core.security import SecurityUtils
|
|
|
|
password = "TestPassword123!"
|
|
hashed = SecurityUtils.hash_password(password)
|
|
|
|
# Debe ser diferente del original
|
|
assert hashed != password
|
|
|
|
# Debe poder verificarse
|
|
assert SecurityUtils.verify_password(password, hashed)
|
|
|
|
# Contraseña incorrecta no debe verificar
|
|
assert not SecurityUtils.verify_password("WrongPassword", hashed)
|
|
|
|
async def test_password_not_exposed_in_response(
|
|
self,
|
|
client: AsyncClient,
|
|
test_tenant: Tenant,
|
|
test_admin_user: User,
|
|
auth_headers_admin: dict
|
|
):
|
|
"""Test que el password hash nunca se expone en las respuestas."""
|
|
response = await client.get(
|
|
"/v1/users/me",
|
|
headers={
|
|
**auth_headers_admin,
|
|
"X-Tenant-ID": str(test_tenant.id),
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
assert "password" not in data
|
|
assert "password_hash" not in data
|