test: added unit tests in auth_app and fixed code issues identified during testing
This commit is contained in:
parent
87f9b3a4b9
commit
acc03c093d
4 changed files with 190 additions and 10 deletions
|
|
@ -2,14 +2,33 @@ from django.utils import timezone
|
||||||
from rest_framework.authtoken.models import Token
|
from rest_framework.authtoken.models import Token
|
||||||
from django.db import models
|
from django.db import models
|
||||||
from join.settings import TOKEN_EXPIRATION_TIME
|
from join.settings import TOKEN_EXPIRATION_TIME
|
||||||
|
from rest_framework.authentication import TokenAuthentication
|
||||||
|
from rest_framework.exceptions import AuthenticationFailed
|
||||||
|
|
||||||
class ExpiringToken(Token):
|
class ExpiringToken(Token):
|
||||||
expires_at = models.DateTimeField(null=True, blank=True)
|
expires_at = models.DateTimeField(null=True, blank=True)
|
||||||
|
|
||||||
def is_expired(self):
|
def is_expired(self):
|
||||||
return self.expires_at and self.expires_at < timezone.now()
|
if self.expires_at is None:
|
||||||
|
return True
|
||||||
|
return self.expires_at < timezone.now()
|
||||||
|
|
||||||
def save(self, *args, **kwargs):
|
def save(self, *args, **kwargs):
|
||||||
if not self.expires_at:
|
if not self.expires_at:
|
||||||
self.expires_at = timezone.now() + TOKEN_EXPIRATION_TIME
|
self.expires_at = timezone.now() + TOKEN_EXPIRATION_TIME
|
||||||
super().save(*args, **kwargs)
|
super().save(*args, **kwargs)
|
||||||
|
|
||||||
|
class ExpiringTokenAuthentication(TokenAuthentication):
|
||||||
|
model = ExpiringToken
|
||||||
|
|
||||||
|
def authenticate_credentials(self, key):
|
||||||
|
token = self.model.objects.filter(key=key).first()
|
||||||
|
|
||||||
|
if not token:
|
||||||
|
raise AuthenticationFailed({"error": "Invalid token."})
|
||||||
|
|
||||||
|
if token.is_expired():
|
||||||
|
token.delete()
|
||||||
|
raise AuthenticationFailed({"error": "No active session or token already expired"})
|
||||||
|
|
||||||
|
return (token.user, token)
|
||||||
|
|
@ -1,8 +1,9 @@
|
||||||
from rest_framework import serializers
|
from rest_framework import serializers
|
||||||
|
from django.core.validators import MinLengthValidator
|
||||||
|
|
||||||
class LoginSerializer(serializers.Serializer):
|
class LoginSerializer(serializers.Serializer):
|
||||||
email = serializers.EmailField()
|
email = serializers.EmailField()
|
||||||
password = serializers.CharField(write_only=True)
|
password = serializers.CharField(validators=[MinLengthValidator(limit_value=8)])
|
||||||
|
|
||||||
def validate(self, data):
|
def validate(self, data):
|
||||||
email = data.get('email')
|
email = data.get('email')
|
||||||
|
|
|
||||||
|
|
@ -1,3 +1,161 @@
|
||||||
from django.test import TestCase
|
import json
|
||||||
|
from django.contrib.auth import get_user_model
|
||||||
|
from django.utils import timezone
|
||||||
|
from rest_framework.test import APITestCase
|
||||||
|
from rest_framework import status
|
||||||
|
from join.settings import TOKEN_EXPIRATION_TIME
|
||||||
|
from .models import ExpiringToken
|
||||||
|
|
||||||
# Create your tests here.
|
User = get_user_model()
|
||||||
|
|
||||||
|
class AuthTests(APITestCase):
|
||||||
|
def setUp(self):
|
||||||
|
self.user = User.objects.create_user(
|
||||||
|
email="test@example.com",
|
||||||
|
password="securepassword",
|
||||||
|
first_name="Test"
|
||||||
|
)
|
||||||
|
self.token = ExpiringToken.objects.create(user=self.user, expires_at=timezone.now() + TOKEN_EXPIRATION_TIME)
|
||||||
|
self.login_url = "/auth/login/"
|
||||||
|
self.logout_url = "/auth/logout/"
|
||||||
|
self.auth_url = "/auth/"
|
||||||
|
|
||||||
|
def test_login_success(self):
|
||||||
|
response = self.client.post(
|
||||||
|
self.login_url,
|
||||||
|
data=json.dumps({"email": "test@example.com", "password": "securepassword"}),
|
||||||
|
content_type="application/json"
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
|
||||||
|
def test_login_invalid_credentials(self):
|
||||||
|
response = self.client.post(
|
||||||
|
self.login_url,
|
||||||
|
data=json.dumps({"email": "test@example.com", "password": "wrongpassword"}),
|
||||||
|
content_type="application/json"
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
|
||||||
|
def test_login_inactive_user(self):
|
||||||
|
self.user.is_active = False
|
||||||
|
self.user.save()
|
||||||
|
|
||||||
|
response = self.client.post(
|
||||||
|
self.login_url,
|
||||||
|
data=json.dumps({"email": "test@example.com", "password": "securepassword"}),
|
||||||
|
content_type="application/json"
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)
|
||||||
|
|
||||||
|
def test_logout_success(self):
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.post(self.logout_url)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
self.assertIn("message", response.data)
|
||||||
|
|
||||||
|
def test_logout_without_token(self):
|
||||||
|
response = self.client.post(self.logout_url)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
|
||||||
|
def test_auth_valid_token(self):
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
self.assertEqual(response.data["user_id"], self.user.id)
|
||||||
|
|
||||||
|
def test_auth_expired_token(self):
|
||||||
|
self.token.expires_at = timezone.now() - TOKEN_EXPIRATION_TIME
|
||||||
|
self.token.save()
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
self.assertIn("error", response.data)
|
||||||
|
|
||||||
|
def test_auth_no_token(self):
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
self.assertIn("error", response.data)
|
||||||
|
|
||||||
|
def test_auth_valid_token_with_fresh_expiration(self):
|
||||||
|
self.token.expires_at = timezone.now() + timezone.timedelta(minutes=10)
|
||||||
|
self.token.save()
|
||||||
|
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
self.assertEqual(response.data["user_id"], self.user.id)
|
||||||
|
|
||||||
|
def test_auth_token_exactly_at_expiration(self):
|
||||||
|
self.token.expires_at = timezone.now()
|
||||||
|
self.token.save()
|
||||||
|
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
self.assertIn("error", response.data)
|
||||||
|
self.assertEqual(response.data["error"], "Token expired, please log in again")
|
||||||
|
|
||||||
|
def test_logout_with_expired_token(self):
|
||||||
|
self.token.expires_at = timezone.now() - timezone.timedelta(days=1)
|
||||||
|
self.token.save()
|
||||||
|
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.post(self.logout_url)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
self.assertIn("error", response.data)
|
||||||
|
self.assertEqual(response.data["error"], "No active session or token already expired")
|
||||||
|
|
||||||
|
def test_login_invalid_email_format(self):
|
||||||
|
response = self.client.post(
|
||||||
|
self.login_url,
|
||||||
|
data=json.dumps({"email": "invalidemail", "password": "securepassword"}),
|
||||||
|
content_type="application/json"
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||||
|
self.assertIn("email", response.data)
|
||||||
|
|
||||||
|
def test_login_password_too_short(self):
|
||||||
|
data = {'email': 'user@example.com', 'password': 'short'}
|
||||||
|
response = self.client.post(self.login_url, data, format='json')
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST)
|
||||||
|
|
||||||
|
def test_logout_token_deletion(self):
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.post(self.logout_url)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
self.assertIn("detail", response.data)
|
||||||
|
self.assertEqual(response.data["detail"], "Invalid token.")
|
||||||
|
|
||||||
|
def test_auth_inactive_user(self):
|
||||||
|
self.user.is_active = False
|
||||||
|
self.user.save()
|
||||||
|
|
||||||
|
self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}')
|
||||||
|
response = self.client.get(self.auth_url)
|
||||||
|
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)
|
||||||
|
self.assertIn("detail", response.data)
|
||||||
|
self.assertEqual(response.data["detail"], "User inactive or deleted.")
|
||||||
|
|
||||||
|
def test_token_for_different_users(self):
|
||||||
|
user2 = User.objects.create_user(
|
||||||
|
email="test2@example.com",
|
||||||
|
password="securepassword2",
|
||||||
|
first_name="Test2"
|
||||||
|
)
|
||||||
|
response = self.client.post(
|
||||||
|
self.login_url,
|
||||||
|
data=json.dumps({"email": "test2@example.com", "password": "securepassword2"}),
|
||||||
|
content_type="application/json"
|
||||||
|
)
|
||||||
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
||||||
|
self.assertNotEqual(response.data["user_id"], self.user.id)
|
||||||
|
|
@ -6,7 +6,7 @@ from join.settings import TOKEN_EXPIRATION_TIME
|
||||||
from .serializer import LoginSerializer
|
from .serializer import LoginSerializer
|
||||||
from rest_framework.permissions import IsAuthenticated
|
from rest_framework.permissions import IsAuthenticated
|
||||||
from django.utils import timezone
|
from django.utils import timezone
|
||||||
from .models import ExpiringToken
|
from .models import ExpiringToken, ExpiringTokenAuthentication
|
||||||
|
|
||||||
|
|
||||||
class LoginView(APIView):
|
class LoginView(APIView):
|
||||||
|
|
@ -42,15 +42,17 @@ class LoginView(APIView):
|
||||||
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
|
||||||
|
|
||||||
class LogoutView(APIView):
|
class LogoutView(APIView):
|
||||||
authentication_classes = [authentication.TokenAuthentication]
|
authentication_classes = [ExpiringTokenAuthentication]
|
||||||
permission_classes = [IsAuthenticated]
|
permission_classes = [IsAuthenticated]
|
||||||
|
|
||||||
def post(self, request):
|
def post(self, request):
|
||||||
token = getattr(request.user, 'auth_token', None)
|
token = getattr(request.user, 'auth_token', None)
|
||||||
|
|
||||||
if token is None:
|
if token is None or not isinstance(token, ExpiringToken):
|
||||||
return Response({"error": "No active session or token already expired"}, status=status.HTTP_400_BAD_REQUEST)
|
return Response({"detail": "Invalid token."}, status=status.HTTP_401_UNAUTHORIZED)
|
||||||
|
|
||||||
|
if token.is_expired():
|
||||||
|
return Response({"error": "No active session or token already expired"}, status=status.HTTP_401_UNAUTHORIZED)
|
||||||
token.delete()
|
token.delete()
|
||||||
return Response({"message": "Successfully logged out"}, status=status.HTTP_200_OK)
|
return Response({"message": "Successfully logged out"}, status=status.HTTP_200_OK)
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue