forked from 0xWheatyz/SPARC
Merge pull request 'Expand JWT auth integration tests to 33 cases' (#1627) from feature/1624-jwt-auth-tests into main
Expand JWT auth integration tests to 33 cases (#1627)
This commit was merged in pull request #1627.
This commit is contained in:
+209
-10
@@ -1,13 +1,29 @@
|
|||||||
"""Tests for JWT authentication flow: register, login, protected routes, refresh, admin access."""
|
"""Tests for JWT authentication flow: register, login, protected routes, refresh, admin access.
|
||||||
|
|
||||||
from datetime import datetime, timezone
|
Covers all five scenarios required by issue #1624:
|
||||||
|
1. Registration (POST /auth/register)
|
||||||
|
2. Login (POST /auth/login)
|
||||||
|
3. Protected route access (GET /auth/me) -- valid, missing, expired, wrong-type tokens
|
||||||
|
4. Token refresh (POST /auth/refresh)
|
||||||
|
5. Admin-only endpoints (GET /admin/users, PATCH role, DELETE user)
|
||||||
|
|
||||||
|
All tests use mocked DB fixtures and require no live database.
|
||||||
|
"""
|
||||||
|
|
||||||
|
from datetime import datetime, timedelta, timezone
|
||||||
from unittest.mock import MagicMock, patch
|
from unittest.mock import MagicMock, patch
|
||||||
|
|
||||||
|
import jwt as pyjwt
|
||||||
import pytest
|
import pytest
|
||||||
from fastapi.testclient import TestClient
|
from fastapi.testclient import TestClient
|
||||||
|
|
||||||
from SPARC.api import app
|
from SPARC.api import app
|
||||||
from SPARC.auth import create_access_token, create_refresh_token
|
from SPARC.auth import (
|
||||||
|
JWT_ALGORITHM,
|
||||||
|
JWT_SECRET,
|
||||||
|
create_access_token,
|
||||||
|
create_refresh_token,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
@@ -171,13 +187,6 @@ class TestGetMe:
|
|||||||
|
|
||||||
def test_expired_token_returns_401(self, client, mock_db):
|
def test_expired_token_returns_401(self, client, mock_db):
|
||||||
"""An expired token should return 401."""
|
"""An expired token should return 401."""
|
||||||
# Create a token that has already expired
|
|
||||||
from datetime import timedelta
|
|
||||||
|
|
||||||
import jwt as pyjwt
|
|
||||||
|
|
||||||
from SPARC.auth import JWT_ALGORITHM, JWT_SECRET
|
|
||||||
|
|
||||||
payload = {
|
payload = {
|
||||||
"sub": "1",
|
"sub": "1",
|
||||||
"email": "user@test.com",
|
"email": "user@test.com",
|
||||||
@@ -301,3 +310,193 @@ class TestAdminUsers:
|
|||||||
|
|
||||||
assert response.status_code == 400
|
assert response.status_code == 400
|
||||||
assert "own role" in response.json()["detail"].lower()
|
assert "own role" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
def test_role_change_nonexistent_user_returns_404(self, client, mock_db):
|
||||||
|
"""Changing role for a user that does not exist should return 404."""
|
||||||
|
admin = _make_admin_user()
|
||||||
|
mock_db.get_user_by_id.return_value = admin
|
||||||
|
mock_db.update_user_role.return_value = None
|
||||||
|
|
||||||
|
response = client.patch(
|
||||||
|
"/admin/users/999/role",
|
||||||
|
json={"role": "admin"},
|
||||||
|
headers=_auth_header(admin),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 404
|
||||||
|
assert "not found" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
def test_regular_user_cannot_change_role(self, client, mock_db):
|
||||||
|
"""Non-admin user should receive 403 when trying to change roles."""
|
||||||
|
user = _make_regular_user()
|
||||||
|
mock_db.get_user_by_id.return_value = user
|
||||||
|
|
||||||
|
response = client.patch(
|
||||||
|
"/admin/users/1/role",
|
||||||
|
json={"role": "admin"},
|
||||||
|
headers=_auth_header(user),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
|
class TestAdminDeleteUser:
|
||||||
|
"""DELETE /admin/users/{user_id}"""
|
||||||
|
|
||||||
|
def test_admin_can_delete_user(self, client, mock_db):
|
||||||
|
"""Admin should be able to delete another user."""
|
||||||
|
admin = _make_admin_user()
|
||||||
|
mock_db.get_user_by_id.return_value = admin
|
||||||
|
mock_db.delete_user.return_value = True
|
||||||
|
|
||||||
|
response = client.delete(
|
||||||
|
"/admin/users/2",
|
||||||
|
headers=_auth_header(admin),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert "deleted" in response.json()["message"].lower()
|
||||||
|
mock_db.delete_user.assert_called_once_with(2)
|
||||||
|
|
||||||
|
def test_admin_cannot_delete_self(self, client, mock_db):
|
||||||
|
"""Admin should not be able to delete themselves."""
|
||||||
|
admin = _make_admin_user()
|
||||||
|
mock_db.get_user_by_id.return_value = admin
|
||||||
|
|
||||||
|
response = client.delete(
|
||||||
|
"/admin/users/1",
|
||||||
|
headers=_auth_header(admin),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 400
|
||||||
|
assert "yourself" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
def test_delete_nonexistent_user_returns_404(self, client, mock_db):
|
||||||
|
"""Deleting a user that does not exist should return 404."""
|
||||||
|
admin = _make_admin_user()
|
||||||
|
mock_db.get_user_by_id.return_value = admin
|
||||||
|
mock_db.delete_user.return_value = False
|
||||||
|
|
||||||
|
response = client.delete(
|
||||||
|
"/admin/users/999",
|
||||||
|
headers=_auth_header(admin),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 404
|
||||||
|
assert "not found" in response.json()["detail"].lower()
|
||||||
|
|
||||||
|
def test_regular_user_cannot_delete_user(self, client, mock_db):
|
||||||
|
"""Non-admin user should receive 403 when trying to delete users."""
|
||||||
|
user = _make_regular_user()
|
||||||
|
mock_db.get_user_by_id.return_value = user
|
||||||
|
|
||||||
|
response = client.delete(
|
||||||
|
"/admin/users/1",
|
||||||
|
headers=_auth_header(user),
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 403
|
||||||
|
|
||||||
|
def test_no_token_cannot_delete_user(self, client):
|
||||||
|
"""Missing token should be rejected for delete endpoint."""
|
||||||
|
response = client.delete("/admin/users/1")
|
||||||
|
assert response.status_code in (401, 403)
|
||||||
|
|
||||||
|
|
||||||
|
class TestEdgeCases:
|
||||||
|
"""Additional edge-case tests for auth robustness."""
|
||||||
|
|
||||||
|
def test_register_invalid_email_returns_422(self, client, mock_db):
|
||||||
|
"""Registration with an invalid email format should return 422."""
|
||||||
|
response = client.post(
|
||||||
|
"/auth/register",
|
||||||
|
json={"email": "not-an-email", "password": "securepass123"},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 422
|
||||||
|
|
||||||
|
def test_register_short_password_returns_422(self, client, mock_db):
|
||||||
|
"""Registration with a password shorter than 8 chars should return 422."""
|
||||||
|
response = client.post(
|
||||||
|
"/auth/register",
|
||||||
|
json={"email": "user@test.com", "password": "short"},
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 422
|
||||||
|
|
||||||
|
def test_register_missing_fields_returns_422(self, client, mock_db):
|
||||||
|
"""Registration with missing fields should return 422."""
|
||||||
|
response = client.post("/auth/register", json={})
|
||||||
|
assert response.status_code == 422
|
||||||
|
|
||||||
|
def test_login_missing_fields_returns_422(self, client, mock_db):
|
||||||
|
"""Login with missing fields should return 422."""
|
||||||
|
response = client.post("/auth/login", json={"email": "user@test.com"})
|
||||||
|
assert response.status_code == 422
|
||||||
|
|
||||||
|
def test_malformed_token_returns_401(self, client, mock_db):
|
||||||
|
"""A completely malformed token string should return 401."""
|
||||||
|
response = client.get(
|
||||||
|
"/auth/me",
|
||||||
|
headers={"Authorization": "Bearer not.a.valid.jwt.token"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
def test_token_with_wrong_secret_returns_401(self, client, mock_db):
|
||||||
|
"""A token signed with a different secret should return 401."""
|
||||||
|
payload = {
|
||||||
|
"sub": "1",
|
||||||
|
"email": "user@test.com",
|
||||||
|
"role": "user",
|
||||||
|
"exp": datetime.now(timezone.utc) + timedelta(hours=1),
|
||||||
|
"type": "access",
|
||||||
|
}
|
||||||
|
wrong_secret_token = pyjwt.encode(payload, "wrong-secret", algorithm=JWT_ALGORITHM)
|
||||||
|
|
||||||
|
response = client.get(
|
||||||
|
"/auth/me",
|
||||||
|
headers={"Authorization": f"Bearer {wrong_secret_token}"},
|
||||||
|
)
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
def test_token_for_deleted_user_returns_401(self, client, mock_db):
|
||||||
|
"""A valid token for a user no longer in the DB should return 401."""
|
||||||
|
user = _make_regular_user()
|
||||||
|
mock_db.get_user_by_id.return_value = None # user was deleted
|
||||||
|
|
||||||
|
response = client.get("/auth/me", headers=_auth_header(user))
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
def test_refresh_for_deleted_user_returns_401(self, client, mock_db):
|
||||||
|
"""Refreshing a token for a deleted user should return 401."""
|
||||||
|
user = _make_regular_user()
|
||||||
|
mock_db.get_user_by_id.return_value = None
|
||||||
|
refresh = create_refresh_token(user["id"], user["email"], user["role"])
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/auth/refresh", json={"refresh_token": refresh}
|
||||||
|
)
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
def test_login_returns_decodable_tokens(self, client, mock_db):
|
||||||
|
"""Tokens returned by login should be decodable and contain expected claims."""
|
||||||
|
user = _make_regular_user()
|
||||||
|
mock_db.authenticate_user.return_value = user
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/auth/login",
|
||||||
|
json={"email": "user@test.com", "password": "correctpassword"},
|
||||||
|
)
|
||||||
|
|
||||||
|
data = response.json()
|
||||||
|
access_payload = pyjwt.decode(
|
||||||
|
data["access_token"], JWT_SECRET, algorithms=[JWT_ALGORITHM]
|
||||||
|
)
|
||||||
|
assert access_payload["sub"] == str(user["id"])
|
||||||
|
assert access_payload["email"] == user["email"]
|
||||||
|
assert access_payload["type"] == "access"
|
||||||
|
|
||||||
|
refresh_payload = pyjwt.decode(
|
||||||
|
data["refresh_token"], JWT_SECRET, algorithms=[JWT_ALGORITHM]
|
||||||
|
)
|
||||||
|
assert refresh_payload["type"] == "refresh"
|
||||||
|
|||||||
Reference in New Issue
Block a user