166 lines
6.1 KiB
Python
166 lines
6.1 KiB
Python
from unittest.mock import Mock, patch
|
|
|
|
from django.test import TestCase
|
|
from rest_framework.exceptions import ValidationError
|
|
from rest_framework_simplejwt.tokens import RefreshToken
|
|
|
|
from apps.users.models import LoginAttempt, User
|
|
from apps.users.services.auth import (
|
|
change_password,
|
|
generate_and_send_otp,
|
|
login_with_otp,
|
|
login_with_password,
|
|
logout_user,
|
|
register_user_with_otp,
|
|
register_user_with_password,
|
|
reset_password_with_otp,
|
|
)
|
|
|
|
|
|
class FakeRedisConnection:
|
|
def __init__(self):
|
|
self.store = {}
|
|
|
|
def get(self, key):
|
|
return self.store.get(key)
|
|
|
|
def setex(self, key, timeout, value):
|
|
self.store[key] = str(value).encode("utf-8")
|
|
|
|
def delete(self, key):
|
|
self.store.pop(key, None)
|
|
|
|
|
|
class AuthServiceTests(TestCase):
|
|
def test_register_user_with_password_creates_user_and_tokens(self):
|
|
tokens = register_user_with_password("09120000001", "secret123")
|
|
|
|
self.assertTrue(User.objects.filter(mobile="09120000001").exists())
|
|
self.assertIn("access", tokens)
|
|
self.assertIn("refresh", tokens)
|
|
|
|
@patch("apps.users.services.auth.send_verification_sms.delay")
|
|
@patch("apps.users.services.auth.get_redis_connection")
|
|
def test_generate_and_send_otp_stores_code_and_schedules_sms(
|
|
self,
|
|
get_redis_connection,
|
|
send_delay,
|
|
):
|
|
User.objects.create_user(mobile="09120000002", password="secret123")
|
|
fake_redis = FakeRedisConnection()
|
|
get_redis_connection.return_value = fake_redis
|
|
|
|
generate_and_send_otp("09120000002", "login")
|
|
|
|
self.assertIn("verification_code:09120000002", fake_redis.store)
|
|
send_delay.assert_called_once()
|
|
|
|
@patch("apps.users.services.auth.record_login_attempt")
|
|
def test_login_with_password_records_success(self, record_login_attempt):
|
|
user = User.objects.create_user(mobile="09120000003", password="secret123")
|
|
|
|
tokens = login_with_password("09120000003", "secret123", request=None)
|
|
|
|
self.assertIn("access", tokens)
|
|
record_login_attempt.assert_called_once_with(
|
|
None,
|
|
user,
|
|
LoginAttempt.StatusType.SUCCESS,
|
|
)
|
|
|
|
@patch("apps.users.services.auth.record_login_attempt")
|
|
def test_login_with_password_rejects_invalid_password(self, record_login_attempt):
|
|
User.objects.create_user(mobile="09120000004", password="secret123")
|
|
|
|
with self.assertRaises(ValidationError):
|
|
login_with_password("09120000004", "wrong-password", request=None)
|
|
|
|
record_login_attempt.assert_called_once()
|
|
|
|
@patch("apps.users.services.auth.record_login_attempt")
|
|
@patch("apps.users.services.auth.get_redis_connection")
|
|
def test_login_with_otp_creates_user_and_consumes_code(
|
|
self,
|
|
get_redis_connection,
|
|
record_login_attempt,
|
|
):
|
|
fake_redis = FakeRedisConnection()
|
|
fake_redis.setex("verification_code:09120000005", 120, "12345")
|
|
get_redis_connection.return_value = fake_redis
|
|
|
|
tokens = login_with_otp("09120000005", "12345", request=None)
|
|
|
|
self.assertTrue(User.objects.filter(mobile="09120000005").exists())
|
|
self.assertIn("access", tokens)
|
|
self.assertNotIn("verification_code:09120000005", fake_redis.store)
|
|
record_login_attempt.assert_called_once()
|
|
|
|
@patch("apps.users.services.auth.get_redis_connection")
|
|
def test_register_user_with_otp_verifies_code_and_marks_user_verified(
|
|
self,
|
|
get_redis_connection,
|
|
):
|
|
fake_redis = FakeRedisConnection()
|
|
fake_redis.setex("verification_code:09120000006", 120, "12345")
|
|
get_redis_connection.return_value = fake_redis
|
|
|
|
tokens = register_user_with_otp(
|
|
mobile="09120000006",
|
|
code="12345",
|
|
password="secret123",
|
|
first_name="OTP",
|
|
last_name="User",
|
|
)
|
|
|
|
user = User.objects.get(mobile="09120000006")
|
|
self.assertTrue(user.is_verified)
|
|
self.assertTrue(user.check_password("secret123"))
|
|
self.assertIn("refresh", tokens)
|
|
|
|
@patch("apps.users.services.auth.get_redis_connection")
|
|
def test_reset_password_with_otp_updates_password(self, get_redis_connection):
|
|
user = User.objects.create_user(mobile="09120000007", password="OldSecret1!")
|
|
fake_redis = FakeRedisConnection()
|
|
fake_redis.setex("verification_code:09120000007", 120, "12345")
|
|
get_redis_connection.return_value = fake_redis
|
|
|
|
reset_password_with_otp("09120000007", "12345", "NewSecret1!")
|
|
|
|
user.refresh_from_db()
|
|
self.assertTrue(user.check_password("NewSecret1!"))
|
|
self.assertNotIn("verification_code:09120000007", fake_redis.store)
|
|
|
|
def test_change_password_updates_existing_user_password(self):
|
|
user = User.objects.create_user(mobile="09120000008", password="OldSecret1!")
|
|
|
|
change_password(user, "OldSecret1!", "NewSecret1!")
|
|
|
|
user.refresh_from_db()
|
|
self.assertTrue(user.check_password("NewSecret1!"))
|
|
self.assertIsNotNone(user.password_updated_at)
|
|
|
|
@patch("apps.users.services.auth.get_redis_connection")
|
|
def test_reset_password_with_otp_rejects_reused_password(self, get_redis_connection):
|
|
User.objects.create_user(mobile="09120000070", password="OldSecret1!")
|
|
fake_redis = FakeRedisConnection()
|
|
fake_redis.setex("verification_code:09120000070", 120, "12345")
|
|
get_redis_connection.return_value = fake_redis
|
|
|
|
with self.assertRaises(ValidationError):
|
|
reset_password_with_otp("09120000070", "12345", "OldSecret1!")
|
|
|
|
def test_change_password_rejects_weak_password(self):
|
|
user = User.objects.create_user(mobile="09120000071", password="OldSecret1!")
|
|
|
|
with self.assertRaises(ValidationError):
|
|
change_password(user, "OldSecret1!", "weakpass")
|
|
|
|
def test_logout_user_blacklists_refresh_token(self):
|
|
user = User.objects.create_user(mobile="09120000009", password="secret123")
|
|
refresh = str(RefreshToken.for_user(user))
|
|
|
|
logout_user(refresh)
|
|
|
|
with self.assertRaises(ValidationError):
|
|
logout_user(refresh)
|