prvi kod za live uporabu

backend/modules/fleet/tests/test_notifications.py

@@ -0,0 +1,201 @@
+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())