adds when filter

app/signals/apps/api/tests/test_history_endpoint.py

@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: MPL-2.0
 # Copyright (C) 2019 - 2023 Vereniging van Nederlandse Gemeenten, Gemeente Amsterdam
 import os
+import urllib.parse
 from datetime import timedelta
 
 from django.contrib.auth.models import Permission
@@ -25,7 +26,7 @@
 
 
 class TestHistoryAction(SIAReadWriteUserMixin, SignalsBaseApiTestCase):
-    def setUp(self):
+    def setUp(self) -> None:
         slo = ServiceLevelObjectiveFactory.create()
         self.signal = SignalFactory.create(user_assignment=None, category_assignment__category=slo.category)
         SignalLogService.log_create_initial(self.signal)
@@ -38,7 +39,7 @@ def setUp(self):
             )
         )
 
-    def test_history_action_present(self):
+    def test_history_action_present(self) -> None:
         response = self.client.get(f'/signals/v1/private/signals/{self.signal.id}/history')
         self.assertEqual(response.status_code, 401)
 
@@ -51,7 +52,7 @@ def test_history_action_present(self):
         data = response.json()
         self.assertJsonSchema(self.list_history_schema, data)
 
-    def test_history_entry_contents(self):
+    def test_history_entry_contents(self) -> None:
         keys = ['identifier', 'when', 'what', 'action', 'description', 'who', '_signal']
 
         self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
@@ -67,7 +68,7 @@ def test_history_entry_contents(self):
             for k in keys:
                 self.assertIn(k, entry)
 
-    def test_update_shows_up(self):
+    def test_update_shows_up(self) -> None:
         # Get a baseline for the Signal history
         self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
         self.client.force_authenticate(user=self.sia_read_write_user)
@@ -99,7 +100,7 @@ def test_update_shows_up(self):
         self.assertEqual(new_entry['who'], self.user.username)
         self.assertEqual(new_entry['description'], status.text)
 
-    def test_sla_in_history(self):
+    def test_sla_in_history(self) -> None:
         # Get a baseline for the Signal history
         self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
         self.client.force_authenticate(user=self.sia_read_write_user)
@@ -124,7 +125,7 @@ def test_sla_in_history(self):
         self.assertEqual(self.signal.category_assignments.all().order_by('created_at')[0].category.handling_message,
                          sla_description_in_history[0])
 
-    def test_sla_only_once_in_history(self):
+    def test_sla_only_once_in_history(self) -> None:
         # Update the Signal category, and check that we only have the original SLA handling message in the history
         now = timezone.now()
         hours = 1
@@ -174,7 +175,7 @@ def test_sla_only_once_in_history(self):
         self.assertEqual(self.signal.category_assignments.all().order_by('created_at')[0].category.handling_message,
                          sla_description_in_history[0])
 
-    def test_handling_message_in_history_is_constant(self):
+    def test_handling_message_in_history_is_constant(self) -> None:
         # SIG-3555 because history is not tracked for the Category model and its
         # handling_message. Changing the handling_message will also change the
         # history of every Signal that has that Category associated with.
@@ -187,7 +188,7 @@ def test_handling_message_in_history_is_constant(self):
         cat2 = CategoryFactory.create(name='cat2')
         ServiceLevelObjectiveFactory.create(category=cat2)
 
-        signal = SignalFactoryValidLocation(category_assignment__category=cat1)
+        signal = SignalFactoryValidLocation.create(category_assignment__category=cat1)
         SignalLogService.log_create_initial(signal)
 
         self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
@@ -222,7 +223,7 @@ def test_handling_message_in_history_is_constant(self):
         self.assertEqual(category_assignments[0].stored_handling_message, message_1)
         self.assertEqual(category_assignments[2].stored_handling_message, message_2)
 
-    def test_null_stored_handling_message(self):
+    def test_null_stored_handling_message(self) -> None:
         # SIG-3555 old complaints/Signals will have no stored handling messages on their associated CategoryAssignments.
         # This test fakes that by setting one category's handling message to null and checks that the correct
         # placeholder is returned.
