Merge pull request #249 from hmpf/fix-236

Fix sending notifications to multiple profiles

src/argus/notificationprofile/factories.py

@@ -0,0 +1,74 @@
+from datetime import time
+
+import factory
+
+from argus.auth.factories import PersonUserFactory
+from argus.notificationprofile import models
+
+
+__all__ = [
+    "TimeslotFactory",
+    "TimeRecurrenceFactory",
+    "MinimalTimeRecurrenceFactory",
+    "MaximalTimeRecurrenceFactory",
+    "NotificationProfileFactory",
+    "FilterFactory",
+]
+
+
+class TimeslotFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = models.Timeslot
+
+    user = factory.SubFactory(PersonUserFactory)
+    name = factory.Faker("word")
+
+
+class TimeRecurrenceFactory(factory.django.DjangoModelFactory):
+    "A random TimeRecurrence"
+
+    class Meta:
+        model = models.TimeRecurrence
+
+    timeslot = factory.SubFactory(TimeslotFactory)
+    start = factory.Faker("time_object")
+    end = factory.Faker("time_object")
+    days = factory.Faker("random_sample", elements=[models.TimeRecurrence.Day])
+
+
+class MinimalTimeRecurrenceFactory(TimeRecurrenceFactory):
+    "A TimeRecurrence that should just about never occur"
+    start = time(hour=5, minute=0)
+    end = start
+    days = [7]
+
+
+class MaximalTimeRecurrenceFactory(TimeRecurrenceFactory):
+    "A TimeRecurrence that always occurs"
+
+    class Meta:
+        model = models.TimeRecurrence
+
+    timeslot = factory.SubFactory(TimeslotFactory)
+    start = time.min
+    end = time.max
+    days = list(range(1, 7 + 1))
+
+
+class NotificationProfileFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = models.NotificationProfile
+
+    user = factory.SubFactory(PersonUserFactory, user=factory.SelfAttribute("..timeslot"))
+    timeslot = factory.SubFactory(TimeslotFactory)
+    media = models.NotificationProfile.Media.EMAIL
+    active = factory.Faker("boolean")
+
+
+class FilterFactory(factory.django.DjangoModelFactory):
+    class Meta:
+        model = models.Filter
+
+    user = factory.SubFactory(PersonUserFactory)
+    name = factory.Faker("word")
+    filter_string = '{"sourceSystemIds": [], "tags": []}'

src/argus/notificationprofile/media/__init__.py

@@ -56,9 +56,11 @@ def send_notifications_to_users(event: Event):
     LOG.info('Notification: sending event "%s"', event)
     sent = False
     for profile in NotificationProfile.objects.select_related("user"):
+        LOG.debug('Notification: checking profile "%s"', profile)
         if profile.incident_fits(event.incident):
             send_notification(profile, event)
             sent = True
+        LOG.debug('Notification: sent? %s: profile "%s", event "%s"', sent, profile, event)
     if not sent:
         LOG.info('Notification: no listeners for "%s"', event)
         return

src/argus/notificationprofile/media/email.py

@@ -16,7 +16,7 @@
 
 
 def modelinstance_to_dict(obj):
-    dict_ = vars(obj)
+    dict_ = vars(obj).copy()
     dict_.pop("_state")
     return dict_
 

src/argus/notificationprofile/models.py

@@ -202,8 +202,9 @@ def filtered_incidents(self):
         return reduce(or_, qs)
 
     def incident_fits(self, incident: Incident):
+        assert incident.source, "incident does not have a source -2"
         if not self.active:
             return False
-        return self.timeslot.timestamp_is_within_time_recurrences(incident.start_time) and any(
-            f.incident_fits(incident) for f in self.filters.all()
-        )
+        is_selected_by_time = self.timeslot.timestamp_is_within_time_recurrences(incident.start_time)
+        is_selected_by_filters = any(f.incident_fits(incident) for f in self.filters.all())
+        return is_selected_by_time and is_selected_by_filters

tests/notificationprofile/bugfixes/test_github236.py

@@ -0,0 +1,61 @@
+from django.core import mail
+from django.test import TestCase
+import json
+
+from argus.auth.factories import PersonUserFactory
+from argus.incident.models import create_fake_incident, get_or_create_default_instances
+from argus.notificationprofile.factories import (
+    TimeslotFactory,
+    NotificationProfileFactory,
+    FilterFactory,
+    MaximalTimeRecurrenceFactory,
+    MinimalTimeRecurrenceFactory,
+)
+from argus.notificationprofile.media import send_notifications_to_users
+from argus.notificationprofile.models import Timeslot, NotificationProfile, Filter
+from argus.util.testing import disconnect_signals, connect_signals
+
+
+"""See:
+
+https://github.com/Uninett/Argus/issues/236
+"""
+
+
+class SendingNotificationTest(TestCase):
+    def setUp(self):
+        disconnect_signals()
+
+        # Create two separate timeslots
+        user = PersonUserFactory()
+        timeslot1 = TimeslotFactory(user=user)
+        MaximalTimeRecurrenceFactory(timeslot=timeslot1)
+        timeslot2 = TimeslotFactory(user=user)
+        MinimalTimeRecurrenceFactory(timeslot=timeslot2)
+
+        # Create a filter that matches your test incident
+        (_, _, argus_source) = get_or_create_default_instances()
+        filter_dict = {"sourceSystemIds": [argus_source.id], "tags": []}
+        filter_string = json.dumps(filter_dict)
+        filter = FilterFactory(user=user, filter_string=filter_string)
+
+        # Create two notification profiles that match this filter, but attached to each their timeslot
+        self.np1 = NotificationProfileFactory(user=user, timeslot=timeslot1, active=True)
+        self.np1.filters.add(filter)
+        self.np2 = NotificationProfileFactory(user=user, timeslot=timeslot2, active=True)
+        self.np2.filters.add(filter)
+
+    def tearDown(self):
+        connect_signals()
+
+    def test_sending_event_to_multiple_profiles_of_the_same_user_should_not_raise_exception(self):
+        LOG_PREFIX = "INFO:argus.notificationprofile.media:"
+        # Send a test event
+        self.incident = create_fake_incident()
+        event = self.incident.events.get(type="STA")
+        with self.settings(SEND_NOTIFICATIONS=True):
+            try:
+                send_notifications_to_users(event)
+            except AttributeError:
+                self.fail("send_notifications_to_users() should not raise an AttributeError")
+            self.assertTrue(bool(mail.outbox), "Mail should have been sent")

tests/notificationprofile/test_media.py

@@ -0,0 +1,13 @@
+from django.test import TestCase
+
+from argus.notificationprofile.factories import TimeslotFactory
+from argus.notificationprofile.media.email import modelinstance_to_dict
+
+
+class SerializeModelTest(TestCase):
+    def test_modelinstance_to_dict_should_not_change_modelinstance(self):
+        instance = TimeslotFactory()
+        attributes1 = vars(instance)
+        modelinstance_to_dict(instance)
+        attributes2 = vars(instance)
+        self.assertEqual(attributes1, attributes2)