From ebd54d8c1bf15c96cd69a6e1533af4b3ca0ec9ca Mon Sep 17 00:00:00 2001 From: davelopez <46503462+davelopez@users.noreply.github.com> Date: Tue, 12 May 2026 19:31:17 +0200 Subject: [PATCH] Rename API and service methods to clarify bulk storage operations --- .../webapps/galaxy/api/history_contents.py | 16 ++++---- .../galaxy/services/history_contents.py | 6 +-- lib/galaxy_test/base/populators.py | 40 +++++++++++++------ .../test_bulk_storage_operations.py | 33 ++++++++------- 4 files changed, 58 insertions(+), 37 deletions(-) diff --git a/lib/galaxy/webapps/galaxy/api/history_contents.py b/lib/galaxy/webapps/galaxy/api/history_contents.py index 9a551f6302f..0ec7a7fba21 100644 --- a/lib/galaxy/webapps/galaxy/api/history_contents.py +++ b/lib/galaxy/webapps/galaxy/api/history_contents.py @@ -842,44 +842,44 @@ class FastAPIHistoryContents: "/api/histories/{history_id}/contents/bulk/storage/preview", summary="Previews a storage bulk operation for selected history contents.", ) - def storage_operation_preview( + def bulk_storage_operation_preview( self, history_id: HistoryIDPathParam, trans: ProvidesHistoryContext = DependsOnTrans, filter_query_params: ValueFilterQueryParams = Depends(get_value_filter_query_params), payload: StorageOperationPreviewRequest = Body(...), ) -> StorageOperationPreviewResponse: - return self.service.storage_operation_preview(trans, history_id, filter_query_params, payload) + return self.service.bulk_storage_operation_preview(trans, history_id, filter_query_params, payload) @router.post( "/api/histories/{history_id}/contents/bulk/storage/execute", summary="Executes a previously previewed storage bulk operation snapshot.", ) - def storage_operation_execute( + def bulk_storage_operation_execute( self, history_id: HistoryIDPathParam, trans: ProvidesHistoryContext = DependsOnTrans, payload: StorageOperationExecuteRequest = Body(...), ) -> StorageOperationExecuteResponse: - return self.service.storage_operation_execute(trans, history_id, payload) + return self.service.bulk_storage_operation_execute(trans, history_id, payload) @router.get( "/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}", summary="Returns run status summary for a storage bulk operation.", ) - def storage_operation_run( + def bulk_storage_operation_run( self, history_id: HistoryIDPathParam, run_id: DecodedDatabaseIdField, trans: ProvidesHistoryContext = DependsOnTrans, ) -> StorageOperationRunResponse: - return self.service.storage_operation_run(trans, history_id, run_id) + return self.service.bulk_storage_operation_run(trans, history_id, run_id) @router.get( "/api/histories/{history_id}/contents/bulk/storage/runs/{run_id}/items", summary="Returns paginated per-item details for a storage bulk operation run.", ) - def storage_operation_run_items( + def bulk_storage_operation_run_items( self, history_id: HistoryIDPathParam, run_id: DecodedDatabaseIdField, @@ -889,7 +889,7 @@ class FastAPIHistoryContents: limit: int = StorageRunLimitQueryParam, search: Optional[str] = StorageRunSearchQueryParam, ) -> list[StorageOperationRunItemStatus]: - run_items, total_matches = self.service.storage_operation_run_items( + run_items, total_matches = self.service.bulk_storage_operation_run_items( trans, history_id, run_id, diff --git a/lib/galaxy/webapps/galaxy/services/history_contents.py b/lib/galaxy/webapps/galaxy/services/history_contents.py index cdfe96c2d8a..37198da31a7 100644 --- a/lib/galaxy/webapps/galaxy/services/history_contents.py +++ b/lib/galaxy/webapps/galaxy/services/history_contents.py @@ -750,7 +750,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto success_count = len(contents) - len(errors) return HistoryContentBulkOperationResult(success_count=success_count, errors=errors) - def storage_operation_preview( + def bulk_storage_operation_preview( self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField, @@ -809,7 +809,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto trans.sa_session.commit() return StorageOperationExecuteResponse(run=self.storage_operation_manager.to_run_summary(run)) - def storage_operation_run( + def bulk_storage_operation_run( self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField, @@ -826,7 +826,7 @@ class HistoriesContentsService(ServiceBase, ServesExportStores, ConsumesModelSto summary = self.storage_operation_manager.to_run_summary(run) return StorageOperationRunResponse(run=summary) - def storage_operation_run_items( + def bulk_storage_operation_run_items( self, trans: ProvidesHistoryContext, history_id: DecodedDatabaseIdField, diff --git a/lib/galaxy_test/base/populators.py b/lib/galaxy_test/base/populators.py index 02ccd43f485..2952b097285 100644 --- a/lib/galaxy_test/base/populators.py +++ b/lib/galaxy_test/base/populators.py @@ -1504,7 +1504,9 @@ class BaseDatasetPopulator(BasePopulator): get_storage_response = self._get(storage_url) return get_storage_response - def storage_preview_raw(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> Response: + def bulk_storage_operation_preview_raw( + self, history_id: str, payload: dict[str, Any], expected_status: int = 200 + ) -> Response: response = self._post( f"histories/{history_id}/contents/bulk/storage/preview", data=payload, @@ -1513,10 +1515,14 @@ class BaseDatasetPopulator(BasePopulator): api_asserts.assert_status_code_is(response, expected_status) return response - def storage_preview(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> dict[str, Any]: - return self.storage_preview_raw(history_id, payload, expected_status=expected_status).json() + def bulk_storage_operation_preview( + self, history_id: str, payload: dict[str, Any], expected_status: int = 200 + ) -> dict[str, Any]: + return self.bulk_storage_operation_preview_raw(history_id, payload, expected_status=expected_status).json() - def storage_execute_raw(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> Response: + def bulk_storage_operation_execute_raw( + self, history_id: str, payload: dict[str, Any], expected_status: int = 200 + ) -> Response: response = self._post( f"histories/{history_id}/contents/bulk/storage/execute", data=payload, @@ -1525,15 +1531,25 @@ class BaseDatasetPopulator(BasePopulator): api_asserts.assert_status_code_is(response, expected_status) return response - def storage_execute(self, history_id: str, payload: dict[str, Any], expected_status: int = 200) -> dict[str, Any]: - return self.storage_execute_raw(history_id, payload, expected_status=expected_status).json() + def bulk_storage_operation_execute( + self, history_id: str, payload: dict[str, Any], expected_status: int = 200 + ) -> dict[str, Any]: + return self.bulk_storage_operation_execute_raw(history_id, payload, expected_status=expected_status).json() - def storage_run_status(self, history_id: str, run_id: str, expected_status: int = 200) -> dict[str, Any]: + def bulk_storage_operation_run_status_raw( + self, history_id: str, run_id: str, expected_status: int = 200 + ) -> Response: response = self._get(f"histories/{history_id}/contents/bulk/storage/runs/{run_id}") api_asserts.assert_status_code_is(response, expected_status) + return response + + def bulk_storage_operation_run_status( + self, history_id: str, run_id: str, expected_status: int = 200 + ) -> dict[str, Any]: + response = self.bulk_storage_operation_run_status_raw(history_id, run_id, expected_status=expected_status) return response.json() - def storage_run_items( + def bulk_storage_operation_run_items( self, history_id: str, run_id: str, @@ -1552,7 +1568,7 @@ class BaseDatasetPopulator(BasePopulator): api_asserts.assert_status_code_is(response, expected_status) return response.json() - def wait_for_storage_run( + def wait_for_bulk_storage_operation_run( self, history_id: str, run_id: str, @@ -1561,14 +1577,14 @@ class BaseDatasetPopulator(BasePopulator): timeout: timeout_type = DEFAULT_TIMEOUT, ) -> dict[str, Any]: def is_terminal(): - run_data = self.storage_run_status(history_id, run_id) + run_data = self.bulk_storage_operation_run_status(history_id, run_id) if run_data["run"]["state"] in ("completed", "failed"): if include_items_on_terminal: - run_data["items"] = self.storage_run_items(history_id, run_id, search=search) + run_data["items"] = self.bulk_storage_operation_run_items(history_id, run_id, search=search) return run_data return None - return wait_on(is_terminal, "storage run completion", timeout=timeout) + return wait_on(is_terminal, "bulk storage operation run completion", timeout=timeout) def get_roles(self) -> list: using_requirement("admin") diff --git a/test/integration/objectstore/test_bulk_storage_operations.py b/test/integration/objectstore/test_bulk_storage_operations.py index 1ec91879cb8..611b7bfafec 100644 --- a/test/integration/objectstore/test_bulk_storage_operations.py +++ b/test/integration/objectstore/test_bulk_storage_operations.py @@ -36,7 +36,10 @@ from typing import ( from unittest.mock import patch from uuid import uuid4 -from galaxy.celery.tasks import recover_stale_storage_operation_runs +from galaxy.celery.tasks import ( + prune_expired_storage_operations, + recover_stale_storage_operation_runs, +) from galaxy.managers.dataset_storage_operations import ( DatasetStorageOperationManager, StorageOperationRunExecutor, @@ -171,7 +174,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): items: list[dict[str, Any]], expected_status: int = 200, ) -> dict[str, Any]: - return self.dataset_populator.storage_preview( + return self.dataset_populator.bulk_storage_operation_preview( history_id, { "mode": "move", @@ -188,7 +191,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): skip_ineligible: bool = True, expected_status: int = 200, ) -> dict[str, Any]: - return self.dataset_populator.storage_execute( + return self.dataset_populator.bulk_storage_operation_execute( history_id, { "snapshot_id": snapshot_id, @@ -213,7 +216,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): skip_ineligible=skip_ineligible, ) run = execute_result["run"] - final = self.dataset_populator.wait_for_storage_run( + final = self.dataset_populator.wait_for_bulk_storage_operation_run( history_id, run["run_id"], include_items_on_terminal=include_items_on_terminal, @@ -222,7 +225,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): return preview, run, final def _run_items(self, history_id: str, run_id: str, search: str | None = None) -> list[dict[str, Any]]: - return self.dataset_populator.storage_run_items(history_id, run_id, search=search) + return self.dataset_populator.bulk_storage_operation_run_items(history_id, run_id, search=search) def _run_items_by_dataset_id(self, items: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: return {item["dataset_id"]: item for item in items} @@ -269,7 +272,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): else: executor.execute_run(snapshot) - return self._app.security.encode_id(run.id) + return self._encode_id(run.id) def _item(self, hda_id: str) -> dict[str, Any]: return {"id": hda_id, "history_content_type": "dataset"} @@ -746,7 +749,9 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): def test_get_run_unknown_id_returns_404(self): """GET on an unknown run_id returns HTTP 404.""" with self.dataset_populator.test_history() as history_id: - self.dataset_populator.storage_run_status(history_id, self._unknown_encoded_id(), expected_status=404) + self.dataset_populator.bulk_storage_operation_run_status( + history_id, self._unknown_encoded_id(), expected_status=404 + ) # ------------------------------------------------------------------ collection (HDCA) tests @@ -819,7 +824,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): skip_ineligible=False, ) run = execute_result["run"] - final = self.dataset_populator.wait_for_storage_run( + final = self.dataset_populator.wait_for_bulk_storage_operation_run( history_id, run["run_id"], include_items_on_terminal=True, @@ -857,7 +862,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): baseline_default_count, baseline_separate_count = self._store_file_counts() first_execute = self._execute_snapshot(history_id, preview["snapshot_id"], skip_ineligible=True) - first_run = self.dataset_populator.wait_for_storage_run( + first_run = self.dataset_populator.wait_for_bulk_storage_operation_run( history_id, first_execute["run"]["run_id"], include_items_on_terminal=True, @@ -889,7 +894,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): self._assert_dataset_store_and_content(history_id, control["id"], DEFAULT_OBJECT_STORE_ID, "control\n") second_execute = self._execute_snapshot(history_id, preview["snapshot_id"], skip_ineligible=True) - second_run = self.dataset_populator.wait_for_storage_run( + second_run = self.dataset_populator.wait_for_bulk_storage_operation_run( history_id, second_execute["run"]["run_id"], include_items_on_terminal=True, @@ -947,7 +952,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): skip_ineligible=False, force_checksum_mismatch=True, ) - first_run_status = self.dataset_populator.storage_run_status(history_id, first_run_id) + first_run_status = self.dataset_populator.bulk_storage_operation_run_status(history_id, first_run_id) self._assert_run_counts(first_run_status["run"], succeeded=0, failed=1, skipped=0) assert first_run_status["run"]["total_bytes_processed"] == 0 @@ -972,7 +977,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): # Re-run same snapshot without forced corruption; move should now succeed. second_run_id = self._execute_snapshot_sync(sa_session, snapshot, skip_ineligible=False) - second_run_status = self.dataset_populator.storage_run_status(history_id, second_run_id) + second_run_status = self.dataset_populator.bulk_storage_operation_run_status(history_id, second_run_id) self._assert_run_counts(second_run_status["run"], succeeded=1, failed=0, skipped=0) assert second_run_status["run"]["total_bytes_processed"] > 0 @@ -1099,7 +1104,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): assert result.state == StorageOperationRunState.completed encoded_run_id = self._app.security.encode_id(recovered_run.id) - final = self.dataset_populator.storage_run_status(history_id, encoded_run_id) + final = self.dataset_populator.bulk_storage_operation_run_status(history_id, encoded_run_id) self._assert_run_counts(final["run"], succeeded=2, failed=0, skipped=0) items = self._run_items(history_id, encoded_run_id) @@ -1213,7 +1218,7 @@ class TestBulkStorageOperationsIntegration(BaseObjectStoreIntegrationTestCase): assert result.state == StorageOperationRunState.completed encoded_run_id = self._app.security.encode_id(recovered_run.id) - final = self.dataset_populator.storage_run_status(history_id, encoded_run_id) + final = self.dataset_populator.bulk_storage_operation_run_status(history_id, encoded_run_id) self._assert_run_counts(final["run"], succeeded=1, failed=0, skipped=0) items = self._run_items(history_id, encoded_run_id)