Add purge_history abstraction and wait for purge to complete if async

lib/galaxy_test/api/test_datasets.py

@@ -896,7 +896,7 @@ class TestDatasetsApi(ApiTestCase):
         self.dataset_populator.wait_for_history_jobs(history_id)
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         # now we can't update the datatype
         response = self._put(f"histories/{history_id}/contents/{hda_id}", data={"datatype": "tabular"}, json=True)

lib/galaxy_test/api/test_histories.py

@@ -601,7 +601,7 @@ class TestHistoriesApi(ApiTestCase, BaseHistories):
         assert show_response["name"] == "Immutable Name"
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         # we cannot update the name anymore
         response = self._update(history_id, {"name": "New Name"})
@@ -617,7 +617,7 @@ class TestHistoriesApi(ApiTestCase, BaseHistories):
         self.dataset_populator.new_dataset(history_id, content="TestContents")
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         # we cannot add another dataset
         with self.assertRaisesRegex(AssertionError, "History is immutable"):
@@ -631,7 +631,7 @@ class TestHistoriesApi(ApiTestCase, BaseHistories):
         self._update(history_id, {"tags": ["FirstTag"]})
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         # we cannot add another tag
         response = self._update(history_id, {"tags": ["SecondTag"]})

lib/galaxy_test/api/test_history_contents.py

@@ -1052,7 +1052,7 @@ class TestHistoryContentsApi(ApiTestCase):
         }
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         with self.assertRaisesRegex(AssertionError, "History is immutable"):
             self.dataset_populator.run_tool("cat1", inputs=inputs, history_id=history_id)
@@ -1061,7 +1061,7 @@ class TestHistoryContentsApi(ApiTestCase):
         hdca = self._create_pair_collection(history_id)
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         body = dict(name="newnameforpair")
         update_response = self._put(
@@ -1074,7 +1074,7 @@ class TestHistoryContentsApi(ApiTestCase):
         hda1 = self._wait_for_new_hda(history_id)
 
         # once we purge the history, it becomes immutable
-        self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        self.dataset_populator.purge_history(history_id)
 
         update_response = self._update(history_id, hda1["id"], dict(name="Updated Name"))
         self._assert_status_code_is(update_response, 403)

lib/galaxy_test/api/test_workflows.py

@@ -9095,7 +9095,7 @@ outer_input:
     def test_cannot_run_workflow_on_immutable_history(self) -> None:
         with self.dataset_populator.test_history() as history_id:
             # once we purge the history, it becomes immutable
-            self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+            self.dataset_populator.purge_history(history_id)
 
             with self.assertRaisesRegex(AssertionError, "History is immutable"):
                 self.workflow_populator.run_workflow(

lib/galaxy_test/base/populators.py

@@ -867,6 +867,14 @@ class BaseDatasetPopulator(BasePopulator):
         delete_response = self._delete(f"histories/{history_id}")
         delete_response.raise_for_status()
 
+    def purge_history(self, history_id: str) -> dict:
+        purge_response = self._delete(f"histories/{history_id}", data={"purge": True}, json=True)
+        purge_response.raise_for_status()
+        result = purge_response.json()
+        if purge_task := result.get("purge_task"):
+            self.wait_on_task_id(purge_task["id"])
+        return result
+
     def delete_dataset(
         self,
         history_id: str,

test/integration/test_purge_datasets.py

@@ -89,27 +89,9 @@ class TestPurgeDatasetsIntegration(integration_util.IntegrationTestCase):
         assert self._file_exists_on_disk(dataset_file1)
         assert self._file_exists_on_disk(dataset_file2)
 
-        # Purge the entire history
-        purge_response = self._delete(f"histories/{self.test_history_id}", data={"purge": True}, json=True)
-        self._assert_status_code_is_ok(purge_response)
-        purge_result = purge_response.json()
-
-        # Verify history is purged
+        purge_result = self.dataset_populator.purge_history(self.test_history_id)
         assert purge_result["purged"]
         assert purge_result["deleted"]
-
-        # Verify the response contains a purge_task with an id
-        assert "purge_task" in purge_result
-        purge_task = purge_result["purge_task"]
-        assert "id" in purge_task
-        purge_task_id = purge_task["id"]
-
-        # Wait for the celery task to complete via the tasks API
-        self.dataset_populator.wait_on_task_id(purge_task_id)
-
-        # After task completion, HDAs should be purged and files deleted
-        self.dataset_populator.wait_for_purge(self.test_history_id, hda1_id)
-        self.dataset_populator.wait_for_purge(self.test_history_id, hda2_id)
         assert not self._file_exists_on_disk(dataset_file1)
         assert not self._file_exists_on_disk(dataset_file2)
 
@@ -123,16 +105,8 @@ class TestPurgeDatasetsIntegration(integration_util.IntegrationTestCase):
             dataset_file = self._get_underlying_dataset_on_disk(hda_id)
             assert self._file_exists_on_disk(dataset_file)
 
-        # Purge via admin API (anonymous histories are only accessible to admins)
-        purge_response = self._delete(f"histories/{history_id}", data={"purge": True}, admin=True, json=True)
-        self._assert_status_code_is_ok(purge_response)
-        purge_result = purge_response.json()
-        assert purge_result["purged"]
-
-        # Wait for the celery purge task to finish
-        assert "purge_task" in purge_result
-        task_ok = self.dataset_populator.wait_on_task_id(purge_result["purge_task"]["id"])
-        assert task_ok
+            purge_result = self.dataset_populator.purge_history(history_id)
+            assert purge_result["purged"]
 
         assert not self._file_exists_on_disk(dataset_file)