from datetime import datetime, time, timedelta from unittest.mock import patch from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile from django.core.management import call_command from django.test import TestCase from django.utils import timezone from accounts.document_expiry_reminders import ( run_all_document_expiry_reminders, send_document_expiry_reminders_for_days_before, send_expiry_hourly_reminders, ) from accounts.models import ( DocumentExpiryReminderLog, KYCDocument, TransporterProfile, UserRole, Vehicle, VehicleDocument, ) User = get_user_model() class DocumentExpiryReminderTests(TestCase): def setUp(self): self.today = timezone.localdate() def _create_transporter(self, username, account_type): user = User.objects.create_user(username, email=f'{username}@test.com', password='secret') UserRole.objects.create(user=user, role=UserRole.Role.TRANSPORTER, phone='03000000001', language='en') TransporterProfile.objects.create( user=user, account_type=account_type, local=True, country_to_country=False, ) return user @patch('accounts.document_expiry_reminders.notify_users') def test_sends_kyc_reminder_exactly_days_before_expiry(self, mock_notify): driver = self._create_transporter('driver@test.com', TransporterProfile.AccountType.DRIVER) doc = KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.DRIVER_LICENSE, expiry_date=self.today + timedelta(days=7), ) sent = send_document_expiry_reminders_for_days_before(7) self.assertEqual(sent, 1) mock_notify.assert_called_once() user_ids, event, payload, *_ = mock_notify.call_args[0] self.assertEqual(user_ids, [driver.id]) self.assertEqual(event, 'DOCUMENT_EXPIRING') self.assertEqual(payload['days_before'], '7') self.assertEqual(payload['document_id'], str(doc.pk)) self.assertTrue( DocumentExpiryReminderLog.objects.filter( kind=DocumentExpiryReminderLog.DocKind.KYC, document_id=doc.pk, expiry_date=doc.expiry_date, days_before=7, reminder_hour=DocumentExpiryReminderLog.REMINDER_HOUR_DAY_WINDOW, reminder_on=DocumentExpiryReminderLog.REMINDER_ON_DAY_WINDOW, ).exists() ) @patch('accounts.document_expiry_reminders.notify_users') def test_idempotent_per_reminder_window(self, mock_notify): driver = self._create_transporter('driver2@test.com', TransporterProfile.AccountType.DRIVER) KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.PERMIT, expiry_date=self.today + timedelta(days=31), ) first = send_document_expiry_reminders_for_days_before(31) second = send_document_expiry_reminders_for_days_before(31) self.assertEqual(first, 1) self.assertEqual(second, 0) mock_notify.assert_called_once() @patch('accounts.document_expiry_reminders.notify_users') def test_sends_separate_reminders_for_31_7_and_1_day_windows(self, mock_notify): driver = self._create_transporter('driver3@test.com', TransporterProfile.AccountType.DRIVER) expiry = self.today + timedelta(days=31) doc = KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.PASSPORT_COPY, expiry_date=expiry, ) sent_31 = send_document_expiry_reminders_for_days_before(31) sent_7 = send_document_expiry_reminders_for_days_before(7) sent_1 = send_document_expiry_reminders_for_days_before(1) self.assertEqual(sent_31, 1) self.assertEqual(sent_7, 0) self.assertEqual(sent_1, 0) self.assertEqual(mock_notify.call_count, 1) self.assertEqual( DocumentExpiryReminderLog.objects.filter( kind=DocumentExpiryReminderLog.DocKind.KYC, document_id=doc.pk, expiry_date=expiry, ).count(), 1, ) @patch('accounts.document_expiry_reminders.notify_users') def test_skips_fleet_driver_accounts(self, mock_notify): fleet_driver = self._create_transporter( 'fleetdriver@test.com', TransporterProfile.AccountType.TRANSPORTER_DRIVER, ) KYCDocument.objects.create( user=fleet_driver, document_type=KYCDocument.DocumentType.DRIVER_LICENSE, expiry_date=self.today + timedelta(days=7), ) sent = send_document_expiry_reminders_for_days_before(7) self.assertEqual(sent, 0) mock_notify.assert_not_called() @patch('accounts.document_expiry_reminders.notify_users') def test_notifies_fleet_owner_for_vehicle_document(self, mock_notify): owner = self._create_transporter('owner@test.com', TransporterProfile.AccountType.FLEET_OWNER) vehicle = Vehicle.objects.create( owner=owner, vehicle_type='Flatbed', registration_number='ABC-123', load_capacity='10.00', ) doc = VehicleDocument.objects.create( vehicle=vehicle, document_type=VehicleDocument.DocumentType.INSURANCE, file=SimpleUploadedFile('insurance.pdf', b'pdf'), expiry_date=self.today + timedelta(days=1), ) sent = send_document_expiry_reminders_for_days_before(1) self.assertEqual(sent, 1) mock_notify.assert_called_once() user_ids, event, payload = mock_notify.call_args[0][:3] self.assertEqual(user_ids, [owner.id]) self.assertEqual(event, 'DOCUMENT_EXPIRING') self.assertEqual(payload['kind'], 'VEHICLE') self.assertEqual(payload['vehicle_id'], str(vehicle.pk)) self.assertEqual(payload['days_before'], '1') self.assertEqual(payload['document_id'], str(doc.pk)) @patch('accounts.document_expiry_reminders.notify_users') def test_management_command_runs_all_windows(self, mock_notify): driver = self._create_transporter('cmd@test.com', TransporterProfile.AccountType.DRIVER) KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.COUNTRY_GCC, expiry_date=self.today + timedelta(days=7), ) call_command('send_document_expiry_reminders') mock_notify.assert_called_once() results = run_all_document_expiry_reminders() self.assertEqual(results['days_before'][7], 0) @patch('accounts.document_expiry_reminders.notify_users') @patch('accounts.document_expiry_reminders.timezone.localtime') def test_expiry_day_hourly_reminder_once_per_hour(self, mock_localtime, mock_notify): mock_localtime.return_value = timezone.make_aware(datetime.combine(self.today, time(10, 15))) driver = self._create_transporter('hourly@test.com', TransporterProfile.AccountType.DRIVER) KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.DRIVER_LICENSE, expiry_date=self.today, ) first = send_expiry_hourly_reminders() second = send_expiry_hourly_reminders() self.assertEqual(first, 1) self.assertEqual(second, 0) self.assertEqual(mock_notify.call_count, 1) payload = mock_notify.call_args[0][2] self.assertEqual(payload['days_before'], '0') self.assertEqual(payload['reminder_hour'], '10') self.assertEqual(payload['reminder_on'], self.today.isoformat()) mock_localtime.return_value = timezone.make_aware(datetime.combine(self.today, time(11, 5))) third = send_expiry_hourly_reminders() self.assertEqual(third, 1) self.assertEqual(mock_notify.call_count, 2) @patch('accounts.document_expiry_reminders.notify_users') @patch('accounts.document_expiry_reminders.timezone.localdate') @patch('accounts.document_expiry_reminders.timezone.localtime') def test_expired_document_hourly_reminder_continues_on_later_days( self, mock_localtime, mock_localdate, mock_notify, ): expired_on = self.today - timedelta(days=3) mock_localdate.return_value = self.today mock_localtime.return_value = timezone.make_aware(datetime.combine(self.today, time(9, 0))) driver = self._create_transporter('expired@test.com', TransporterProfile.AccountType.DRIVER) KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.DRIVER_LICENSE, expiry_date=expired_on, ) sent_today = send_expiry_hourly_reminders() self.assertEqual(sent_today, 1) payload = mock_notify.call_args[0][2] self.assertEqual(payload['expired'], 'true') self.assertEqual(payload['reminder_on'], self.today.isoformat()) yesterday = self.today - timedelta(days=1) mock_localdate.return_value = yesterday mock_localtime.return_value = timezone.make_aware(datetime.combine(yesterday, time(9, 0))) sent_yesterday = send_expiry_hourly_reminders() self.assertEqual(sent_yesterday, 1) self.assertEqual(mock_notify.call_count, 2) @patch('accounts.document_expiry_reminders.notify_users') @patch('accounts.document_expiry_reminders.timezone.localdate') @patch('accounts.document_expiry_reminders.timezone.localtime') def test_expired_document_hourly_idempotent_within_same_hour( self, mock_localtime, mock_localdate, mock_notify, ): expired_on = self.today - timedelta(days=5) mock_localdate.return_value = self.today mock_localtime.return_value = timezone.make_aware(datetime.combine(self.today, time(14, 5))) driver = self._create_transporter('expired2@test.com', TransporterProfile.AccountType.DRIVER) KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.DRIVER_LICENSE, expiry_date=expired_on, ) first = send_expiry_hourly_reminders() second = send_expiry_hourly_reminders() third = send_expiry_hourly_reminders() self.assertEqual(first, 1) self.assertEqual(second, 0) self.assertEqual(third, 0) self.assertEqual(mock_notify.call_count, 1) @patch('accounts.document_expiry_reminders.notify_users') def test_updated_expiry_date_starts_fresh_reminder_cycle(self, mock_notify): driver = self._create_transporter('renew@test.com', TransporterProfile.AccountType.DRIVER) doc = KYCDocument.objects.create( user=driver, document_type=KYCDocument.DocumentType.PERMIT, expiry_date=self.today, ) send_expiry_hourly_reminders() self.assertEqual(mock_notify.call_count, 1) doc.expiry_date = self.today + timedelta(days=31) doc.save(update_fields=['expiry_date']) send_expiry_hourly_reminders() sent_31 = send_document_expiry_reminders_for_days_before(31) self.assertEqual(sent_31, 1) self.assertEqual(mock_notify.call_count, 2)