201 lines
8.3 KiB
Python
201 lines
8.3 KiB
Python
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()) |