Add tests for planned maintenance

johannaengland · hmpf · commit f897c15ce06b · 2025-11-27T14:31:57.000+01:00
diff --git a/tests/plannedmaintenance/test_models.py b/tests/plannedmaintenance/test_models.py
@@ -3,29 +3,78 @@
 from django.test import TestCase, tag
 from django.utils import timezone
 
-from argus.auth.factories import PersonUserFactory
+from argus.plannedmaintenance.factories import PlannedMaintenanceFactory
 from argus.plannedmaintenance.models import MODIFICATION_WINDOW_PM, PlannedMaintenanceTask
 from argus.util.testing import disconnect_signals, connect_signals
 
 
+@tag("database")
+class PlannedMaintenanceQuerySetTests(TestCase):
+    def setUp(self):
+        disconnect_signals()
+
+        now = timezone.now()
+        self.future_pm = PlannedMaintenanceFactory(start_time=now + timedelta(days=1))
+        self.current_pm = PlannedMaintenanceFactory(start_time=now - timedelta(minutes=5))
+        self.past_pm = PlannedMaintenanceFactory(start_time=now - timedelta(days=1), end_time=now - timedelta(hours=12))
+
+    def teardown(self):
+        connect_signals()
+
+    def test_future_returns_only_pms_with_start_time_in_future(self):
+        future_pms = PlannedMaintenanceTask.objects.future()
+
+        self.assertNotIn(self.current_pm, future_pms)
+        self.assertNotIn(self.past_pm, future_pms)
+        self.assertIn(self.future_pm, future_pms)
+
+    def test_past_returns_only_pms_with_end_time_in_past(self):
+        past_pms = PlannedMaintenanceTask.objects.past()
+
+        self.assertNotIn(self.current_pm, past_pms)
+        self.assertNotIn(self.future_pm, past_pms)
+        self.assertIn(self.past_pm, past_pms)
+
+    def test_current_returns_only_pms_with_start_time_in_past_and_end_time_in_future(self):
+        current_pms = PlannedMaintenanceTask.objects.current()
+
+        self.assertNotIn(self.future_pm, current_pms)
+        self.assertNotIn(self.past_pm, current_pms)
+        self.assertIn(self.current_pm, current_pms)
+
+    def test_active_at_time_returns_only_pms_with_start_time_before_and_end_time_after(self):
+        active_at_time_pms = PlannedMaintenanceTask.objects.active_at_time(
+            self.future_pm.start_time + timedelta(minutes=1)
+        )
+
+        self.assertNotIn(self.past_pm, active_at_time_pms)
+        self.assertIn(self.future_pm, active_at_time_pms)
+        self.assertIn(self.current_pm, active_at_time_pms)
+
+
 @tag("database")
 class PlannedMaintenanceTaskTests(TestCase):
     def setUp(self):
         disconnect_signals()
 
-        self.user = PersonUserFactory()
+        now = timezone.now()
+        self.future_pm = PlannedMaintenanceFactory(start_time=now + timedelta(days=1))
+        self.current_pm = PlannedMaintenanceFactory(start_time=now - timedelta(minutes=5))
+        self.past_pm = PlannedMaintenanceFactory(
+            start_time=timezone.now() - MODIFICATION_WINDOW_PM - timedelta(hours=2),
+            end_time=timezone.now() - MODIFICATION_WINDOW_PM - timedelta(hours=1),
+        )
 
     def teardown(self):
         connect_signals()
 
     def test_given_open_pm_task_modifiable_is_true(self):
-        open_pm_task = PlannedMaintenanceTask(owner=self.user)
-
+        open_pm_task = PlannedMaintenanceTask(created_by=self.user)
         self.assertTrue(open_pm_task.modifiable)
 
     def test_given_recently_ended_pm_task_modifiable_is_true(self):
