Delayed job status (#1102)

* tasks and delayed update job status

* tests

* black/lint

* docs

* fix docs

* Update django pin in docs

---------

Co-authored-by: Nathan Swain <swainn@users.noreply.github.com>

docs/docs_environment.yml

@@ -12,7 +12,7 @@ dependencies:
   - python
   - pip
   - tethys_dataset_services >=2.0.0
-  - django =3.2.*
+  - django =4.2.*
   - sphinx
   - sphinx-argparse
   - make

docs/installation/production/manual/configuration/basic/file_permissions.rst

@@ -91,7 +91,7 @@ For convenience, you may consider setting up these or similar aliases in the act
 .. _selinux_configuration:
 
 3. Security-Enhanced Linux File Permissions (Rocky Linux, May not Apply)
-===================================================================
+========================================================================
 
 If you are installing Tethys Portal on a Rocky Linux or RedHat system that has `Security-Enhanced Linux (SELinux) <https://en.wikipedia.org/wiki/Security-Enhanced_Linux>`_ enabled and set to enforcing mode, you may need to perform additional setup to allow the server processes to access files.
 

docs/installation/production/manual/configuration/basic/supervisor.rst

@@ -82,7 +82,7 @@ Create a symbolic links from the two configuration files generated in the previo
         Replace ``<TETHYS_HOME>`` with the path to the Tethys home directory as noted in :ref:`production_portal_config` section.
 
 5. Modify :file:`supervisord.conf` (Rocky Linux Only)
-================================================
+=====================================================
 
 For Rocky Linux systems, modify :file:`supervisord.conf` to recognize our configuration files:
 

docs/tethys_sdk/jobs.rst

@@ -228,9 +228,9 @@ For example, a URL may look something like this:
 
     http://example.com/update-job-status/27/
 
-The output would look something like this:
+The response would look something like this:
 
-.. code-block:: python
+.. code-block:: javascript
 
     {"success": true}
 
@@ -241,6 +241,26 @@ This URL can be retrieved from the job manager with the ``get_job_status_callbac
     job_manager = App.get_job_manager()
     callback_url = job_manager.get_job_status_callback_url(request, job_id)
 
+The callback URL can be used to update the jobs status after a specified delay by passing the ``delay`` query parameter:
+
+.. code-block::
+
+    http://<host>/update-job-status/<job_id>/?delay=<delay_in_seconds>
+
+For example, to schedule a job update in 30 seconds:
+
+.. code-block::
+
+    http://<host>/update-job-status/27/?delay=30
+
+In this case the response would look like this:
+
+.. code-block:: javascript
+
+    {"success": "scheduled"}
+
+This delay can be useful so the job itself can hit the endpoint just before completing to trigger the Tethys Portal to check its status after it has time to complete and exit. This will allow the portal to register that the job has completed and start any data transfer that is triggered upon job completion.
+
 Custom Statuses
 ---------------
 Custom statuses can be given to jobs simply by assigning the ``status`` attribute:

tests/unit_tests/test_tethys_apps/test_views.py

@@ -7,8 +7,6 @@
     handoff_capabilities,
     handoff,
     send_beta_feedback_email,
-    update_job_status,
-    update_dask_job_status,
 )
 
 
