Loading lib/galaxy/tool_util/unittest_utils/__init__.py +3 −3 Original line number Diff line number Diff line from galaxy.util.bunch import Bunch from unittest.mock import Mock def mock_trans(has_user=True, is_admin=False): """A mock ``trans`` object for exposing user info to toolbox filter unit tests.""" trans = Bunch(user_is_admin=is_admin) trans = Mock(user_is_admin=is_admin) if has_user: trans.user = Bunch(preferences={}) trans.user = Mock(preferences={}) else: trans.user = None return trans lib/galaxy/util/lazy_process.py +1 −0 Original line number Diff line number Diff line Loading @@ -30,6 +30,7 @@ class LazyProcess: with self.thread_lock: self.allow_process_request = False if self.running: assert self.process # tell type checker it can not be None if self.running self.process.terminate() time.sleep(.01) if self.running: Loading lib/galaxy/webapps/galaxy/api/extended_metadata.py +16 −23 Original line number Diff line number Diff line Loading @@ -2,13 +2,14 @@ API operations on annotations. """ import logging from typing import Generic, Optional, TypeVar from galaxy import ( managers, web model, web, ) from galaxy.webapps.base.controller import ( HTTPNotImplemented, UsesExtendedMetadataMixin, UsesLibraryMixinItems, UsesStoredWorkflowMixin Loading @@ -17,10 +18,16 @@ from . import BaseGalaxyAPIController, depends log = logging.getLogger(__name__) T = TypeVar('T') class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetadataMixin, UsesLibraryMixinItems, UsesStoredWorkflowMixin): @web.legacy_expose_api class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetadataMixin, UsesLibraryMixinItems, UsesStoredWorkflowMixin, Generic[T]): exmeta_item_id: str def _get_item_from_id(self, trans, idstr, check_writable=True) -> Optional[T]: ... @web.expose_api def index(self, trans, **kwd): idnum = kwd[self.exmeta_item_id] item = self._get_item_from_id(trans, idnum, check_writable=False) Loading @@ -29,7 +36,7 @@ class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetada if ex_meta is not None: return ex_meta.data @web.legacy_expose_api @web.expose_api def create(self, trans, payload, **kwd): idnum = kwd[self.exmeta_item_id] item = self._get_item_from_id(trans, idnum, check_writable=True) Loading @@ -41,26 +48,12 @@ class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetada ex_obj = self.create_extended_metadata(trans, payload) self.set_item_extended_metadata_obj(trans, item, ex_obj) @web.legacy_expose_api def delete(self, trans, **kwd): idnum = kwd[self.tagged_item_id] item = self._get_item_from_id(trans, idnum, check_writable=True) if item is not None: ex_obj = self.get_item_extended_metadata_obj(trans, item) if ex_obj is not None: self.unset_item_extended_metadata_obj(trans, item) self.delete_extended_metadata(trans, ex_obj) @web.legacy_expose_api def undelete(self, trans, **kwd): raise HTTPNotImplemented() class LibraryDatasetExtendMetadataController(BaseExtendedMetadataController): class LibraryDatasetExtendMetadataController(BaseExtendedMetadataController[model.LibraryDatasetDatasetAssociation]): controller_name = "library_dataset_extended_metadata" exmeta_item_id = "library_content_id" def _get_item_from_id(self, trans, idstr, check_writable=True): def _get_item_from_id(self, trans, idstr, check_writable=True) -> Optional[model.LibraryDatasetDatasetAssociation]: if check_writable: item = self.get_library_dataset_dataset_association(trans, idstr) if trans.app.security_agent.can_modify_library_item(trans.get_current_user_roles(), item): Loading @@ -72,12 +65,12 @@ class LibraryDatasetExtendMetadataController(BaseExtendedMetadataController): return None class HistoryDatasetExtendMetadataController(BaseExtendedMetadataController): class HistoryDatasetExtendMetadataController(BaseExtendedMetadataController[model.HistoryDatasetAssociation]): controller_name = "history_dataset_extended_metadata" exmeta_item_id = "history_content_id" hda_manager: managers.hdas.HDAManager = depends(managers.hdas.HDAManager) def _get_item_from_id(self, trans, idstr, check_writable=True): def _get_item_from_id(self, trans, idstr, check_writable=True) -> Optional[model.HistoryDatasetAssociation]: decoded_idstr = self.decode_id(idstr) if check_writable: return self.hda_manager.get_owned(decoded_idstr, trans.user, current_history=trans.history) Loading lib/galaxy/webapps/galaxy/api/library_datasets.py +3 −2 Original line number Diff line number Diff line Loading @@ -456,8 +456,9 @@ class LibraryDatasetsController(BaseGalaxyAPIController, UsesVisualizationMixin, path = os.path.join(full_dir, path) if unsafe_walk(path, allowlist=[full_dir] + trans.app.config.user_library_import_symlink_allowlist, username=username): # the path is a dir and contains files that symlink outside the user dir error = 'User attempted to import a path that resolves to a path outside of their import dir: %s -> %s', \ error = 'User attempted to import a path that resolves to a path outside of their import dir: %s -> %s' % ( path, os.path.realpath(path) ) if trans.app.config.user_library_import_check_permissions: error += ' or is not readable for them.' log.error(error) Loading lib/galaxy_test/driver/driver_util.py +1 −1 Original line number Diff line number Diff line Loading @@ -621,7 +621,7 @@ def setup_shed_tools_for_test(app, tmpdir, testing_migrated_tools, testing_insta app.toolbox = tools.ToolBox(tool_configs, app.config.tool_path, app) def build_galaxy_app(simple_kwargs): def build_galaxy_app(simple_kwargs) -> GalaxyUniverseApplication: """Build a Galaxy app object from a simple keyword arguments. Construct paste style complex dictionary and use load_app_properties so Loading Loading
lib/galaxy/tool_util/unittest_utils/__init__.py +3 −3 Original line number Diff line number Diff line from galaxy.util.bunch import Bunch from unittest.mock import Mock def mock_trans(has_user=True, is_admin=False): """A mock ``trans`` object for exposing user info to toolbox filter unit tests.""" trans = Bunch(user_is_admin=is_admin) trans = Mock(user_is_admin=is_admin) if has_user: trans.user = Bunch(preferences={}) trans.user = Mock(preferences={}) else: trans.user = None return trans
lib/galaxy/util/lazy_process.py +1 −0 Original line number Diff line number Diff line Loading @@ -30,6 +30,7 @@ class LazyProcess: with self.thread_lock: self.allow_process_request = False if self.running: assert self.process # tell type checker it can not be None if self.running self.process.terminate() time.sleep(.01) if self.running: Loading
lib/galaxy/webapps/galaxy/api/extended_metadata.py +16 −23 Original line number Diff line number Diff line Loading @@ -2,13 +2,14 @@ API operations on annotations. """ import logging from typing import Generic, Optional, TypeVar from galaxy import ( managers, web model, web, ) from galaxy.webapps.base.controller import ( HTTPNotImplemented, UsesExtendedMetadataMixin, UsesLibraryMixinItems, UsesStoredWorkflowMixin Loading @@ -17,10 +18,16 @@ from . import BaseGalaxyAPIController, depends log = logging.getLogger(__name__) T = TypeVar('T') class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetadataMixin, UsesLibraryMixinItems, UsesStoredWorkflowMixin): @web.legacy_expose_api class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetadataMixin, UsesLibraryMixinItems, UsesStoredWorkflowMixin, Generic[T]): exmeta_item_id: str def _get_item_from_id(self, trans, idstr, check_writable=True) -> Optional[T]: ... @web.expose_api def index(self, trans, **kwd): idnum = kwd[self.exmeta_item_id] item = self._get_item_from_id(trans, idnum, check_writable=False) Loading @@ -29,7 +36,7 @@ class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetada if ex_meta is not None: return ex_meta.data @web.legacy_expose_api @web.expose_api def create(self, trans, payload, **kwd): idnum = kwd[self.exmeta_item_id] item = self._get_item_from_id(trans, idnum, check_writable=True) Loading @@ -41,26 +48,12 @@ class BaseExtendedMetadataController(BaseGalaxyAPIController, UsesExtendedMetada ex_obj = self.create_extended_metadata(trans, payload) self.set_item_extended_metadata_obj(trans, item, ex_obj) @web.legacy_expose_api def delete(self, trans, **kwd): idnum = kwd[self.tagged_item_id] item = self._get_item_from_id(trans, idnum, check_writable=True) if item is not None: ex_obj = self.get_item_extended_metadata_obj(trans, item) if ex_obj is not None: self.unset_item_extended_metadata_obj(trans, item) self.delete_extended_metadata(trans, ex_obj) @web.legacy_expose_api def undelete(self, trans, **kwd): raise HTTPNotImplemented() class LibraryDatasetExtendMetadataController(BaseExtendedMetadataController): class LibraryDatasetExtendMetadataController(BaseExtendedMetadataController[model.LibraryDatasetDatasetAssociation]): controller_name = "library_dataset_extended_metadata" exmeta_item_id = "library_content_id" def _get_item_from_id(self, trans, idstr, check_writable=True): def _get_item_from_id(self, trans, idstr, check_writable=True) -> Optional[model.LibraryDatasetDatasetAssociation]: if check_writable: item = self.get_library_dataset_dataset_association(trans, idstr) if trans.app.security_agent.can_modify_library_item(trans.get_current_user_roles(), item): Loading @@ -72,12 +65,12 @@ class LibraryDatasetExtendMetadataController(BaseExtendedMetadataController): return None class HistoryDatasetExtendMetadataController(BaseExtendedMetadataController): class HistoryDatasetExtendMetadataController(BaseExtendedMetadataController[model.HistoryDatasetAssociation]): controller_name = "history_dataset_extended_metadata" exmeta_item_id = "history_content_id" hda_manager: managers.hdas.HDAManager = depends(managers.hdas.HDAManager) def _get_item_from_id(self, trans, idstr, check_writable=True): def _get_item_from_id(self, trans, idstr, check_writable=True) -> Optional[model.HistoryDatasetAssociation]: decoded_idstr = self.decode_id(idstr) if check_writable: return self.hda_manager.get_owned(decoded_idstr, trans.user, current_history=trans.history) Loading
lib/galaxy/webapps/galaxy/api/library_datasets.py +3 −2 Original line number Diff line number Diff line Loading @@ -456,8 +456,9 @@ class LibraryDatasetsController(BaseGalaxyAPIController, UsesVisualizationMixin, path = os.path.join(full_dir, path) if unsafe_walk(path, allowlist=[full_dir] + trans.app.config.user_library_import_symlink_allowlist, username=username): # the path is a dir and contains files that symlink outside the user dir error = 'User attempted to import a path that resolves to a path outside of their import dir: %s -> %s', \ error = 'User attempted to import a path that resolves to a path outside of their import dir: %s -> %s' % ( path, os.path.realpath(path) ) if trans.app.config.user_library_import_check_permissions: error += ' or is not readable for them.' log.error(error) Loading
lib/galaxy_test/driver/driver_util.py +1 −1 Original line number Diff line number Diff line Loading @@ -621,7 +621,7 @@ def setup_shed_tools_for_test(app, tmpdir, testing_migrated_tools, testing_insta app.toolbox = tools.ToolBox(tool_configs, app.config.tool_path, app) def build_galaxy_app(simple_kwargs): def build_galaxy_app(simple_kwargs) -> GalaxyUniverseApplication: """Build a Galaxy app object from a simple keyword arguments. Construct paste style complex dictionary and use load_app_properties so Loading