Kitos
c5eb6f6dc1
Revoke tokens by jti in a dedicated Redis DB, honor TTL from JWT exp on logout, reject revoked tokens in get_current_user, and add FakeRedis-backed API tests.
125 lines
3.6 KiB
Python
125 lines
3.6 KiB
Python
"""Tests for authentication endpoints."""
|
|
|
|
import pytest
|
|
|
|
|
|
def test_login_success(client, admin_user):
|
|
"""Test successful login returns a token."""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": "admin", "password": "admin123"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "access_token" in data
|
|
assert data["token_type"] == "bearer"
|
|
|
|
|
|
def test_login_wrong_password(client, admin_user):
|
|
"""Test login with wrong password returns 400."""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": "admin", "password": "wrongpassword"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_login_nonexistent_user(client):
|
|
"""Test login with non-existent user returns 400."""
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": "nobody", "password": "password"},
|
|
)
|
|
assert response.status_code == 400
|
|
|
|
|
|
def test_login_inactive_user(client, db):
|
|
"""Test login with inactive user returns 400."""
|
|
from app.auth import hash_password
|
|
from app.models.user import User
|
|
|
|
user = User(
|
|
username="inactive",
|
|
hashed_password=hash_password("password"),
|
|
role="viewer",
|
|
is_active=False,
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": "inactive", "password": "password"},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_get_me_with_token(client, admin_user, admin_token):
|
|
"""Test /auth/me returns current user with valid token."""
|
|
response = client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {admin_token}"},
|
|
)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["username"] == "admin"
|
|
assert data["role"] == "admin"
|
|
|
|
|
|
def test_get_me_without_token(client):
|
|
"""Test /auth/me returns 401 without token."""
|
|
response = client.get("/api/v1/auth/me")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_get_me_invalid_token(client):
|
|
"""Test /auth/me returns 401 with invalid token."""
|
|
response = client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": "Bearer invalidtoken"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_logout_revokes_token(client, admin_user):
|
|
"""After logout, the same JWT must be rejected (Redis blacklist).
|
|
|
|
Prefer Authorization over the HttpOnly cookie so logout blacklists the
|
|
same token the client sends on the next request; clear cookies to avoid
|
|
stale jar state across calls.
|
|
"""
|
|
login = client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": "admin", "password": "admin123"},
|
|
)
|
|
assert login.status_code == 200
|
|
token = login.json()["access_token"]
|
|
|
|
client.cookies.clear()
|
|
out = client.post(
|
|
"/api/v1/auth/logout",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert out.status_code == 200
|
|
|
|
from jose import jwt
|
|
|
|
from app.config import settings
|
|
from app.infrastructure.redis_client import get_redis_blacklist
|
|
|
|
payload = jwt.decode(
|
|
token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM],
|
|
)
|
|
jti = payload.get("jti")
|
|
assert jti
|
|
assert get_redis_blacklist().exists(f"blacklist:{jti}")
|
|
|
|
client.cookies.clear()
|
|
assert not len(client.cookies), "cookie jar must be empty to force Authorization Bearer"
|
|
me = client.get(
|
|
"/api/v1/auth/me",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert me.status_code == 401
|
|
assert me.json()["detail"] == "Token has been revoked"
|