@@ -291,55 +289,3 @@ def test_send_beta_feedback_email_send_mail_exception(
         mock_json_response.assert_called_once_with(
             {"success": False, "error": "Failed to send email: foo_error"}
         )
-
-    @mock.patch("tethys_apps.views.JsonResponse")
-    @mock.patch("tethys_apps.views.TethysJob")
-    def test_update_job_status(self, mock_tethysjob, mock_json_response):
-        mock_request = mock.MagicMock()
-        mock_job_id = mock.MagicMock()
-        mock_job1 = mock.MagicMock()
-        mock_job1.status = True
-        mock_tethysjob.objects.get_subclass.return_value = mock_job1
-
-        update_job_status(mock_request, mock_job_id)
-        mock_tethysjob.objects.get_subclass.assert_called_once_with(id=mock_job_id)
-        mock_json_response.assert_called_once_with({"success": True})
-
-    @mock.patch("tethys_apps.views.JsonResponse")
-    @mock.patch("tethys_apps.views.TethysJob")
-    def test_update_job_statusException(self, mock_tethysjob, mock_json_response):
-        mock_request = mock.MagicMock()
-        mock_job_id = mock.MagicMock()
-        mock_tethysjob.objects.get_subclass.side_effect = Exception
-
-        update_job_status(mock_request, mock_job_id)
-        mock_tethysjob.objects.get_subclass.assert_called_once_with(id=mock_job_id)
-        mock_json_response.assert_called_once_with({"success": False})
-
-    @mock.patch("tethys_apps.views.JsonResponse")
-    @mock.patch("tethys_apps.views.DaskJob")
-    def test_update_dask_job_status(self, mock_daskjob, mock_json_response):
-        mock_request = mock.MagicMock()
-        mock_job_key = mock.MagicMock()
-        mock_job1 = mock.MagicMock()
-        mock_job1.status = True
-        mock_job2 = mock.MagicMock()
-        mock_daskjob.objects.filter.return_value = [mock_job1, mock_job2]
-
-        # Call the method
-        update_dask_job_status(mock_request, mock_job_key)
-
-        # check results
-        mock_daskjob.objects.filter.assert_called_once_with(key=mock_job_key)
-        mock_json_response.assert_called_once_with({"success": True})
-
-    @mock.patch("tethys_apps.views.JsonResponse")
-    @mock.patch("tethys_apps.views.DaskJob")
-    def test_update_dask_job_statusException(self, mock_daskjob, mock_json_response):
-        mock_request = mock.MagicMock()
-        mock_job_key = mock.MagicMock()
-        mock_daskjob.objects.filter.side_effect = Exception
-
-        update_dask_job_status(mock_request, mock_job_key)
-        mock_daskjob.objects.filter.assert_called_once_with(key=mock_job_key)
-        mock_json_response.assert_called_once_with({"success": False})

tests/unit_tests/test_tethys_compute/test_tasks.py

@@ -0,0 +1,55 @@
+import unittest
+from unittest import mock
+
+import tethys_compute.tasks as tethys_compute_tasks
+
+
+async def noop():
+    pass
+
+
+def raise_error():
+    raise Exception()
+
+
+class TestTasks(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    @mock.patch("tethys_compute.tasks._run_after_delay", new_callable=mock.MagicMock)
+    @mock.patch("tethys_compute.tasks.asyncio.create_task")
+    def test_create_task(self, mock_aio_ct, mock_run_delay):
+        mock_func = mock.MagicMock()
+        mock_coro = mock.MagicMock()
+        mock_run_delay.return_value = mock_coro
+        tethys_compute_tasks.create_task(mock_func)
+        mock_aio_ct.assert_called_with(mock_coro)
+        mock_run_delay.assert_called_with(
+            mock_func, delay=0, periodic=False, count=None
+        )
+
+    @mock.patch("tethys_compute.tasks.logger")
+    async def test_run_after_delay(self, mock_log):
+        await tethys_compute_tasks._run_after_delay(
+            noop, delay=0, periodic=False, count=None
+        )
+        mock_log.info.assert_called()
+
+    @mock.patch("tethys_compute.tasks.logger")
+    @mock.patch("tethys_compute.tasks.asyncio.sleep")
+    async def test_run_after_delay_periodic(self, mock_sleep, mock_log):
+        await tethys_compute_tasks._run_after_delay(
+            noop, delay=30, periodic=True, count=2
+        )
+        mock_sleep.assert_called_with(30)
+        mock_log.info.assert_called()
+
+    @mock.patch("tethys_compute.tasks.logger")
+    async def test_run_after_delay_exception(self, mock_log):
+        await tethys_compute_tasks._run_after_delay(
+            raise_error, delay=0, periodic=False, count=None
+        )
+        self.assertEqual(mock_log.info.call_count, 2)

tests/unit_tests/test_tethys_compute/test_views/test_update_status.py

@@ -0,0 +1,120 @@
+import unittest
+from unittest import mock
+
+import tethys_compute.views.update_status as tethys_compute_update_status
+
+
+class TestUpdateStatus(unittest.IsolatedAsyncioTestCase):
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    @mock.patch("tethys_compute.views.update_status.TethysJob.objects.get_subclass")
+    async def test_get_job(self, mock_tj):
+        mock_user = mock.MagicMock(is_staff=False)
+        mock_user.has_perm.return_value = False
+        await tethys_compute_update_status.get_job("job_id", mock_user)
+        mock_tj.assert_called_with(id="job_id", user=mock_user)
+
+    @mock.patch("tethys_compute.views.update_status.TethysJob.objects.get_subclass")
+    async def test_get_job_staff(self, mock_tj):
+        mock_user = mock.MagicMock(is_staff=True)
+        await tethys_compute_update_status.get_job("job_id", mock_user)
+        mock_tj.assert_called_with(id="job_id")
+
+    @mock.patch("tethys_compute.views.update_status.TethysJob.objects.get_subclass")
+    async def test_get_job_has_permission(self, mock_tj):
+        mock_user = mock.MagicMock(is_staff=False)
+        mock_user.has_perm.return_value = True
+        await tethys_compute_update_status.get_job("job_id", mock_user)
+        mock_tj.assert_called_with(id="job_id")
+
+    @mock.patch("tethys_compute.views.update_status.logger")
+    @mock.patch("tethys_compute.views.update_status.JsonResponse")
+    @mock.patch("tethys_compute.views.update_status.TethysJob")
+    async def test_update_job_status(self, mock_tethysjob, mock_json_response, _):
+        mock_request = mock.MagicMock(GET={})
+        mock_job_id = mock.MagicMock()
+        mock_job1 = mock.MagicMock()
+        mock_job1.status = True
+        mock_tethysjob.objects.get_subclass.return_value = mock_job1
+
+        await tethys_compute_update_status.update_job_status(mock_request, mock_job_id)
+        mock_tethysjob.objects.get_subclass.assert_called_once_with(id=mock_job_id)
+        mock_json_response.assert_called_once_with({"success": True})
+
+    @mock.patch("tethys_compute.views.update_status.create_task")
+    @mock.patch("tethys_compute.views.update_status.logger")
+    @mock.patch("tethys_compute.views.update_status.JsonResponse")
+    async def test_update_job_status_with_delay(
+        self, mock_json_response, mock_log, mock_ct
+    ):
+        mock_request = mock.MagicMock(GET={"delay": "1"})
+        mock_job_id = mock.MagicMock()
+
+        await tethys_compute_update_status.update_job_status(mock_request, mock_job_id)
+        mock_json_response.assert_called_once_with({"success": "scheduled"})
+        mock_log.debug.assert_called_once()
+        mock_ct.assert_called_with(
+            tethys_compute_update_status._update_job_status, mock_job_id, delay=1
+        )
+
+    @mock.patch("tethys_compute.views.update_status.create_task")
+    @mock.patch("tethys_compute.views.update_status.logger")
+    @mock.patch("tethys_compute.views.update_status.JsonResponse")
+    async def test_update_job_status_with_delay_exception(
+        self, mock_json_response, mock_log, mock_ct
+    ):
+        mock_request = mock.MagicMock(GET={"delay": "1"})
+        mock_job_id = mock.MagicMock()
+        mock_ct.side_effect = Exception
+
+        await tethys_compute_update_status.update_job_status(mock_request, mock_job_id)
+        mock_json_response.assert_called_once_with({"success": False})
+        mock_log.warning.assert_called_once()
+
+    @mock.patch("tethys_compute.views.update_status.logger")
+    @mock.patch("tethys_compute.views.update_status.JsonResponse")
+    @mock.patch("tethys_compute.views.update_status.TethysJob")
+    async def test_update_job_statusException(
+        self, mock_tethysjob, mock_json_response, mock_log
+    ):
+        mock_request = mock.MagicMock(GET={})
+        mock_job_id = mock.MagicMock()
+        mock_tethysjob.objects.get_subclass.side_effect = Exception
+
+        await tethys_compute_update_status.update_job_status(mock_request, mock_job_id)
+        mock_tethysjob.objects.get_subclass.assert_called_once_with(id=mock_job_id)
+        mock_json_response.assert_called_once_with({"success": False})
+        mock_log.warning.assert_called_once()
+
+    @mock.patch("tethys_compute.views.update_status.JsonResponse")
+    @mock.patch("tethys_compute.views.update_status.DaskJob")
+    def test_update_dask_job_status(self, mock_daskjob, mock_json_response):
+        mock_request = mock.MagicMock()
+        mock_job_key = mock.MagicMock()
+        mock_job1 = mock.MagicMock()
+        mock_job1.status = True
+        mock_job2 = mock.MagicMock()
+        mock_daskjob.objects.filter.return_value = [mock_job1, mock_job2]
+
+        # Call the method
+        tethys_compute_update_status.update_dask_job_status(mock_request, mock_job_key)
+
+        # check results
+        mock_daskjob.objects.filter.assert_called_once_with(key=mock_job_key)
+        mock_json_response.assert_called_once_with({"success": True})
+
+    @mock.patch("tethys_compute.views.update_status.JsonResponse")
+    @mock.patch("tethys_compute.views.update_status.DaskJob")
+    def test_update_dask_job_statusException(self, mock_daskjob, mock_json_response):
+        mock_request = mock.MagicMock()
+        mock_job_key = mock.MagicMock()
+        mock_daskjob.objects.filter.side_effect = Exception
+
+        tethys_compute_update_status.update_dask_job_status(mock_request, mock_job_key)
+        mock_daskjob.objects.filter.assert_called_once_with(key=mock_job_key)
+        mock_json_response.assert_called_once_with({"success": False})