-        open_pm_task = PlannedMaintenanceTask(
-            owner=self.user,
+        open_pm_task = PlannedMaintenanceFactory(
+            created_by=self.user,
             start_time=timezone.now() - timedelta(hours=1),
             end_time=timezone.now(),
         )
@@ -34,9 +83,42 @@ def test_given_recently_ended_pm_task_modifiable_is_true(self):
 
     def test_given_long_ago_ended_pm_task_modifiable_is_false(self):
         open_pm_task = PlannedMaintenanceTask(
-            owner=self.user,
+            created_by=self.user,
             start_time=timezone.now() - MODIFICATION_WINDOW_PM - timedelta(hours=2),
             end_time=timezone.now() - MODIFICATION_WINDOW_PM - timedelta(hours=1),
         )
+        self.assertFalse(self.past_pm.modifiable)
+
+    def test_given_active_pm_current_is_true(self):
+        pm = PlannedMaintenanceFactory(start_time=timezone.now() - timedelta(minutes=5))
+        self.assertTrue(pm.current)
+
+    def test_given_past_pm_current_is_false(self):
+        now = timezone.now()
+        pm = PlannedMaintenanceFactory(start_time=now - timedelta(days=1), end_time=now - timedelta(hours=12))
+        self.assertFalse(pm.current)
+
+    def test_given_future_pm_current_is_false(self):
+        pm = PlannedMaintenanceFactory(start_time=timezone.now() + timedelta(days=1))
+        self.assertFalse(pm.current)
+
+    def test_given_future_pm_cancel_deletes_task(self):
+        pm_id = self.future_pm.id
+        self.future_pm.cancel()
+
+        self.assertFalse(PlannedMaintenanceTask.objects.filter(id=pm_id).exists())
+
+    def test_given_past_pm_cancel_does_nothing(self):
+        past_end_time = self.past_pm.end_time
+        self.past_pm.cancel()
+        self.past_pm.refresh_from_db()
+
+        self.assertEqual(self.past_pm.end_time, past_end_time)
+
+    def test_given_current_pm_cancel_sets_end_time(self):
+        current_pm_end_time = self.current_pm.end_time
+        self.current_pm.cancel()
+        self.current_pm.refresh_from_db()
 
-        self.assertFalse(open_pm_task.modifiable)
+        self.assertNotEqual(self.current_pm.end_time, current_pm_end_time)
+        self.assertLess(self.current_pm.end_time, timezone.now())
diff --git a/tests/plannedmaintenance/test_utils.py b/tests/plannedmaintenance/test_utils.py
@@ -0,0 +1,113 @@
+from datetime import timedelta
+
+from django.test import TestCase, tag
+from django.utils import timezone
+
+from argus.filter.factories import FilterFactory
+from argus.incident.factories import EventFactory, StatefulIncidentFactory
+from argus.incident.models import Event, Incident
+from argus.plannedmaintenance.factories import PlannedMaintenanceFactory
+from argus.plannedmaintenance.models import PlannedMaintenanceTask
+from argus.plannedmaintenance.utils import (
+    event_covered_by_planned_maintenance,
+    incidents_covered_by_planned_maintenance_task,
+)
+from argus.util.testing import disconnect_signals, connect_signals
+
+
+@tag("database")
+class TestIncidentsCoveredByPlannedMaintenanceTask(TestCase):
+    def setUp(self):
+        disconnect_signals()
+        self.open_incident = StatefulIncidentFactory(level=5)
+        self.maxlevel_filter = FilterFactory(filter={"maxlevel": 5})
+
+    def teardown(self):
+        connect_signals()
+
+    def test_returns_only_open_incidents(self):
+        closed_incident = StatefulIncidentFactory(end_time=timezone.now())
+        pm = PlannedMaintenanceFactory()
+        pm.filters.add(self.maxlevel_filter)
+        incidents = incidents_covered_by_planned_maintenance_task(queryset=Incident.objects.all(), pm_task=pm)
+
+        self.assertIn(self.open_incident, incidents)
+        self.assertNotIn(closed_incident, incidents)
+
+    def test_given_past_pm_returns_no_incidents(self):
+        pm = PlannedMaintenanceFactory(
+            start_time=timezone.now() - timedelta(days=2), end_time=timezone.now() - timedelta(days=1)
+        )
+        pm.filters.add(self.maxlevel_filter)
+
+        incidents = incidents_covered_by_planned_maintenance_task(queryset=Incident.objects.all(), pm_task=pm)
+        self.assertFalse(incidents)
+
+    def test_given_future_pm_returns_no_incidents(self):
+        pm = PlannedMaintenanceFactory(start_time=timezone.now() + timedelta(days=2))
+        pm.filters.add(self.maxlevel_filter)
+
+        incidents = incidents_covered_by_planned_maintenance_task(queryset=Incident.objects.all(), pm_task=pm)
+        self.assertFalse(incidents)
+
+    def test_given_filter_only_returns_incidents_matching_filter(self):
+        important_incident = StatefulIncidentFactory(level=1)
+        pm = PlannedMaintenanceFactory()
+        pm.filters.add(FilterFactory(filter={"maxlevel": 1}))
+
+        incidents = incidents_covered_by_planned_maintenance_task(queryset=Incident.objects.all(), pm_task=pm)
+
+        self.assertIn(important_incident, incidents)
+        self.assertNotIn(self.open_incident, incidents)
+
+
+@tag("database")
+class TestEventCoveredByPlannedMaintenanceTasks(TestCase):
+    def setUp(self):
+        disconnect_signals()
+        self.start_event = EventFactory(type=Event.Type.INCIDENT_START)
+        self.current_pm = PlannedMaintenanceFactory()
+        self.maxlevel_filter = FilterFactory(filter={"maxlevel": 5})
+
+    def teardown(self):
+        connect_signals()
+
+    def test_given_relevant_event_return_true(self):
+        self.current_pm.filters.add(self.maxlevel_filter)
+
+        covered = event_covered_by_planned_maintenance(self.start_event)
+
+        self.assertTrue(covered)
+
+    def test_given_wrong_event_type_return_false(self):
+        self.current_pm.filters.add(FilterFactory(filter={"event_type": Event.Type.ACKNOWLEDGE}))
+
+        covered = event_covered_by_planned_maintenance(self.start_event)
+
+        self.assertFalse(covered)
+
+    def test_given_wrong_maxlevel_return_false(self):
+        self.current_pm.filters.add(FilterFactory(filter={"maxlevel": 1}))
+
+        covered = event_covered_by_planned_maintenance(self.start_event)
+
+        self.assertFalse(covered)
+
+    def test_given_pm_with_multiple_filters_only_matching_one_filter_return_false(self):
+        self.current_pm.filters.add(FilterFactory(filter={"maxlevel": 1}))
+        self.current_pm.filters.add(FilterFactory(filter={"event_type": Event.Type.INCIDENT_START}))
+
+        covered = event_covered_by_planned_maintenance(self.start_event)
+
+        self.assertFalse(covered)
+
+    def test_given_no_pms_active_at_time_return_false(self):
+        now = timezone.now()
+        past_pm = PlannedMaintenanceFactory(start_time=now - timedelta(days=5), end_time=now - timedelta(days=4))
+        past_pm.filters.add(self.maxlevel_filter)
+
+        covered = event_covered_by_planned_maintenance(
+            self.start_event, PlannedMaintenanceTask.objects.filter(id=past_pm.id)
+        )
+
+        self.assertFalse(covered)
