From 0bdf44c90c0968132209109535b9dd6bc9f1d8eb Mon Sep 17 00:00:00 2001 From: Karl Determe Date: Fri, 14 Aug 2026 21:30:04 +0200 Subject: [PATCH] test: add integration tests for interventions and update models logic --- streetup/interventions/models.py | 56 ++++++++++ streetup/interventions/tests/test_general.py | 108 +++++++++++++++++++ 2 files changed, 164 insertions(+) diff --git a/streetup/interventions/models.py b/streetup/interventions/models.py index 9fbe651..ef46c08 100644 --- a/streetup/interventions/models.py +++ b/streetup/interventions/models.py @@ -505,6 +505,8 @@ class Intervention(models.Model): if old_status and old_status != self.status: self.notify_subscribers(old_status, self.status) + if self.status == 'on_pause': + self.notify_manager_on_pause() def notify_subscribers(self, old_status, new_status): from django.contrib.auth.models import User @@ -515,6 +517,8 @@ class Intervention(models.Model): subscribers = User.objects.filter(intervention_subscriptions__intervention=self) if self.last_modified_by: subscribers = subscribers.exclude(pk=self.last_modified_by.pk) + if self.status == 'on_pause' and self.intervention_manager_id: + subscribers = subscribers.exclude(pk=self.intervention_manager_id) if subscribers.exists(): old_label = dict(STATUS_CHOICES).get(old_status, old_status) @@ -547,6 +551,58 @@ class Intervention(models.Model): f"Error sending subscription notification for intervention {self.code}" ) + def notify_manager_on_pause(self): + """Notifie le gestionnaire de l'intervention lorsqu'elle passe en pause.""" + if not self.intervention_manager_id: + return + + # Ne pas notifier le gestionnaire s'il est lui-même l'auteur de la mise en pause + if self.last_modified_by_id and self.last_modified_by_id == self.intervention_manager_id: + return + + from django.urls import reverse + from django.utils.translation import gettext as _ + from notifications.services import notify_user + + pause_reason_label = dict(PAUSE_REASON_CHOICES).get(self.pause_reason, self.pause_reason) + if self.pause_reason == 'other' and self.pause_reason_other: + reason_text = f"{pause_reason_label} : {self.pause_reason_other}" + elif pause_reason_label: + reason_text = str(pause_reason_label) + else: + reason_text = None + + title = _("Intervention %(code)s mise en pause") % {'code': self.code} + if reason_text: + content = _("L'intervention « %(title)s » a été mise en pause.\nMotif : %(reason)s") % { + 'title': self.title, + 'reason': reason_text, + } + else: + content = _("L'intervention « %(title)s » a été mise en pause.") % { + 'title': self.title, + } + + action_url = reverse('interventions:intervention_detail', args=[self.id]) + + try: + notify_user( + recipient=self.intervention_manager, + title=title, + content=content, + notification_type='alert', + action_url=action_url, + action_label=_("Voir l'intervention"), + source_app='interventions', + sender=self.last_modified_by, + is_system=True, + ) + except Exception: + import logging + logging.getLogger(__name__).exception( + f"Error sending pause notification to manager for intervention {self.code}" + ) + def sync_daily_assignments(self): """Synchronise les assignations quotidiennes de l'intervention.""" start_dt = self.planned_begin_time or self.begin_time or self.expected_begin_time or self.creation_time diff --git a/streetup/interventions/tests/test_general.py b/streetup/interventions/tests/test_general.py index c7f10d2..418a65d 100644 --- a/streetup/interventions/tests/test_general.py +++ b/streetup/interventions/tests/test_general.py @@ -675,6 +675,114 @@ class InterventionManagerTests(TestCase): output = pdf.output(dest='S') self.assertIsNotNone(output) + def test_pause_notification_sent_to_manager_when_paused_by_other(self): + from notifications.models import NotificationRecipient + User = get_user_model() + other_user = User.objects.create_user(username='technician_user', password='pwd') + other_config = UserConfig.objects.create(user=other_user, is_intern=True) + other_config.roles.add(self.role_admin) + UserThematics.objects.create( + user_config=other_config, + thematic=self.thematic, + can_view_interventions=True, + can_edit_interventions=True, + ) + intervention = Intervention.objects.create( + title='Pause Notification Test', + status='in_progress', + contract=self.contract, + thematic=self.thematic, + symptom=self.symptom, + intervention_manager=self.user, + ) + + from interventions.views import attempt_status_update + request = mock.Mock() + success, msg, reload = attempt_status_update( + request, + intervention, + other_user, + 'on_pause', + pause_reason='order_material', + ) + self.assertTrue(success, msg) + intervention.refresh_from_db() + self.assertEqual(intervention.status, 'on_pause') + + recipients = NotificationRecipient.objects.filter(user=self.user) + self.assertTrue(recipients.exists()) + notif = recipients.last().notification + self.assertIn(intervention.code, notif.title) + self.assertEqual(notif.notification_type, 'alert') + from interventions.models import PAUSE_REASON_CHOICES + expected_reason = str(dict(PAUSE_REASON_CHOICES).get('order_material')) + self.assertIn(expected_reason, notif.content) + self.assertEqual(notif.sender, other_user) + + def test_pause_notification_not_sent_when_paused_by_manager_themself(self): + from notifications.models import NotificationRecipient + intervention = Intervention.objects.create( + title='Pause By Manager Test', + status='in_progress', + contract=self.contract, + thematic=self.thematic, + symptom=self.symptom, + intervention_manager=self.user, + ) + + from interventions.views import attempt_status_update + request = mock.Mock() + success, msg, reload = attempt_status_update( + request, + intervention, + self.user, + 'on_pause', + pause_reason='waiting_for_supplier', + ) + self.assertTrue(success, msg) + intervention.refresh_from_db() + self.assertEqual(intervention.status, 'on_pause') + + # No notification should be sent to the manager themselves + recipients = NotificationRecipient.objects.filter(user=self.user) + self.assertFalse(recipients.exists()) + + def test_pause_notification_not_sent_when_no_manager(self): + from notifications.models import NotificationRecipient + User = get_user_model() + other_user = User.objects.create_user(username='tech2', password='pwd') + other_config = UserConfig.objects.create(user=other_user, is_intern=True) + other_config.roles.add(self.role_admin) + UserThematics.objects.create( + user_config=other_config, + thematic=self.thematic, + can_view_interventions=True, + can_edit_interventions=True, + ) + intervention = Intervention.objects.create( + title='Pause No Manager Test', + status='in_progress', + contract=self.contract, + thematic=self.thematic, + symptom=self.symptom, + intervention_manager=None, + ) + + from interventions.views import attempt_status_update + request = mock.Mock() + success, msg, reload = attempt_status_update( + request, + intervention, + other_user, + 'on_pause', + pause_reason='reschedule', + ) + self.assertTrue(success, msg) + intervention.refresh_from_db() + self.assertEqual(intervention.status, 'on_pause') + + self.assertEqual(NotificationRecipient.objects.count(), 0) + def test_geojson_filtering_by_assigned_team(self): from contracts.models import CompanyTeam, CompanyMember from django.http import QueryDict