Loading lib/galaxy/managers/lddas.py +14 −0 Original line number Diff line number Diff line import logging from typing import ( Any, Optional, ) from galaxy import model from galaxy.managers import base as manager_base Loading Loading @@ -26,6 +30,16 @@ class LDDAManager(DatasetAssociationManager): trans, id, "LibraryDatasetDatasetAssociation", check_ownership=False, check_accessible=check_accessible ) def is_owner(self, item: model.Base, user: Optional[model.User], **kwargs: Any) -> bool: """ Return True if user owns the item. """ if not isinstance(item, model.LibraryDatasetDatasetAssociation): raise TypeError('"item" must be a LibraryDatasetDatasetAssociation.') if self.app.config.is_admin_user(user): return True return item.user == user def _set_permissions(self, trans, library_dataset, role_ids_dict): # Check Git history for an older broken implementation, but it was broken # and security related and had not test coverage so it was deleted. Loading lib/galaxy/webapps/galaxy/services/datasets.py +2 −1 Original line number Diff line number Diff line Loading @@ -735,6 +735,7 @@ class DatasetsService(ServiceBase, UsesVisualizationMixin): try: manager = self.dataset_manager_by_type[dataset.src] dataset_instance = manager.get_owned(dataset.id, trans.user) if hasattr(dataset_instance, "history"): manager.error_unless_mutable(dataset_instance.history) if dataset.src == DatasetSourceType.hda: self.hda_manager.error_if_uploading(dataset_instance) Loading lib/galaxy_test/api/test_datasets.py +19 −0 Original line number Diff line number Diff line Loading @@ -18,10 +18,12 @@ from galaxy_test.base.api_asserts import assert_has_keys from galaxy_test.base.decorators import ( requires_admin, requires_new_history, requires_new_library, ) from galaxy_test.base.populators import ( DatasetCollectionPopulator, DatasetPopulator, LibraryPopulator, skip_without_datatype, skip_without_tool, ) Loading @@ -46,6 +48,7 @@ class TestDatasetsApi(ApiTestCase): def setUp(self): super().setUp() self.dataset_populator = DatasetPopulator(self.galaxy_interactor) self.library_populator = LibraryPopulator(self.galaxy_interactor) self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor) def test_index(self): Loading Loading @@ -705,6 +708,22 @@ class TestDatasetsApi(ApiTestCase): for purged_source_id in expected_purged_source_ids: self.dataset_populator.wait_for_purge(history_id, purged_source_id["id"]) @requires_new_history @requires_new_library def test_delete_batch_lddas(self): # Create a library dataset ld = self.library_populator.new_library_dataset("lda_test_library") ldda_id = ld["ldda_id"] # Delete the library dataset using the delete_batch endpoint delete_payload = {"datasets": [{"id": ldda_id, "src": "ldda"}]} deleted_result = self._delete_batch_with_payload(delete_payload) assert deleted_result["success_count"] == 1 # Ensure the library dataset is deleted library_dataset = self.library_populator.show_ldda(ldda_id=ldda_id) assert library_dataset["deleted"] is True def test_delete_batch_error(self): num_datasets = 4 dataset_map: Dict[int, str] = {} Loading Loading
lib/galaxy/managers/lddas.py +14 −0 Original line number Diff line number Diff line import logging from typing import ( Any, Optional, ) from galaxy import model from galaxy.managers import base as manager_base Loading Loading @@ -26,6 +30,16 @@ class LDDAManager(DatasetAssociationManager): trans, id, "LibraryDatasetDatasetAssociation", check_ownership=False, check_accessible=check_accessible ) def is_owner(self, item: model.Base, user: Optional[model.User], **kwargs: Any) -> bool: """ Return True if user owns the item. """ if not isinstance(item, model.LibraryDatasetDatasetAssociation): raise TypeError('"item" must be a LibraryDatasetDatasetAssociation.') if self.app.config.is_admin_user(user): return True return item.user == user def _set_permissions(self, trans, library_dataset, role_ids_dict): # Check Git history for an older broken implementation, but it was broken # and security related and had not test coverage so it was deleted. Loading
lib/galaxy/webapps/galaxy/services/datasets.py +2 −1 Original line number Diff line number Diff line Loading @@ -735,6 +735,7 @@ class DatasetsService(ServiceBase, UsesVisualizationMixin): try: manager = self.dataset_manager_by_type[dataset.src] dataset_instance = manager.get_owned(dataset.id, trans.user) if hasattr(dataset_instance, "history"): manager.error_unless_mutable(dataset_instance.history) if dataset.src == DatasetSourceType.hda: self.hda_manager.error_if_uploading(dataset_instance) Loading
lib/galaxy_test/api/test_datasets.py +19 −0 Original line number Diff line number Diff line Loading @@ -18,10 +18,12 @@ from galaxy_test.base.api_asserts import assert_has_keys from galaxy_test.base.decorators import ( requires_admin, requires_new_history, requires_new_library, ) from galaxy_test.base.populators import ( DatasetCollectionPopulator, DatasetPopulator, LibraryPopulator, skip_without_datatype, skip_without_tool, ) Loading @@ -46,6 +48,7 @@ class TestDatasetsApi(ApiTestCase): def setUp(self): super().setUp() self.dataset_populator = DatasetPopulator(self.galaxy_interactor) self.library_populator = LibraryPopulator(self.galaxy_interactor) self.dataset_collection_populator = DatasetCollectionPopulator(self.galaxy_interactor) def test_index(self): Loading Loading @@ -705,6 +708,22 @@ class TestDatasetsApi(ApiTestCase): for purged_source_id in expected_purged_source_ids: self.dataset_populator.wait_for_purge(history_id, purged_source_id["id"]) @requires_new_history @requires_new_library def test_delete_batch_lddas(self): # Create a library dataset ld = self.library_populator.new_library_dataset("lda_test_library") ldda_id = ld["ldda_id"] # Delete the library dataset using the delete_batch endpoint delete_payload = {"datasets": [{"id": ldda_id, "src": "ldda"}]} deleted_result = self._delete_batch_with_payload(delete_payload) assert deleted_result["success_count"] == 1 # Ensure the library dataset is deleted library_dataset = self.library_populator.show_ldda(ldda_id=ldda_id) assert library_dataset["deleted"] is True def test_delete_batch_error(self): num_datasets = 4 dataset_map: Dict[int, str] = {} Loading