@@ -231,7 +232,7 @@ def test_null_stored_handling_message(self):
         cat1 = CategoryFactory.create(name='cat1', handling_message=message_1)
         ServiceLevelObjectiveFactory.create(category=cat1)
 
-        signal = SignalFactoryValidLocation(category_assignment__category=cat1)
+        signal = SignalFactoryValidLocation.create(category_assignment__category=cat1)
         SignalLogService.log_create_initial(signal)
 
         self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
@@ -242,7 +243,7 @@ def test_null_stored_handling_message(self):
         response_json = response.json()
         self.assertEqual(response_json[0]['description'], 'Servicebelofte onbekend')
 
-    def test_history_no_permissions(self):
+    def test_history_no_permissions(self) -> None:
         """
         The sia_read_user does not have a link with any department and also is not configured with the permission
         "sia_can_view_all_categories". Therefore, it should not be able to see a Signal, and it's history.
@@ -252,16 +253,109 @@ def test_history_no_permissions(self):
         self.assertEqual(response.status_code, 403)
 
 
+class TestHistoryFilters(SIAReadWriteUserMixin, SignalsBaseApiTestCase):
+    def setUp(self) -> None:
+        self.slo = ServiceLevelObjectiveFactory.create()
+
+        # Create a signal that is timestamped yesterday
+        with freeze_time(timezone.now() - timedelta(days=1)):
+            self.signal = SignalFactory.create(user_assignment=None, category_assignment__category=self.slo.category)
+            SignalLogService.log_create_initial(self.signal)
+
+        self.list_history_schema = self.load_json_schema(
+            os.path.join(
+                THIS_DIR,
+                'json_schema',
+                'get_signals_v1_private_signals_{pk}_history.json'
+            )
+        )
+
+        update = {
+            'text': 'DIT IS EEN TEST',
+            'state': workflow.BEHANDELING,
+            'user': self.user,
+        }
+
+        self.status = Signal.actions.update_status(
+            update,
+            self.signal
+        )
+
+        SignalLogService.log_update_status(self.status)
+
+    def test_history_filter_with_created_after(self) -> None:
+        self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
+        self.client.force_authenticate(user=self.sia_read_write_user)
+
+        an_hour_ago = (timezone.now() - timedelta(hours=1)).isoformat()
+        encoded_an_hour_ago = urllib.parse.quote(an_hour_ago)
+
+        response = self.client.get(
+            f'/signals/v1/private/signals/{self.signal.id}/history?created_after={encoded_an_hour_ago}'
+        )
+
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+
+        self.assertJsonSchema(self.list_history_schema, data)
+        self.assertEqual(len(data), 1)
+
+        new_entry = data[0]  # most recent status should be first
+        self.assertEqual(new_entry['who'], self.user.username)
+        self.assertEqual(new_entry['description'], self.status.text)
+        self.assertEqual(new_entry['what'], 'UPDATE_STATUS')
+
+    def test_history_filter_with_created_after_in_the_future(self) -> None:
+        self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
+        self.client.force_authenticate(user=self.sia_read_write_user)
+
+        tomorrow = (timezone.now() + timedelta(days=1)).isoformat()
+        encoded_tomorrow = urllib.parse.quote(tomorrow)
+
+        response = self.client.get(
+            f'/signals/v1/private/signals/{self.signal.id}/history?created_after={encoded_tomorrow}'
+        )
+        self.assertEqual(response.status_code, 200)
+
+        data = response.json()
+        self.assertEqual(len(data), 0)
+
+    def test_history_filter_with_created_after_and_what(self) -> None:
+        self.sia_read_write_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
+        self.client.force_authenticate(user=self.sia_read_write_user)
+
+        day_before_yesterday = (timezone.now() - timedelta(days=2)).isoformat()
+        created_after = urllib.parse.quote(day_before_yesterday)
+
+        what = 'UPDATE_SLA'
+        response = self.client.get(
+            f'/signals/v1/private/signals/{self.signal.id}/history?created_after={created_after}&what={what}'
+        )
+
+        self.assertEqual(response.status_code, 200)
+        data = response.json()
+
+        self.assertJsonSchema(self.list_history_schema, data)
+        self.assertEqual(len(data), 1)
+
+        new_entry = data[0]  # most recent status should be first
+        self.assertEqual(new_entry['description'], self.slo.category.handling_message)
+        self.assertEqual(new_entry['what'], what)
+
+
 class TestHistoryForFeedback(SignalsBaseApiTestCase, SIAReadUserMixin):
-    def setUp(self):
+    def setUp(self) -> None:
         slo = ServiceLevelObjectiveFactory.create()
-        self.signal = SignalFactoryValidLocation(user_assignment=None, category_assignment__category=slo.category)
+        self.signal = SignalFactoryValidLocation.create(
+            user_assignment=None,
+            category_assignment__category=slo.category
+        )
         SignalLogService.log_create_initial(self.signal)
 
         self.feedback_endpoint = '/signals/v1/public/feedback/forms/{token}'
         self.history_endpoint = '/signals/v1/private/signals/{id}/history'
 
-    def test_submit_feedback_check_history(self):
+    def test_submit_feedback_check_history(self) -> None:
         # get a user privileged to read from API
         read_user = self.sia_read_user
         read_user.user_permissions.add(Permission.objects.get(codename='sia_can_view_all_categories'))
@@ -275,7 +369,7 @@ def test_submit_feedback_check_history(self):
         response_data = response.json()
         self.assertEqual(len(response_data), 6)
 
-        feedback = FeedbackFactory(_signal=self.signal, is_satisfied=None)
+        feedback = FeedbackFactory.create(_signal=self.signal, is_satisfied=None)
 
         # Note the unhappy flow regarding feedback is tested in the feedback
         # app. Here we only check that it shows up in the history.
@@ -302,12 +396,12 @@ def test_submit_feedback_check_history(self):
         response_data = response.json()
         self.assertEqual(len(response_data), 1)
 
-    def test_history_entry_description_property(self):
+    def test_history_entry_description_property(self) -> None:
         # equivalent to submitting feedback:
         text = 'TEXT'
         text_extra = 'TEXT_EXTRA'
 
-        feedback = FeedbackFactory(
+        feedback = FeedbackFactory.create(
             _signal=self.signal,
             is_satisfied=True,
             allows_contact=False,
@@ -336,7 +430,7 @@ def test_history_entry_description_property(self):
         self.assertIn(f'Toelichting: {text_extra}', history_entry['description'])
         self.assertIn('Toestemming contact opnemen: Nee', history_entry['description'])
 
-    def test_history_no_permissions(self):
+    def test_history_no_permissions(self) -> None:
         """
         The sia_read_user does not have a link with any department and also is not configured with the permission
         "sia_can_view_all_categories". Therefore, it should not be able to see a Signal, and it's history.

