from datetime import timedelta from unittest.mock import patch from django.contrib.auth import get_user_model from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient from accounts.models import PasswordResetOTP, TransporterProfile, UserRole User = get_user_model() class ForgotPasswordAPITests(TestCase): def setUp(self): self.client = APIClient() # 1. Shipper self.shipper = User.objects.create_user( 'shipper-reset@test.com', email='shipper-reset@test.com', password='oldPassword123', first_name='Shipper', last_name='User', ) UserRole.objects.create(user=self.shipper, role=UserRole.Role.SHIPPER, phone='', language='en') # 2. Fleet Owner Transporter self.fleet_owner = User.objects.create_user( 'owner-reset@test.com', email='owner-reset@test.com', password='oldPassword123', first_name='Fleet', last_name='Owner', ) UserRole.objects.create(user=self.fleet_owner, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=self.fleet_owner, account_type=TransporterProfile.AccountType.FLEET_OWNER, company_name='Fast Fleet Ltd', ) # 3. Individual Driver Transporter self.ind_driver = User.objects.create_user( 'driver-reset@test.com', email='driver-reset@test.com', password='oldPassword123', first_name='Ind', last_name='Driver', ) UserRole.objects.create(user=self.ind_driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=self.ind_driver, account_type=TransporterProfile.AccountType.DRIVER, ) # 4. Transporter Driver (Fleet Driver) self.fleet_driver = User.objects.create_user( 'fleetdriver-reset@test.com', email='fleetdriver-reset@test.com', password='oldPassword123', first_name='Fleet', last_name='Driver', ) UserRole.objects.create(user=self.fleet_driver, role=UserRole.Role.TRANSPORTER, phone='', language='en') TransporterProfile.objects.create( user=self.fleet_driver, account_type=TransporterProfile.AccountType.TRANSPORTER_DRIVER, ) @patch('api.views.send_mail') def test_forgot_password_request_success_all_roles(self, mock_send_mail): """Covers shippers, fleet owners, individual drivers, and transporter drivers.""" users = [self.shipper, self.fleet_owner, self.ind_driver, self.fleet_driver] for user in users: response = self.client.post( '/api/auth/forgot-password/', {'email': user.email.upper()}, # Case insensitivity test format='json', ) self.assertEqual(response.status_code, 200) data = response.json() self.assertEqual(data['status_code'], 200) self.assertEqual(data['data']['email'], user.email) # Check DB record otp_record = PasswordResetOTP.objects.filter(user=user, is_used=False).first() self.assertIsNotNone(otp_record) self.assertEqual(len(otp_record.otp), 6) self.assertTrue(otp_record.is_valid) # Ensure emails were attempted self.assertEqual(mock_send_mail.call_count, len(users)) def test_forgot_password_non_existent_email(self): response = self.client.post( '/api/auth/forgot-password/', {'email': 'nonexistent@test.com'}, format='json', ) self.assertEqual(response.status_code, 404) def test_forgot_password_disabled_user(self): self.shipper.is_active = False self.shipper.save() response = self.client.post( '/api/auth/forgot-password/', {'email': self.shipper.email}, format='json', ) self.assertEqual(response.status_code, 400) def test_verify_otp_success_and_failures(self): # Create OTP self.client.post('/api/auth/forgot-password/', {'email': self.fleet_driver.email}, format='json') otp_record = PasswordResetOTP.objects.get(user=self.fleet_driver, is_used=False) # 1. Incorrect OTP res_fail = self.client.post( '/api/auth/verify-otp/', {'email': self.fleet_driver.email, 'otp': '000000'}, format='json', ) self.assertEqual(res_fail.status_code, 400) # 2. Correct OTP res_ok = self.client.post( '/api/auth/verify-otp/', {'email': self.fleet_driver.email, 'otp': otp_record.otp}, format='json', ) self.assertEqual(res_ok.status_code, 200) data = res_ok.json()['data'] self.assertEqual(data['email'], self.fleet_driver.email) self.assertEqual(data['reset_token'], otp_record.token) self.assertTrue(data['verified']) def test_expired_otp_rejected(self): otp_record = PasswordResetOTP.objects.create( user=self.ind_driver, email=self.ind_driver.email, otp='999888', token='tok123', expires_at=timezone.now() - timedelta(minutes=5), ) response = self.client.post( '/api/auth/verify-otp/', {'email': self.ind_driver.email, 'otp': '999888'}, format='json', ) self.assertEqual(response.status_code, 400) def test_reset_password_success_and_login_with_new_password(self): # Request OTP self.client.post('/api/auth/forgot-password/', {'email': self.fleet_owner.email}, format='json') otp_record = PasswordResetOTP.objects.get(user=self.fleet_owner, is_used=False) # Reset password reset_res = self.client.post( '/api/auth/reset-password/', { 'email': self.fleet_owner.email, 'otp': otp_record.otp, 'new_password': 'BrandNewPassword123!', }, format='json', ) self.assertEqual(reset_res.status_code, 200) data = reset_res.json()['data'] self.assertEqual(data['email'], self.fleet_owner.email) self.assertEqual(data['persona_type'], 'TRANSPORTER_FLEET_OWNER') self.assertIsNotNone(data['token']) # OTP is now marked as used otp_record.refresh_from_db() self.assertTrue(otp_record.is_used) # Cannot reuse the same OTP reuse_res = self.client.post( '/api/auth/reset-password/', { 'email': self.fleet_owner.email, 'otp': otp_record.otp, 'new_password': 'AnotherNewPassword123!', }, format='json', ) self.assertEqual(reuse_res.status_code, 400) # Login with old password fails login_old = self.client.post( '/api/auth/login/', {'email': self.fleet_owner.email, 'password': 'oldPassword123'}, format='json', ) self.assertEqual(login_old.status_code, 401) # Login with new password succeeds login_new = self.client.post( '/api/auth/login/', {'email': self.fleet_owner.email, 'password': 'BrandNewPassword123!'}, format='json', ) self.assertEqual(login_new.status_code, 200) self.assertEqual(login_new.json()['data']['email'], self.fleet_owner.email) def test_reset_password_using_reset_token(self): self.client.post('/api/auth/forgot-password/', {'email': self.shipper.email}, format='json') otp_record = PasswordResetOTP.objects.get(user=self.shipper, is_used=False) reset_res = self.client.post( '/api/auth/reset-password/', { 'email': self.shipper.email, 'reset_token': otp_record.token, 'new_password': 'ShipperNewPassword2026', }, format='json', ) self.assertEqual(reset_res.status_code, 200) # Verify login login_res = self.client.post( '/api/auth/login/', {'email': self.shipper.email, 'password': 'ShipperNewPassword2026'}, format='json', ) self.assertEqual(login_res.status_code, 200) def test_password_short_length_rejected(self): self.client.post('/api/auth/forgot-password/', {'email': self.shipper.email}, format='json') otp_record = PasswordResetOTP.objects.get(user=self.shipper, is_used=False) reset_res = self.client.post( '/api/auth/reset-password/', { 'email': self.shipper.email, 'otp': otp_record.otp, 'new_password': 'short', }, format='json', ) self.assertEqual(reset_res.status_code, 400)