import uuid from django.test import TestCase, override_settings from django.core import mail from rest_framework.exceptions import ValidationError as DRFValidationError from django.contrib.auth import get_user_model from unittest.mock import patch from modules.fleet.models import Vehicle, WorkOrder, VehicleServiceRecord, VehicleNotification from modules.fleet.services import VehicleService, NotificationService class NotificationEmailTests(TestCase): @override_settings( EMAIL_BACKEND='django.core.mail.backends.locmem.EmailBackend', DEFAULT_FROM_EMAIL='no-reply@example.test' ) def test_create_notification_sends_email_and_records_is_sent(self): User = get_user_model() user = User.objects.create_user(username=f"user_{uuid.uuid4().hex[:6]}", email=f"user+{uuid.uuid4()}@example.test", password="pass") # Ensure outbox is empty mail.outbox.clear() title = "Test email notification" message = "This is a test message." with patch('django.db.transaction.on_commit', lambda f: f()): notif = NotificationService.create_notification( recipient=user, title=title, message=message, level="info", send_email=True ) # One email should have been sent self.assertEqual(len(mail.outbox), 1) sent = mail.outbox[0] self.assertIn(title, sent.subject) self.assertIn(message, sent.body) # The notification record should exist and flagged as sent notif.refresh_from_db() self.assertTrue(notif.is_sent) self.assertEqual(notif.recipient, user) self.assertEqual(notif.title, title) class NotificationRealtimeTests(TestCase): @patch('modules.fleet.services.PusherService.broadcast_notification') def test_create_notification_broadcasts_after_on_commit(self, mock_broadcast): User = get_user_model() user = User.objects.create_user( username=f"ws_user_{uuid.uuid4().hex[:6]}", email=f"ws+{uuid.uuid4()}@example.test", password="pass" ) with patch('django.db.transaction.on_commit', lambda fn: fn()): notif = NotificationService.create_notification( recipient=user, title="Realtime test", message="Provjera pusher poruke", level="warning", send_email=False, metadata={"source": "unit-test"} ) self.assertEqual(notif.recipient, user) mock_broadcast.assert_called_once_with( user_id=user.id, title="Realtime test", message="Provjera pusher poruke", level="warning", notification_id=notif.id, metadata={"source": "unit-test"} ) @patch('modules.fleet.services.PusherService.broadcast_notification') def test_create_notification_without_recipient_does_not_broadcast(self, mock_broadcast): NotificationService.create_notification( recipient=None, title="No recipient", message="Broadcast se ne smije pozvati", level="info", send_email=False ) mock_broadcast.assert_not_called() class WorkOrderAndServiceRecordNotificationTests(TestCase): def setUp(self): User = get_user_model() # creator who will create work order self.creator = User.objects.create_user(username=f"creator_{uuid.uuid4().hex[:6]}", email=f"creator+{uuid.uuid4()}@example.test", password="pass") # assigned servicer (different user) self.servicer = User.objects.create_user(username=f"servicer_{uuid.uuid4().hex[:6]}", email=f"servicer+{uuid.uuid4()}@example.test", password="pass") # another technician who will perform the service self.technician = User.objects.create_user(username=f"tech_{uuid.uuid4().hex[:6]}", email=f"tech+{uuid.uuid4()}@example.test", password="pass") self.vehicle = Vehicle.objects.create( registration_number=f"REG-{uuid.uuid4().hex[:6].upper()}", make="TestMake", model="TestModel", current_mileage=10000, service_interval_km=15000, assigned_servicer=self.servicer ) def test_create_work_order_creates_notifications_for_creator_and_servicer(self): # Precondition: no notifications VehicleNotification.objects.all().delete() data = { "vehicle": self.vehicle, "creator": self.creator, "start_mileage": 10000, "end_mileage": 10120, "purpose": "Routine visit" } wo = VehicleService.create_work_order(data) # Ensure work order created self.assertIsNotNone(wo.pk) # Notifications: one for creator, one for assigned servicer creator_notifs = VehicleNotification.objects.filter(recipient=self.creator) servicer_notifs = VehicleNotification.objects.filter(recipient=self.servicer) self.assertTrue(creator_notifs.exists(), "Creator should receive a notification") self.assertTrue(servicer_notifs.exists(), "Assigned servicer should receive a notification") def test_create_service_record_creates_notifications_for_performer_and_assigned_servicer(self): # Precondition: no notifications VehicleNotification.objects.all().delete() data = { "vehicle": self.vehicle, "performed_by": self.technician, "description": "Brake replacement", "parts": "pads;discs", "cost": "200.00", "mileage": 10200 } rec = VehicleService.create_service_record(data) # Ensure service record created self.assertIsNotNone(rec.pk) # There should be a notification for the performer perf_notifs = VehicleNotification.objects.filter(recipient=self.technician) self.assertTrue(perf_notifs.exists(), "Performer should receive a notification") # There should be a notification for the assigned servicer (different from performer) servicer_notifs = VehicleNotification.objects.filter(recipient=self.servicer) self.assertTrue(servicer_notifs.exists(), "Assigned servicer should receive a notification when different from performer") class DeleteWorkOrderPermissionTests(TestCase): def setUp(self): User = get_user_model() self.creator = User.objects.create_user(username=f"creator_{uuid.uuid4().hex[:6]}", email=f"creator+{uuid.uuid4()}@example.test", password="pass") self.other = User.objects.create_user(username=f"other_{uuid.uuid4().hex[:6]}", email=f"other+{uuid.uuid4()}@example.test", password="pass") self.staff = User.objects.create_user(username=f"admin_{uuid.uuid4().hex[:6]}", email=f"admin+{uuid.uuid4()}@example.test", password="pass", is_staff=True) self.vehicle = Vehicle.objects.create( registration_number=f"DEL-{uuid.uuid4().hex[:6].upper()}", current_mileage=5000 ) self.wo = WorkOrder.objects.create(vehicle=self.vehicle, creator=self.creator, start_mileage=5000) def test_creator_can_delete_work_order_and_notification_created(self): # Ensure active initially self.wo.refresh_from_db() self.assertTrue(self.wo.is_active) # Delete by creator VehicleService.delete_work_order(self.wo, user=self.creator) # Refresh and assert deactivated self.wo.refresh_from_db() self.assertFalse(self.wo.is_active) # Notification exists for creator notif_exists = VehicleNotification.objects.filter(recipient=self.creator, title__icontains="obrisan").exists() self.assertTrue(notif_exists) def test_non_creator_cannot_delete_work_order(self): with self.assertRaises(DRFValidationError): VehicleService.delete_work_order(self.wo, user=self.other) def test_staff_can_delete_work_order(self): # staff should be allowed VehicleService.delete_work_order(self.wo, user=self.staff) self.wo.refresh_from_db() self.assertFalse(self.wo.is_active) # Notification exists for creator as well (as per implementation) self.assertTrue(VehicleNotification.objects.filter(recipient=self.creator, title__icontains="obrisan").exists())