app/signals/apps/api/views/signals/private/signals.py

@@ -143,7 +143,11 @@ def check_object_permissions(self, request, obj):
                 )
 
     @extend_schema(
-        parameters=[OpenApiParameter('what', OpenApiTypes.STR, description='Filters a specific "action"')],
+        parameters=[
+            OpenApiParameter('what', OpenApiTypes.STR, description='Filters a specific "action"'),
+            OpenApiParameter('created_after', OpenApiTypes.DATETIME,
+                             description='Filters actions created after a certain timestamp (ISO 8601)'),
+        ],
         responses={HTTP_200_OK: HistoryLogHalSerializer(many=True)}
     )
     @action(detail=True, url_path='history', filterset_class=None, filter_backends=(), pagination_class=None)
@@ -155,6 +159,8 @@ def history(self, *args, **kwargs):
         history_log_qs = signal.history_log.all()
 
         what = self.request.query_params.get('what', None)
+        created_after = self.request.query_params.get('created_after', None)
+
         if what:
             action = Log.translate_what_to_action(what)
             content_type = Log.translate_what_to_content_type(what)
@@ -165,6 +171,9 @@ def history(self, *args, **kwargs):
 
             history_log_qs = history_log_qs.filter(action__iexact=action, content_type__model__iexact=content_type)
 
+        if created_after:
+            history_log_qs = history_log_qs.filter(created_at__gte=created_after)
+
         serializer = HistoryLogHalSerializer(history_log_qs, many=True)
         return Response(serializer.data)