Increase debugability for tristates

changelog.d/+log-tristates.changed.md

@@ -0,0 +1,2 @@
+Changed how tristates (open, acked, stateful) are logged in order to improve
+debuggability.

src/argus/notificationprofile/models.py

@@ -4,7 +4,7 @@
 from functools import reduce
 import logging
 from operator import or_
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Dict, Optional
 
 from django.conf import settings
 from django.contrib.postgres.fields import ArrayField
@@ -19,6 +19,8 @@
 if TYPE_CHECKING:
     from argus.incident.models import Event, Incident
 
+TriState = Optional[bool]
+
 
 LOG = logging.getLogger(__name__)
 
@@ -138,17 +140,20 @@ def is_empty(self):
             and self.is_event_type_empty()
         )
 
-    def incident_fits_tristates(self, incident):
+    def get_incident_tristate_checks(self, incident) -> Dict[str, TriState]:
         if self.are_tristates_empty():
-            return None
-        fits_tristates = []
+            return {}
+        fits_tristates = {}
         for tristate in self.TRINARY_FILTERS:
             filter_tristate = self._get_tristate(tristate)
             if filter_tristate is None:
+                LOG.debug('Tristates: "%s" not in filter, ignoring', tristate)
+                fits_tristates[tristate] = None
                 continue
             incident_tristate = getattr(incident, tristate, None)
-            fits_tristates.append(filter_tristate == incident_tristate)
-        return all(fits_tristates)
+            LOG.debug('Tristates: "%s": filter = %s, incident = %s', tristate, filter_tristate, incident_tristate)
+            fits_tristates[tristate] = filter_tristate == incident_tristate
+        return fits_tristates
 
     def incident_fits_maxlevel(self, incident):
         if self.is_maxlevel_empty():
@@ -277,7 +282,9 @@ def incident_fits(self, incident: Incident):
         checks = {}
         checks["source"] = self.source_system_fits(incident, data)
         checks["tags"] = self.tags_fit(incident, data)
-        checks["tristates"] = self.filter_wrapper.incident_fits_tristates(incident)
+        tristate_checks = self.filter_wrapper.get_incident_tristate_checks(incident)
+        for tristate, result in tristate_checks.items():
+            checks[tristate] = result
         checks["max_level"] = self.filter_wrapper.incident_fits_maxlevel(incident)
         any_failed = False in checks.values()
         if any_failed:
@@ -372,7 +379,8 @@ def incident_fits(self, incident: Incident):
         if not self.active:
             return False
         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())
+        checks = {f: f.incident_fits(incident) for f in self.filters.all()}
+        is_selected_by_filters = False not in checks.values()
         return is_selected_by_time and is_selected_by_filters
 
     def event_fits(self, event: Event):

tests/notificationprofile/test_models.py

@@ -79,52 +79,66 @@ class FilterWrapperIncidentFitsTristatesTests(unittest.TestCase):
     # A tristate must be one of True, False, None
     # "None" is equivalent to the tristate not being mentioned in the filter at all
 
-    def test_incident_fits_tristates_no_tristates_set(self):
+    def test_get_incident_tristate_checks_no_tristates_set(self):
         incident = Mock()
         empty_filter = FilterWrapper({})
-        result = empty_filter.incident_fits_tristates(incident)
-        self.assertEqual(result, None)
+        result = empty_filter.get_incident_tristate_checks(incident)
+        self.assertEqual(result, {})
 
     @override_settings(ARGUS_FALLBACK_FILTER={"acked": True})
-    def test_incident_fits_tristates_no_tristates_set_with_fallback(self):
+    def test_get_incident_tristate_checks_no_tristates_set_with_fallback(self):
         incident = Mock()
         # Shouldn't match
         incident.acked = False
         empty_filter = FilterWrapper({})
-        result = empty_filter.incident_fits_tristates(incident)
-        self.assertEqual(result, False)
+        result = empty_filter.get_incident_tristate_checks(incident)
+        self.assertEqual(result["open"], None)
+        self.assertEqual(result["acked"], False)
+        self.assertEqual(result["stateful"], None)
         # Should match
         incident.acked = True
         empty_filter = FilterWrapper({})
-        result = empty_filter.incident_fits_tristates(incident)
-        self.assertEqual(result, True)
+        result = empty_filter.get_incident_tristate_checks(incident)
+        self.assertNotIn(False, result.values())
+        self.assertEqual(result["open"], None)
+        self.assertEqual(result["acked"], True)
+        self.assertEqual(result["stateful"], None)
 
-    def test_incident_fits_tristates_is_true(self):
+    def test_get_incident_tristate_checks_is_true(self):
         incident = Mock()
         incident.open = True
         incident.acked = False
         incident.stateful = True
-        empty_filter = FilterWrapper({"open": True, "acked": False})
-        result = empty_filter.incident_fits_tristates(incident)
-        self.assertTrue(result)
-
-    def test_incident_fits_tristates_is_false(self):
+        filter = FilterWrapper({"open": True, "acked": False})
+        result = filter.get_incident_tristate_checks(incident)
+        self.assertTrue(set(result.values()))  # result not empty
+        self.assertEqual(result["open"], True)
+        self.assertEqual(result["acked"], True)
+        self.assertEqual(result["stateful"], None)
+
+    def test_get_incident_tristate_checks_is_false(self):
         incident = Mock()
         incident.open = True
         incident.acked = False
         incident.stateful = True
-        empty_filter = FilterWrapper({"open": False, "acked": False})
-        result = empty_filter.incident_fits_tristates(incident)
-        self.assertFalse(result)
+        filter = FilterWrapper({"open": False, "acked": False})
+        result = filter.get_incident_tristate_checks(incident)
+        self.assertIn(False, result.values())
+        self.assertEqual(result["open"], False)
+        self.assertEqual(result["acked"], True)
+        self.assertEqual(result["stateful"], None)
 
     @override_settings(ARGUS_FALLBACK_FILTER={"acked": True})
-    def test_incident_fits_tristates_fallback_should_not_override(self):
+    def test_get_incident_tristate_checks_fallback_should_not_override(self):
         incident = Mock()
         # Should match
         incident.acked = False
         filter = FilterWrapper({"acked": False})
-        result = filter.incident_fits_tristates(incident)
-        self.assertEqual(result, True)
+        result = filter.get_incident_tristate_checks(incident)
+        self.assertNotIn(False, result.values())
+        self.assertEqual(result["open"], None)
+        self.assertEqual(result["acked"], True)
+        self.assertEqual(result["stateful"], None)
 
 
 @tag("unittest")