Merge pull request Amsterdam#1199 from Amsterdam/chore/celery-task-to…

…-call-clearsession-command

Added a Celery task called "clearsessions"

app/signals/apps/signals/tasks.py

@@ -1,7 +1,8 @@
 # SPDX-License-Identifier: MPL-2.0
-# Copyright (C) 2018 - 2022 Gemeente Amsterdam
+# Copyright (C) 2018 - 2023 Gemeente Amsterdam
 import logging
 
+from django.core.management import call_command
 from django.db import connection
 from django.db.models import Q
 from django.utils import timezone
@@ -129,3 +130,11 @@ def refresh_materialized_view_public_signals_geography_feature_collection():
         log.error(f'Failed to execute the query: {refresh_query}', exc_info=e)
     finally:
         cursor.close()
+
+
+@app.task
+def clearsessions():
+    """
+    This task makes it possible to configure the "clearsession" management command through Celery
+    """
+    call_command('clearsessions')

app/signals/apps/signals/tests/test_tasks.py

@@ -1,13 +1,18 @@
 # SPDX-License-Identifier: MPL-2.0
-# Copyright (C) 2018 - 2022 Gemeente Amsterdam, Vereniging van Nederlandse Gemeenten
+# Copyright (C) 2018 - 2023 Gemeente Amsterdam, Vereniging van Nederlandse Gemeenten
+from unittest import mock
 from unittest.mock import call, patch
 
 from django.test import TestCase, TransactionTestCase
 
 from signals.apps.signals import factories
 from signals.apps.signals.models import Status
 from signals.apps.signals.models.signal import Signal
-from signals.apps.signals.tasks import delete_closed_signals, update_status_children_based_on_parent
+from signals.apps.signals.tasks import (
+    clearsessions,
+    delete_closed_signals,
+    update_status_children_based_on_parent
+)
 from signals.apps.signals.workflow import AFGEHANDELD, AFWACHTING, GEANNULEERD, GESPLITST
 
 
@@ -203,3 +208,11 @@ def test_service_is_called(self, mocked_delete_signals):
             call(GEANNULEERD, 365, dry_run=False),
             call(GESPLITST, 365, dry_run=False),
         ])
+
+
+class TestClearSessionsTask(TestCase):
+    @mock.patch('signals.apps.signals.tasks.call_command', autospec=True)
+    def test_clearsessions_is_called(self, mocked_call_command):
+        clearsessions()
+
+        mocked_call_command.assert_called_once_with('clearsessions')