enh: allow to programmatically abort download jobs

Author: paulmueller

CHANGELOG

@@ -2,6 +2,7 @@
  - feat: add circle archiving in the BagIt (RFC 8493) format
  - fix: handle resource downloads whose target directory got deleted
  - fix: check for existence of persistent job list in upload
+ - enh: allow to programmatically abort download jobs
  - ref: move setup wizard to `wizard_init` submodule
 1.0.0
  - fix: do not attempt to updates non-existent eternal jobs on exit

dcoraid/download/job.py

@@ -3,6 +3,7 @@
 import logging
 import pathlib
 import shutil
+import threading
 import traceback
 
 import time
@@ -357,7 +358,7 @@ def set_state(self, state):
                 logger.error(f"{self.traceback}")
         self.state = state
 
-    def task_download_resource(self):
+    def task_download_resource(self, abort_event: threading.Event = None):
         """Start the download
 
         The progress of the download is monitored and written
@@ -434,6 +435,12 @@ def task_download_resource(self):
                                             # for condensed data.
                                             and not self.condensed):
                                         hasher.update(chunk)
+                                    if (abort_event is not None
+                                            and abort_event.is_set()):
+                                        logger.warning(f"Aborting job {self}")
+                                        self.set_state("abort")
+                                        return
+
                     self.sha256sum_dl = hasher.hexdigest()
                     self.end_time = time.perf_counter()
                     self.set_state("downloaded")

tests/test_download_job.py

@@ -1,5 +1,7 @@
 import shutil
 import tempfile
+import threading
+import time
 from unittest import mock
 import uuid
 
@@ -22,6 +24,37 @@ def test_initialize():
     assert dj.state == "init"
 
 
+def test_download_abort(tmp_path):
+    # leukocytes.rtdc
+    res_id = "f7fa778f-6abd-1b53-ae5f-9ce12601d6f8"
+    event_abort = threading.Event()
+    api = common.get_api()
+    dj = job.DownloadJob(api=api,
+                         resource_id=res_id,
+                         download_path=tmp_path,
+                         )
+    assert dj.state == "init"
+
+    thread = threading.Thread(target=lambda: dj.task_download_resource(event_abort))
+    thread.start()
+
+    for ii in range(100):
+        bytes_dl = dj.get_status()["bytes downloaded"]
+        if bytes_dl > 1024*1024:
+            break
+        time.sleep(.1)
+
+    event_abort.set()
+    thread.join(timeout=10)
+
+    assert dj.get_status()["state"] == "abort"
+
+    # There should be an incomplete temporary download file
+    assert not dj.path.exists()
+    assert dj.path_temp.exists()
+    assert dj.path_temp.stat().st_size < dj.get_status()["bytes total"] - 1
+
+
 def test_download_deleted_directory():
     api = common.get_api()
     td = tempfile.mkdtemp(prefix="test-download")