From 349ad2308443e6be9e56c48f32cf03d22c238045 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Sat, 11 Jul 2026 16:44:48 +0200 Subject: [PATCH 1/2] Serialize only file sources required by a job Every job currently serializes every configured and user-defined file source into JobIO. Resolving those configurations can read vault secrets and mint OAuth access tokens, so unrelated jobs can contact every OAuth provider available to a user. Derive the required URIs at dispatch from persisted tool inputs, action-owned request data, and deferred input dataset origins. Directory URI parameters are collected generically, while tool actions report URLs embedded in data-fetch requests, legacy upload paramfiles, and history-import archive parameters. Keep the legacy upload paramfile path stable across job preparation and retries. This avoids duplicating execution metadata in new private job parameters. Serialize only the highest-scoring configured source for each URI and only the user sources addressed by gxuserfiles UUIDs. Filtering happens before user source properties are resolved, preventing credential work for unrelated sources while preserving the complete configuration needed by external runners. Addresses #17811. --- lib/galaxy/files/__init__.py | 43 +++++- lib/galaxy/files/sources/elabftw.py | 7 - lib/galaxy/jobs/__init__.py | 44 ++++-- lib/galaxy/managers/file_source_instances.py | 35 ++++- lib/galaxy/model/__init__.py | 5 + lib/galaxy/tools/actions/__init__.py | 5 + lib/galaxy/tools/actions/history_imp_exp.py | 7 + lib/galaxy/tools/actions/upload.py | 17 +++ lib/galaxy/tools/data_fetch_utils.py | 12 ++ lib/galaxy/tools/parameters/__init__.py | 16 +++ test/unit/app/jobs/test_job_wrapper.py | 64 +++++++++ .../app/managers/test_user_file_sources.py | 135 +++++++++++++++++- .../app/tools/test_collect_directory_uris.py | 30 ++++ test/unit/app/tools/test_data_fetch_utils.py | 47 +++++- test/unit/app/tools/test_history_imp_exp.py | 20 +++ test/unit/app/tools/test_upload_actions.py | 24 ++++ test/unit/files/test_http.py | 12 ++ test/unit/files/test_posix.py | 31 ++++ 18 files changed, 523 insertions(+), 31 deletions(-) create mode 100644 test/unit/app/tools/test_collect_directory_uris.py create mode 100644 test/unit/app/tools/test_upload_actions.py diff --git a/lib/galaxy/files/__init__.py b/lib/galaxy/files/__init__.py index e25948ee0ca..fa95154274b 100644 --- a/lib/galaxy/files/__init__.py +++ b/lib/galaxy/files/__init__.py @@ -51,6 +51,8 @@ class ProvidesFileSourcesTransaction(Protocol): log = logging.getLogger(__name__) +USER_FILE_SOURCES_SCHEME = "gxuserfiles" + class FileSourcePath(NamedTuple): file_source: BaseFilesSource @@ -70,8 +72,7 @@ class UserDefinedFileSources(Protocol): """Entry-point for Galaxy to inject user-defined file sources. Supplied object of this class is used to write out concrete - description of file sources when serializing all file sources - available to a user. + descriptions of user file sources selected for serialization. """ def validate_uri_root(self, uri: str, user_context: "FileSourcesUserContext") -> None: @@ -87,6 +88,7 @@ class UserDefinedFileSources(Protocol): browsable_only: bool | None = False, include_kind: set[PluginKind] | None = None, exclude_kind: set[PluginKind] | None = None, + referenced_uris: set[str] | None = None, ) -> list[dict[str, Any]]: """Write out user file sources as list of config dictionaries.""" # config_dicts: List[FilesSourceProperties] = [] @@ -111,6 +113,7 @@ class NullUserDefinedFileSources(UserDefinedFileSources): browsable_only: bool | None = False, include_kind: set[PluginKind] | None = None, exclude_kind: set[PluginKind] | None = None, + referenced_uris: set[str] | None = None, ) -> list[dict[str, Any]]: return [] @@ -209,6 +212,15 @@ class ConfiguredFileSources: def _parse_plugin_source(self, plugin_source: PluginConfigSource): return self._plugin_loader.load_plugins(plugin_source, self._file_sources_config) + @staticmethod + def _best_score(scores: list[FileSourceScore]) -> FileSourceScore | None: + best = max(scores, key=lambda candidate: candidate.score, default=None) + return best if best is not None and best.score > 0 else None + + def _best_configured_match(self, url: str) -> FileSourceScore | None: + scores = [FileSourceScore(file_source, file_source.score_url_match(url)) for file_source in self._file_sources] + return self._best_score(scores) + def find_best_match(self, url: str) -> BaseFilesSource | None: """Returns the best matching file source for handling a particular url. Each filesource scores its own ability to match a particular url, and the highest scorer with a score > 0 is selected.""" @@ -216,8 +228,8 @@ class ConfiguredFileSources: user_best_score = self._user_defined_file_sources.find_best_match(url) if user_best_score is not None: scores.append(user_best_score) - scores.sort(key=lambda f: f.score, reverse=True) - return next((fsscore.file_source for fsscore in scores if fsscore.score > 0), None) + best = self._best_score(scores) + return best.file_source if best is not None else None def get_file_source_path(self, uri): """Parse uri into a FileSource object and a path relative to its base.""" @@ -280,8 +292,14 @@ class ConfiguredFileSources: browsable_only: bool | None = False, include_kind: set[PluginKind] | None = None, exclude_kind: set[PluginKind] | None = None, + referenced_uris: set[str] | None = None, ) -> list[dict[str, Any]]: rval: list[dict[str, Any]] = [] + referenced_file_sources = None + if referenced_uris is not None: + referenced_file_sources = [ + match.file_source for uri in referenced_uris if (match := self._best_configured_match(uri)) is not None + ] for file_source in self._file_sources: if not file_source.user_has_access(user_context): continue @@ -291,6 +309,9 @@ class ConfiguredFileSources: continue if browsable_only and not file_source.get_browsable(): continue + # Skip sources that are not the best match for any URI required by this job. + if referenced_file_sources is not None and file_source not in referenced_file_sources: + continue el = file_source.to_dict(for_serialization=for_serialization, user_context=user_context) rval.append(el) if user_context: @@ -301,13 +322,23 @@ class ConfiguredFileSources: browsable_only=browsable_only, include_kind=include_kind, exclude_kind=exclude_kind, + referenced_uris=referenced_uris, ) ) return rval - def to_dict(self, for_serialization: bool = False, user_context: "OptionalUserContext" = None) -> dict[str, Any]: + def to_dict( + self, + for_serialization: bool = False, + user_context: "OptionalUserContext" = None, + referenced_uris: set[str] | None = None, + ) -> dict[str, Any]: return { - "file_sources": self.plugins_to_dict(for_serialization=for_serialization, user_context=user_context), + "file_sources": self.plugins_to_dict( + for_serialization=for_serialization, + user_context=user_context, + referenced_uris=referenced_uris, + ), "config": self._file_sources_config.to_dict(), } diff --git a/lib/galaxy/files/sources/elabftw.py b/lib/galaxy/files/sources/elabftw.py index 21d70175540..092f0f8052c 100644 --- a/lib/galaxy/files/sources/elabftw.py +++ b/lib/galaxy/files/sources/elabftw.py @@ -178,16 +178,9 @@ class eLabFTWFilesSource(BaseFilesSource[eLabFTWFileSourceTemplateConfiguration, def get_prefix(self) -> str | None: endpoint: ParseResult = self._get_endpoint() return self.id if self.scheme not in {"elabftw", DEFAULT_SCHEME} else (endpoint.netloc or None) - # it would make better sense to return - # `self.id if self.scheme == USER_FILE_SOURCES_SCHEME else (endpoint.netloc or None)`, where - # `USER_FILE_SOURCES_SCHEME` comes from `galaxy.managers.file_source_instances`; however, that would lead to a - # circular import (maybe `USER_FILE_SOURCES_SCHEME` should be moved to a module in a layer deeper than - # `galaxy.managers`) def get_scheme(self) -> str: return self.scheme if self.scheme and self.scheme != DEFAULT_SCHEME else "elabftw" - # it would make better sense to return `self.scheme if self.scheme == USER_FILE_SOURCES_SCHEME else "elabftw"`, - # but the same circular import issue as above arises def score_url_match(self, url: str) -> int: parsed_url = urlparse(url) diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index fb4a9524ea5..81269b4e55d 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -106,7 +106,10 @@ from galaxy.tools.evaluation import ( ToolEvaluator, UserToolEvaluator, ) -from galaxy.tools.parameters import params_to_json_internal +from galaxy.tools.parameters import ( + collect_directory_uris, + params_to_json_internal, +) from galaxy.util import ( parse_xml_string, RWXRWXRWX, @@ -1081,6 +1084,19 @@ class MinimalJobWrapper(HasResourceParameters): if authnz_manager and trans.user: authnz_manager.refresh_expiring_oidc_tokens(trans, trans.user) + def _referenced_file_source_uris(self, job: Job) -> set[str]: + """Return the file source URIs needed to execute this job.""" + uris: set[str] = set() + if self.tool is not None: + param_dict = self.get_param_dict(job) + uris.update(collect_directory_uris(self.tool.inputs, param_dict)) + uris.update(self.tool.tool_action.iter_referenced_file_source_uris(param_dict)) + for input_association in job.input_datasets + job.input_library_datasets: + dataset = input_association.dataset + if dataset is not None and dataset.has_deferred_data and dataset.dataset is not None: + uris.update(dataset.dataset.source_uris) + return uris + @property def job_io(self) -> JobIO: if self._job_io is None: @@ -1088,6 +1104,7 @@ class MinimalJobWrapper(HasResourceParameters): work_request = WorkRequestContext(self.app, user=job.user, galaxy_session=job.galaxy_session) user_context = ProvidesFileSourcesUserContext(work_request) self._refresh_oidc_tokens_for_job(work_request) + referenced_uris = self._referenced_file_source_uris(job) tool_source = self.tool.tool_source.to_string() if self.tool else None tool_dir = self.tool.tool_dir if self.tool else None self._job_io = JobIO( @@ -1106,7 +1123,11 @@ class MinimalJobWrapper(HasResourceParameters): new_file_path=self.app.config.new_file_path, builds_file_path=self.app.config.builds_file_path, len_file_path=self.app.config.len_file_path, - file_sources_dict=self.app.file_sources.to_dict(for_serialization=True, user_context=user_context), + file_sources_dict=self.app.file_sources.to_dict( + for_serialization=True, + user_context=user_context, + referenced_uris=referenced_uris, + ), user_context=user_context, check_job_script_integrity=self.app.config.check_job_script_integrity, check_job_script_integrity_count=self.app.config.check_job_script_integrity_count, @@ -1240,15 +1261,18 @@ class MinimalJobWrapper(HasResourceParameters): return os.path.abspath(os.path.join(self.working_directory, "outputs", COMMAND_VERSION_FILENAME)) def __prepare_upload_paramfile(self, job): - """Special case paramfile handling for the upload tool. Copies the paramfile to the working directory""" + """Copy the upload paramfile into the working directory and use the stable path.""" new = os.path.join(self.working_directory, "upload_params.json") - param_file_path = json.loads(next(iter(param.value for param in job.parameters if param.name == "paramfile"))) - try: - shutil.copy2(param_file_path, new) - except OSError as exc: - # It won't exist at the old path if setup was interrupted and tried again later - if exc.errno != errno.ENOENT or not os.path.exists(new): - raise + paramfile_parameter = next(iter(param for param in job.parameters if param.name == "paramfile")) + param_file_path = json.loads(paramfile_parameter.value) + if param_file_path != new: + try: + shutil.copy2(param_file_path, new) + except OSError as exc: + # It won't exist at the old path if setup was interrupted and tried again later + if exc.errno != errno.ENOENT or not os.path.exists(new): + raise + paramfile_parameter.value = json.dumps(new) def prepare(self, compute_environment=None): """ diff --git a/lib/galaxy/managers/file_source_instances.py b/lib/galaxy/managers/file_source_instances.py index d666b9f4151..a32663683b9 100644 --- a/lib/galaxy/managers/file_source_instances.py +++ b/lib/galaxy/managers/file_source_instances.py @@ -4,6 +4,7 @@ from typing import ( cast, Literal, ) +from urllib.parse import urlsplit from uuid import uuid4 from pydantic import ( @@ -26,6 +27,7 @@ from galaxy.files import ( FileSourceScore, FileSourcesUserContext, ProvidesFileSourcesUserContext, + USER_FILE_SOURCES_SCHEME, UserDefinedFileSources, ) from galaxy.files.plugins import ( @@ -53,6 +55,7 @@ from galaxy.files.templates.capabilities import ( ) from galaxy.managers.context import ProvidesUserContext from galaxy.model import ( + get_uuid, User, UserFileSource, ) @@ -111,7 +114,21 @@ from ._config_templates import ( log = logging.getLogger(__name__) -USER_FILE_SOURCES_SCHEME = "gxuserfiles" + +def referenced_user_file_source_ids(referenced_uris: set[str]) -> set[str]: + """Return canonical UUID hex strings addressed by ``gxuserfiles`` URIs.""" + ids: set[str] = set() + for uri in referenced_uris: + if not uri.startswith(f"{USER_FILE_SOURCES_SCHEME}://"): + continue + try: + split = urlsplit(uri) + if not split.netloc: + raise ValueError("URI has no authority") + ids.add(get_uuid(split.netloc).hex) + except ValueError as exc: + raise RequestParameterInvalidException(f"Invalid user file source URI [{uri}]") from exc + return ids class UserFileSourceModel(BaseModel): @@ -675,15 +692,24 @@ class UserDefinedFileSourcesImpl(UserDefinedFileSources): )[0] return file_source - def _all_user_file_source_properties(self, user_context: FileSourcesUserContext) -> list[dict[str, Any]]: + def _all_user_file_source_properties( + self, + user_context: FileSourcesUserContext, + referenced_uris: set[str] | None = None, + ) -> list[dict[str, Any]]: username_filter = User.__table__.c.username == user_context.username user: User | None = self._sa_session.query(User).filter(username_filter).one_or_none() if user is None: return [] + referenced_ids = None if referenced_uris is None else referenced_user_file_source_ids(referenced_uris) all_file_source_properties: list[dict[str, Any]] = [] for user_file_source in user.file_sources: if user_file_source.hidden: continue + # Filter before resolving properties because resolution can access the vault or mint + # an OAuth access token. + if referenced_ids is not None and get_uuid(user_file_source.uuid).hex not in referenced_ids: + continue try: files_source_properties = self._file_source_properties(user_file_source) except ValidationError: @@ -736,13 +762,16 @@ class UserDefinedFileSourcesImpl(UserDefinedFileSources): browsable_only: bool | None = False, include_kind: set[PluginKind] | None = None, exclude_kind: set[PluginKind] | None = None, + referenced_uris: set[str] | None = None, ) -> list[dict[str, Any]]: """Write out user file sources as list of config dictionaries.""" if user_context.anonymous: return [] as_dicts = [] - for files_source_properties in self._all_user_file_source_properties(user_context): + for files_source_properties in self._all_user_file_source_properties( + user_context, referenced_uris=referenced_uris + ): files_source_type = files_source_properties["type"] plugin_type_class = self._plugin_loader.get_plugin_type_class(files_source_type) plugin_kind = plugin_type_class.plugin_kind diff --git a/lib/galaxy/model/__init__.py b/lib/galaxy/model/__init__.py index f855d1f165a..dfd53d87d16 100644 --- a/lib/galaxy/model/__init__.py +++ b/lib/galaxy/model/__init__.py @@ -4858,6 +4858,11 @@ class Dataset(Base, StorableObject, Serializable): def is_new(self): return self.state == self.states.NEW + @property + def source_uris(self) -> list[str]: + """The URIs this dataset was populated from (e.g. remote/deferred sources).""" + return [source.source_uri for source in self.sources if source.source_uri] + def in_ready_state(self): return self.state in self.ready_states diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py index 65ee7d743aa..49e0b44008a 100644 --- a/lib/galaxy/tools/actions/__init__.py +++ b/lib/galaxy/tools/actions/__init__.py @@ -4,6 +4,7 @@ import os import re from abc import abstractmethod from collections.abc import ( + Iterable, Mapping, MutableMapping, ) @@ -99,6 +100,10 @@ class ToolAction: produces_real_jobs: bool + def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: + """Yield file source URIs embedded in action-specific parameters.""" + return () + @abstractmethod def execute( self, diff --git a/lib/galaxy/tools/actions/history_imp_exp.py b/lib/galaxy/tools/actions/history_imp_exp.py index 9b7ea531f27..6561a7d88f1 100644 --- a/lib/galaxy/tools/actions/history_imp_exp.py +++ b/lib/galaxy/tools/actions/history_imp_exp.py @@ -2,6 +2,7 @@ import datetime import logging import os import tempfile +from collections.abc import Iterable from galaxy.job_execution.setup import JobWorkingDirectory from galaxy.model import ( @@ -41,6 +42,12 @@ class ImportHistoryToolAction(ToolAction): produces_real_jobs: bool = True + def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: + if param_dict.get("__ARCHIVE_TYPE__") == "url": + archive_source = param_dict.get("__ARCHIVE_SOURCE__") + if isinstance(archive_source, str) and archive_source: + yield archive_source + def execute( self, tool, diff --git a/lib/galaxy/tools/actions/upload.py b/lib/galaxy/tools/actions/upload.py index bfedfd06819..8d1da22e579 100644 --- a/lib/galaxy/tools/actions/upload.py +++ b/lib/galaxy/tools/actions/upload.py @@ -1,6 +1,7 @@ import json import logging import os +from collections.abc import Iterable from galaxy.exceptions import RequestParameterMissingException from galaxy.job_execution.output_collect import copy_collection_metadata_from_target_dict @@ -16,6 +17,7 @@ from galaxy.model.dataset_collections.structure import UninitializedTree from galaxy.schema.credentials import CredentialsContext from galaxy.tools._types import ToolStateJobInstancePopulatedT from galaxy.tools.actions import upload_common +from galaxy.tools.data_fetch_utils import iter_fetch_request_urls from galaxy.tools.execute import ( DatasetCollectionElementsSliceT, DEFAULT_DATASET_COLLECTION_ELEMENTS, @@ -86,6 +88,18 @@ class BaseUploadToolAction(ToolAction): class UploadToolAction(BaseUploadToolAction): + def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: + paramfile = param_dict.get("paramfile") + if not isinstance(paramfile, str): + return + with open(paramfile) as f: + upload_params = json.load(f) + for upload_param in upload_params: + if upload_param.get("type") == "url": + path = upload_param.get("path") + if isinstance(path, str) and path: + yield path + def _setup_job( self, tool, trans: ProvidesHistoryContext, incoming, dataset_upload_inputs, history, preferred_object_store_id ): @@ -112,6 +126,9 @@ class UploadToolAction(BaseUploadToolAction): class FetchUploadToolAction(BaseUploadToolAction): + def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: + return iter_fetch_request_urls(param_dict) + def _setup_job( self, tool, trans: ProvidesHistoryContext, incoming, dataset_upload_inputs, history, preferred_object_store_id ): diff --git a/lib/galaxy/tools/data_fetch_utils.py b/lib/galaxy/tools/data_fetch_utils.py index 2b63e20680a..ce93ac07258 100644 --- a/lib/galaxy/tools/data_fetch_utils.py +++ b/lib/galaxy/tools/data_fetch_utils.py @@ -1,3 +1,5 @@ +import json +from collections.abc import Iterator from datetime import ( datetime, timezone, @@ -6,6 +8,7 @@ from typing import Any from galaxy.authnz.psa_authnz import locate_token_expiration from galaxy.model import User +from galaxy.tools._types import ToolStateJobInstancePopulatedT def iter_fetch_urls(value: Any): @@ -19,6 +22,15 @@ def iter_fetch_urls(value: Any): yield from iter_fetch_urls(child) +def iter_fetch_request_urls(param_dict: ToolStateJobInstancePopulatedT) -> Iterator[str]: + """Yield URLs from a data-fetch tool's serialized request.""" + request_json = param_dict.get("request_json") + if request_json: + for url in iter_fetch_urls(json.loads(request_json)): + if isinstance(url, str) and url: + yield url + + def fetch_uses_authorization_header(request: dict[str, Any], file_sources, user_context) -> bool: for url in iter_fetch_urls(request): file_source_path = file_sources.get_file_source_path(url) diff --git a/lib/galaxy/tools/parameters/__init__.py b/lib/galaxy/tools/parameters/__init__.py index 93870d45f9e..f939f455114 100644 --- a/lib/galaxy/tools/parameters/__init__.py +++ b/lib/galaxy/tools/parameters/__init__.py @@ -22,6 +22,7 @@ from .basic import ( ColumnListParameter, DataCollectionToolParameter, DataToolParameter, + DirectoryUriToolParameter, ParameterValueError, SelectToolParameter, TextToolParameter, @@ -286,6 +287,21 @@ def visit_input_values( ) +def collect_directory_uris( + inputs: ToolInputsT, + input_values: ToolStateJobInstancePopulatedT, +) -> set[str]: + """Collect the values of every ``directory_uri`` parameter (file source write destinations).""" + uris: set[str] = set() + + def _collect(input, value, **kwargs): + if isinstance(input, DirectoryUriToolParameter) and isinstance(value, str) and value: + uris.add(value) + + visit_input_values(inputs, input_values, _collect) + return uris + + def check_param( trans, param: ToolParameter, incoming_value, param_values, simple_errors: bool = True ) -> tuple[Any, str | ValueError | None]: diff --git a/test/unit/app/jobs/test_job_wrapper.py b/test/unit/app/jobs/test_job_wrapper.py index 1876cefd737..94260d24470 100644 --- a/test/unit/app/jobs/test_job_wrapper.py +++ b/test/unit/app/jobs/test_job_wrapper.py @@ -1,10 +1,12 @@ import abc import os from contextlib import contextmanager +from types import SimpleNamespace from typing import ( cast, TYPE_CHECKING, ) +from uuid import uuid4 from galaxy.app_unittest_utils.tools_support import ( MockContext, @@ -12,6 +14,7 @@ from galaxy.app_unittest_utils.tools_support import ( ) from galaxy.jobs import ( JobWrapper, + MinimalJobWrapper, TaskWrapper, ) from galaxy.jobs.handler import BaseJobHandlerQueue @@ -23,6 +26,8 @@ from galaxy.model import ( ) from galaxy.objectstore import BaseObjectStore from galaxy.tools import ToolBox +from galaxy.tools.parameters.basic import DirectoryUriToolParameter +from galaxy.util import XML from galaxy.util.bunch import Bunch from galaxy.util.unittest import TestCase @@ -150,6 +155,11 @@ class MockTool: self.home_target = None self.tmp_target = None self.tool_source = Bunch(to_string=lambda: "") + self.inputs = {} + self.tool_action = SimpleNamespace(iter_referenced_file_source_uris=lambda param_dict: ()) + + def params_from_strings(self, param_dict): + return param_dict def get_job_destination(self, params): return Bunch(runner="local", id="local", params={}) @@ -197,3 +207,57 @@ class MockObjectStore: if kwds.get("base_dir", "") == "job_work": return self.working_directory return None + + +def _minimal_wrapper(param_dict=None, inputs=None, action_uris=()): + tool_action = SimpleNamespace(iter_referenced_file_source_uris=lambda param_dict: action_uris) + return SimpleNamespace( + tool=SimpleNamespace(inputs=inputs or {}, tool_action=tool_action), + get_param_dict=lambda job: param_dict or {}, + ) + + +def _job_with_file_source_inputs(input_datasets=None, input_library_datasets=None): + return SimpleNamespace( + id=1, + input_datasets=input_datasets or [], + input_library_datasets=input_library_datasets or [], + ) + + +def test_referenced_file_source_uris_reads_tool_parameters_and_action(): + destination = "gxfiles://good/out" + fetched = f"gxuserfiles://{uuid4().hex}/input" + destination_param = DirectoryUriToolParameter(None, XML('')) + wrapper = _minimal_wrapper( + param_dict={"destination": destination}, + inputs={"destination": destination_param}, + action_uris=(fetched,), + ) + assert MinimalJobWrapper._referenced_file_source_uris(wrapper, _job_with_file_source_inputs()) == { + destination, + fetched, + } + + +def test_referenced_file_source_uris_empty_for_job_without_sources(): + assert MinimalJobWrapper._referenced_file_source_uris(_minimal_wrapper(), _job_with_file_source_inputs()) == set() + + +def test_referenced_file_source_uris_adds_regular_and_library_input_sources(): + regular_src = f"gxuserfiles://{uuid4().hex}/regular.txt" + library_src = f"gxuserfiles://{uuid4().hex}/library.txt" + hda = SimpleNamespace(has_deferred_data=True, dataset=SimpleNamespace(source_uris=[regular_src])) + ldda = SimpleNamespace(has_deferred_data=True, dataset=SimpleNamespace(source_uris=[library_src])) + job = _job_with_file_source_inputs( + input_datasets=[SimpleNamespace(dataset=hda)], + input_library_datasets=[SimpleNamespace(dataset=ldda)], + ) + assert MinimalJobWrapper._referenced_file_source_uris(_minimal_wrapper(), job) == {regular_src, library_src} + + +def test_referenced_file_source_uris_ignores_materialized_input_sources(): + source = f"gxuserfiles://{uuid4().hex}/materialized.txt" + hda = SimpleNamespace(has_deferred_data=False, dataset=SimpleNamespace(source_uris=[source])) + job = _job_with_file_source_inputs(input_datasets=[SimpleNamespace(dataset=hda)]) + assert MinimalJobWrapper._referenced_file_source_uris(_minimal_wrapper(), job) == set() diff --git a/test/unit/app/managers/test_user_file_sources.py b/test/unit/app/managers/test_user_file_sources.py index 0c9900b99f5..19c24fadf23 100644 --- a/test/unit/app/managers/test_user_file_sources.py +++ b/test/unit/app/managers/test_user_file_sources.py @@ -2,7 +2,10 @@ import os from typing import ( cast, ) -from uuid import uuid4 +from uuid import ( + UUID, + uuid4, +) import pytest import responses @@ -15,7 +18,12 @@ from galaxy.exceptions import ( RequestParameterInvalidException, RequestParameterMissingException, ) -from galaxy.files import FileSourcesUserContext +from galaxy.files import ( + ConfiguredFileSources, + FileSourcesUserContext, + USER_FILE_SOURCES_SCHEME, +) +from galaxy.files.plugins import FileSourcePluginsConfig from galaxy.files.sources import dropbox from galaxy.files.templates import ConfiguredFileSourceTemplates from galaxy.files.templates.examples import get_example @@ -24,13 +32,13 @@ from galaxy.managers.file_source_instances import ( CreateInstancePayload, FileSourceInstancesManager, ModifyInstancePayload, + referenced_user_file_source_ids, TemplateFormDataRequest, TestUpdateInstancePayload, TestUpgradeInstancePayload, UpdateInstancePayload, UpdateInstanceSecretPayload, UpgradeInstancePayload, - USER_FILE_SOURCES_SCHEME, UserDefinedFileSourcesConfig, UserDefinedFileSourcesImpl, UserFileSourceModel, @@ -326,6 +334,97 @@ class TestFileSourcesTestCase(BaseTestCase): assert not status.connection.is_not_ok assert fsspec_fs_init_kwd["token"] == "my_test_access_token" + def test_serialization_mints_no_tokens_when_no_sources_referenced(self, tmp_path, monkeypatch): + self._init_dropbox_env(tmp_path, monkeypatch) + self._create_dropbox_oauth_source(uuid4().hex) + self._create_dropbox_oauth_source(uuid4().hex) + calls = self._count_refresh_token_mints(monkeypatch) + as_dicts = self.file_sources.user_file_sources_to_dicts( + True, + cast(FileSourcesUserContext, self.trans), + referenced_uris=set(), + ) + assert calls == [] + assert as_dicts == [] + + def test_serialization_mints_only_referenced_source(self, tmp_path, monkeypatch): + self._init_dropbox_env(tmp_path, monkeypatch) + referenced_uuid = uuid4().hex + self._create_dropbox_oauth_source(referenced_uuid) + self._create_dropbox_oauth_source(uuid4().hex) + calls = self._count_refresh_token_mints(monkeypatch) + as_dicts = self.file_sources.user_file_sources_to_dicts( + True, + cast(FileSourcesUserContext, self.trans), + referenced_uris={f"gxuserfiles://{referenced_uuid}/some/path"}, + ) + assert calls == [f"refresh_token_{referenced_uuid}"] + assert len(as_dicts) == 1 + assert UUID(as_dicts[0]["id"]).hex == referenced_uuid + assert as_dicts[0]["access_token"] == "my_test_access_token" + + def test_serialization_without_reference_filter_includes_all_sources(self, tmp_path, monkeypatch): + self._init_dropbox_env(tmp_path, monkeypatch) + self._create_dropbox_oauth_source(uuid4().hex) + self._create_dropbox_oauth_source(uuid4().hex) + calls = self._count_refresh_token_mints(monkeypatch) + as_dicts = self.file_sources.user_file_sources_to_dicts( + True, + cast(FileSourcesUserContext, self.trans), + ) + assert len(calls) == 2 + assert len(as_dicts) == 2 + + def test_configured_file_sources_to_dict_threads_referenced_uris(self, tmp_path, monkeypatch): + self._init_dropbox_env(tmp_path, monkeypatch) + referenced_uuid = uuid4().hex + self._create_dropbox_oauth_source(referenced_uuid) + self._create_dropbox_oauth_source(uuid4().hex) + calls = self._count_refresh_token_mints(monkeypatch) + configured = ConfiguredFileSources( + FileSourcePluginsConfig(), + user_defined_file_sources=self.file_sources, + ) + as_dict = configured.to_dict( + for_serialization=True, + user_context=cast(FileSourcesUserContext, self.trans), + referenced_uris={f"gxuserfiles://{referenced_uuid}/x"}, + ) + ids = [s["id"] for s in as_dict["file_sources"]] + assert calls == [f"refresh_token_{referenced_uuid}"] + assert len(ids) == 1 + assert UUID(ids[0]).hex == referenced_uuid + + def _create_dropbox_oauth_source(self, uuid: str) -> None: + config_secret_key = UserFileSource.vault_key_from_uuid(uuid, "_oauth2_refresh_token", None) + # Seed a per-source refresh token so a captured mint identifies which source minted. + self.trans.user_vault.write_secret(config_secret_key, f"refresh_token_{uuid}") + self._create_instance( + CreateInstancePayload( + name=SIMPLE_FILE_SOURCE_NAME, + description=SIMPLE_FILE_SOURCE_DESCRIPTION, + template_id="dropbox", + template_version=0, + variables={}, + secrets={}, + uuid=uuid, + ) + ) + + def _count_refresh_token_mints(self, monkeypatch) -> list: + calls: list = [] + + class MockDropboxDriveFileSystem: + pass + + def mock_get_token_from_refresh_raw(refresh_token, client_pair, config): + calls.append(refresh_token) + return MockResponse({"access_token": "my_test_access_token"}) + + monkeypatch.setattr(config_templates, "get_token_from_refresh_raw", mock_get_token_from_refresh_raw) + monkeypatch.setattr(dropbox.DropboxFilesSource, "required_module", MockDropboxDriveFileSystem) + return calls + def test_onedrive_oauth2_flow(self, tmp_path, monkeypatch): json = { "refresh_token": "my_test_refresh_token", @@ -1197,3 +1296,33 @@ class OneDriveMockResponse: def json(self): return self._json_data + + +def test_referenced_user_file_source_ids_selects_only_user_sources(): + user_uuid = uuid4().hex + uris = { + f"gxuserfiles://{user_uuid}/some/path", + "gxfiles://dropbox/other", + "https://example.com/file.txt", + "drs://example.org/abc", + } + assert referenced_user_file_source_ids(uris) == {user_uuid} + + +def test_referenced_user_file_source_ids_normalizes_dashed_uuid(): + dashed = str(uuid4()) + assert referenced_user_file_source_ids({f"gxuserfiles://{dashed}/x"}) == {dashed.replace("-", "")} + + +def test_referenced_user_file_source_ids_handles_no_match(): + uris = {"gxfiles://dropbox/x", "not-a-uri", ""} + assert referenced_user_file_source_ids(uris) == set() + + +@pytest.mark.parametrize( + "uri", + ["gxuserfiles://[invalid-authority/x", "gxuserfiles://not-a-uuid/x", "gxuserfiles:///x"], +) +def test_referenced_user_file_source_ids_rejects_invalid_user_source_uri(uri): + with pytest.raises(RequestParameterInvalidException, match="Invalid user file source URI"): + referenced_user_file_source_ids({uri}) diff --git a/test/unit/app/tools/test_collect_directory_uris.py b/test/unit/app/tools/test_collect_directory_uris.py new file mode 100644 index 00000000000..b2b20418811 --- /dev/null +++ b/test/unit/app/tools/test_collect_directory_uris.py @@ -0,0 +1,30 @@ +from galaxy.tools.parameters import collect_directory_uris +from galaxy.tools.parameters.basic import ( + DirectoryUriToolParameter, + TextToolParameter, +) +from galaxy.util import XML + + +def _directory_uri_param(name): + return DirectoryUriToolParameter(None, XML(f'')) + + +def _text_param(name): + return TextToolParameter(None, XML(f'')) + + +def test_collect_directory_uris_selects_only_directory_uri_params(): + inputs = {"dest": _directory_uri_param("dest"), "other": _text_param("other")} + values = {"dest": "gxfiles://target/out", "other": "gxfiles://not-a-destination/x"} + assert collect_directory_uris(inputs, values) == {"gxfiles://target/out"} + + +def test_collect_directory_uris_empty_when_no_directory_uri_params(): + inputs = {"other": _text_param("other")} + assert collect_directory_uris(inputs, {"other": "text"}) == set() + + +def test_collect_directory_uris_skips_empty_values(): + inputs = {"dest": _directory_uri_param("dest")} + assert collect_directory_uris(inputs, {"dest": ""}) == set() diff --git a/test/unit/app/tools/test_data_fetch_utils.py b/test/unit/app/tools/test_data_fetch_utils.py index cc8a5a8b2dd..f2d8d9355d8 100644 --- a/test/unit/app/tools/test_data_fetch_utils.py +++ b/test/unit/app/tools/test_data_fetch_utils.py @@ -1,12 +1,26 @@ +import json from datetime import ( datetime, timedelta, timezone, ) -from typing import cast +from typing import ( + Any, + cast, +) from galaxy.model import User -from galaxy.tools.data_fetch_utils import compute_token_expiry_for_provider +from galaxy.tools.actions.upload import FetchUploadToolAction +from galaxy.tools.data_fetch_utils import ( + compute_token_expiry_for_provider, + iter_fetch_request_urls, + iter_fetch_urls, +) + + +class ConcreteFetchUploadToolAction(FetchUploadToolAction): + def get_output_name(self, *args: Any, **kwargs: Any) -> str: + raise NotImplementedError class DummyToken: @@ -62,3 +76,32 @@ def test_compute_token_expiry_for_provider_returns_none_when_token_missing_auth_ token.extra_data = {} user = DummyUser([token]) assert compute_token_expiry_for_provider(cast(User, user), "oidc") is None + + +def test_iter_fetch_request_urls_extracts_urls(): + request = { + "targets": [ + {"elements": [{"src": "url", "url": "gxuserfiles://abc/x"}, {"src": "url", "url": "https://e.com/y"}]} + ] + } + param_dict = {"request_json": json.dumps(request)} + assert set(iter_fetch_request_urls(param_dict)) == {"gxuserfiles://abc/x", "https://e.com/y"} + + +def test_iter_fetch_request_urls_empty_without_request_json(): + assert list(iter_fetch_request_urls({})) == [] + assert list(iter_fetch_request_urls({"request_json": ""})) == [] + + +def test_iter_fetch_request_urls_ignores_non_string_and_empty_urls(): + request = {"targets": [{"elements": [{"src": "url", "url": None}, {"src": "url", "url": ""}]}]} + assert list(iter_fetch_urls(request)) == [None, ""] + assert list(iter_fetch_request_urls({"request_json": json.dumps(request)})) == [] + + +def test_fetch_upload_action_reports_referenced_file_source_uris(): + request = {"targets": [{"elements": [{"src": "url", "url": "gxfiles://source/input"}]}]} + action = ConcreteFetchUploadToolAction() + assert list(action.iter_referenced_file_source_uris({"request_json": json.dumps(request)})) == [ + "gxfiles://source/input" + ] diff --git a/test/unit/app/tools/test_history_imp_exp.py b/test/unit/app/tools/test_history_imp_exp.py index c0b15687322..8b07515c97a 100644 --- a/test/unit/app/tools/test_history_imp_exp.py +++ b/test/unit/app/tools/test_history_imp_exp.py @@ -3,6 +3,7 @@ import os import tarfile import tempfile from shutil import rmtree +from typing import Any from unittest.mock import Mock from sqlalchemy import select @@ -12,6 +13,7 @@ from galaxy.app_unittest_utils.galaxy_mock import MockApp from galaxy.exceptions import MalformedContents from galaxy.model.orm.util import add_object_to_object_session from galaxy.objectstore.unittest_utils import Config as TestConfig +from galaxy.tools.actions.history_imp_exp import ImportHistoryToolAction from galaxy.tools.imp_exp import ( JobExportHistoryArchiveWrapper, JobImportHistoryArchiveWrapper, @@ -28,6 +30,24 @@ HISTORY_ATTRS = """{"hid_counter": 2, "update_time": "2016-02-08 18:38:38.705058 JOBS_ATTRS = """[{"info": null, "tool_id": "upload1", "update_time": "2016-02-08T18:39:23.356482", "stdout": "", "input_mapping": {}, "tool_version": "1.1.4", "traceback": null, "command_line": "python /galaxy/tools/data_source/upload.py /galaxy /scratch/tmppwU9rD /scratch/tmpP4_45Y 1:/scratch/jobs/000/dataset_1_files:/data/000/dataset_1.dat", "exit_code": 0, "output_datasets": [1], "state": "ok", "create_time": "2016-02-08T18:38:39.153873", "params": {"files": [{"to_posix_lines": "Yes", "NAME": "None", "file_data": null, "space_to_tab": null, "url_paste": "/scratch/strio_url_paste_o6nrv8", "__index__": 0, "ftp_files": "", "uuid": "None"}], "paramfile": "/scratch/tmpP4_45Y", "file_type": "auto", "files_metadata": {"file_type": "auto", "__current_case__": 41}, "async_datasets": "None", "dbkey": "?"}, "stderr": ""}]""" +class ConcreteImportHistoryToolAction(ImportHistoryToolAction): + def get_output_name(self, *args: Any, **kwargs: Any) -> str: + raise NotImplementedError + + +def test_import_history_action_reports_url_archive_source(): + action = ConcreteImportHistoryToolAction() + archive_source = "https://example.org/history.tar.gz" + assert list( + action.iter_referenced_file_source_uris({"__ARCHIVE_TYPE__": "url", "__ARCHIVE_SOURCE__": archive_source}) + ) == [archive_source] + assert not list( + action.iter_referenced_file_source_uris( + {"__ARCHIVE_TYPE__": "file", "__ARCHIVE_SOURCE__": "/tmp/history.tar.gz"} + ) + ) + + def t_data_path(name): return os.path.join(galaxy_directory(), "test-data", name) diff --git a/test/unit/app/tools/test_upload_actions.py b/test/unit/app/tools/test_upload_actions.py new file mode 100644 index 00000000000..91903d2ee3c --- /dev/null +++ b/test/unit/app/tools/test_upload_actions.py @@ -0,0 +1,24 @@ +import json +import tempfile +from typing import Any + +from galaxy.tools.actions.upload import UploadToolAction + + +class ConcreteUploadToolAction(UploadToolAction): + def get_output_name(self, *args: Any, **kwargs: Any) -> str: + raise NotImplementedError + + +def test_upload_action_reports_urls_from_paramfile(): + upload_params = [ + {"type": "url", "path": "https://example.org/input.txt"}, + {"type": "file", "path": "/tmp/pasted-input.txt"}, + ] + with tempfile.NamedTemporaryFile(mode="w") as paramfile: + json.dump(upload_params, paramfile) + paramfile.flush() + action = ConcreteUploadToolAction() + assert list(action.iter_referenced_file_source_uris({"paramfile": paramfile.name})) == [ + "https://example.org/input.txt" + ] diff --git a/test/unit/files/test_http.py b/test/unit/files/test_http.py index c0543f446a5..c985b4669de 100644 --- a/test/unit/files/test_http.py +++ b/test/unit/files/test_http.py @@ -40,6 +40,18 @@ def test_file_source_http_specific(): assert_realizes_as(file_sources, test_url, "hello specific world", user_context=user_context) +def test_plugins_to_dict_serializes_only_best_matching_http_source(): + test_url = "https://www.usegalaxy.org/myfile.txt" + user_context = user_context_fixture() + file_sources = configured_file_sources(FILE_SOURCES_CONF) + plugins = file_sources.plugins_to_dict( + for_serialization=True, + user_context=user_context, + referenced_uris={test_url}, + ) + assert [plugin["id"] for plugin in plugins] == ["test1"] + + def test_file_source_another_http_specific(): test_url = "http://www.galaxyproject.org/anotherfile.txt" diff --git a/test/unit/files/test_posix.py b/test/unit/files/test_posix.py index c130c5f541b..e34c2bba01f 100644 --- a/test/unit/files/test_posix.py +++ b/test/unit/files/test_posix.py @@ -21,6 +21,7 @@ from galaxy.files.unittest_utils import ( from ._util import ( assert_realizes_as, assert_realizes_throws_exception, + configured_file_sources, find, find_file_a, list_dir, @@ -556,3 +557,33 @@ def test_get_file_source_path_strips_whitespace(): resolved = file_sources.get_file_source_path("\ngxfiles://test1/a\n") assert resolved.file_source is not None assert resolved.path == "/a" + + +def _two_posix_file_sources(tmp_path): + root_good = tmp_path / "good" + root_other = tmp_path / "other" + root_good.mkdir() + root_other.mkdir() + return configured_file_sources( + [ + {"type": "posix", "id": "good", "root": str(root_good)}, + {"type": "posix", "id": "other", "root": str(root_other)}, + ] + ) + + +def test_plugins_to_dict_serializes_only_referenced_sources(tmp_path): + file_sources = _two_posix_file_sources(tmp_path) + plugins = file_sources.plugins_to_dict(for_serialization=True, referenced_uris={"gxfiles://good/some/file"}) + assert [p["id"] for p in plugins] == ["good"] + + +def test_plugins_to_dict_serializes_nothing_when_no_uris_referenced(tmp_path): + file_sources = _two_posix_file_sources(tmp_path) + assert file_sources.plugins_to_dict(for_serialization=True, referenced_uris=set()) == [] + + +def test_plugins_to_dict_serializes_all_when_referenced_uris_none(tmp_path): + file_sources = _two_posix_file_sources(tmp_path) + plugins = file_sources.plugins_to_dict(for_serialization=True) + assert {p["id"] for p in plugins} == {"good", "other"} From 1c1e413789fe5bb300e65eb807ea2ac51bcedf59 Mon Sep 17 00:00:00 2001 From: mvdbeek Date: Tue, 25 Aug 2026 10:52:20 +0200 Subject: [PATCH 2/2] Make file source URI discovery fail safe Treat action-specific URI discovery as complete only when the concrete action explicitly declares it. Unknown custom actions retain the legacy behavior of serializing every file source. Keep legacy upload paramfile parsing fail-fast and avoid constructing JobIO during failure cleanup so a secondary error cannot mask the original exception. --- lib/galaxy/jobs/__init__.py | 13 +++++-- lib/galaxy/tools/actions/__init__.py | 6 +++ lib/galaxy/tools/actions/data_manager.py | 2 + lib/galaxy/tools/actions/data_source.py | 2 + lib/galaxy/tools/actions/history_imp_exp.py | 2 + lib/galaxy/tools/actions/metadata.py | 1 + lib/galaxy/tools/actions/model_operations.py | 1 + lib/galaxy/tools/actions/upload.py | 21 ++++++++-- test/unit/app/jobs/test_job_wrapper.py | 29 ++++++++++++-- test/unit/app/tools/test_upload_actions.py | 41 ++++++++++++++++++++ 10 files changed, 107 insertions(+), 11 deletions(-) diff --git a/lib/galaxy/jobs/__init__.py b/lib/galaxy/jobs/__init__.py index 81269b4e55d..d0d8f0ca44b 100644 --- a/lib/galaxy/jobs/__init__.py +++ b/lib/galaxy/jobs/__init__.py @@ -1084,13 +1084,16 @@ class MinimalJobWrapper(HasResourceParameters): if authnz_manager and trans.user: authnz_manager.refresh_expiring_oidc_tokens(trans, trans.user) - def _referenced_file_source_uris(self, job: Job) -> set[str]: - """Return the file source URIs needed to execute this job.""" + def _referenced_file_source_uris(self, job: Job) -> set[str] | None: + """Return required URIs, or ``None`` when action discovery is incomplete.""" uris: set[str] = set() if self.tool is not None: + tool_action = self.tool.tool_action + if not tool_action.has_complete_file_source_uri_discovery(): + return None param_dict = self.get_param_dict(job) uris.update(collect_directory_uris(self.tool.inputs, param_dict)) - uris.update(self.tool.tool_action.iter_referenced_file_source_uris(param_dict)) + uris.update(tool_action.iter_referenced_file_source_uris(param_dict)) for input_association in job.input_datasets + job.input_library_datasets: dataset = input_association.dataset if dataset is not None and dataset.has_deferred_data and dataset.dataset is not None: @@ -1437,7 +1440,9 @@ class MinimalJobWrapper(HasResourceParameters): return tool_evaluator def _fix_output_permissions(self): - for path in [dp.real_path for dp in self.job_io.get_mutable_output_fnames()]: + if self._job_io is None: + return + for path in [dp.real_path for dp in self._job_io.get_mutable_output_fnames()]: if os.path.exists(path): util.umask_fix_perms(path, self.app.config.umask, 0o666, self.app.config.gid) diff --git a/lib/galaxy/tools/actions/__init__.py b/lib/galaxy/tools/actions/__init__.py index 49e0b44008a..fc54362e186 100644 --- a/lib/galaxy/tools/actions/__init__.py +++ b/lib/galaxy/tools/actions/__init__.py @@ -99,6 +99,11 @@ class ToolAction: """ produces_real_jobs: bool + file_source_uri_discovery_complete = False + + def has_complete_file_source_uri_discovery(self) -> bool: + """Return whether this concrete action has audited URI discovery.""" + return type(self).__dict__.get("file_source_uri_discovery_complete", False) def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: """Yield file source URIs embedded in action-specific parameters.""" @@ -146,6 +151,7 @@ class DefaultToolAction(ToolAction): """Default tool action is to run an external command""" produces_real_jobs: bool = True + file_source_uri_discovery_complete = True def _collect_input_datasets( self, diff --git a/lib/galaxy/tools/actions/data_manager.py b/lib/galaxy/tools/actions/data_manager.py index c15b3815630..2d28c4031db 100644 --- a/lib/galaxy/tools/actions/data_manager.py +++ b/lib/galaxy/tools/actions/data_manager.py @@ -29,6 +29,8 @@ log = logging.getLogger(__name__) class DataManagerToolAction(DefaultToolAction): """Tool action used for Data Manager Tools""" + file_source_uri_discovery_complete = True + def execute( self, tool, diff --git a/lib/galaxy/tools/actions/data_source.py b/lib/galaxy/tools/actions/data_source.py index 14f3a1b1044..0a485a16eff 100644 --- a/lib/galaxy/tools/actions/data_source.py +++ b/lib/galaxy/tools/actions/data_source.py @@ -8,6 +8,8 @@ log = logging.getLogger(__name__) class DataSourceToolAction(DefaultToolAction): """Tool action used for Data Source Tools""" + file_source_uri_discovery_complete = True + def _get_default_data_name( self, dataset, tool, on_text=None, trans=None, incoming=None, history=None, params=None, job_params=None, **kwd ): diff --git a/lib/galaxy/tools/actions/history_imp_exp.py b/lib/galaxy/tools/actions/history_imp_exp.py index 6561a7d88f1..5af668a949f 100644 --- a/lib/galaxy/tools/actions/history_imp_exp.py +++ b/lib/galaxy/tools/actions/history_imp_exp.py @@ -41,6 +41,7 @@ class ImportHistoryToolAction(ToolAction): """Tool action used for importing a history to an archive.""" produces_real_jobs: bool = True + file_source_uri_discovery_complete = True def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: if param_dict.get("__ARCHIVE_TYPE__") == "url": @@ -124,6 +125,7 @@ class ExportHistoryToolAction(ToolAction): """Tool action used for exporting a history to an archive.""" produces_real_jobs: bool = True + file_source_uri_discovery_complete = True def execute( self, diff --git a/lib/galaxy/tools/actions/metadata.py b/lib/galaxy/tools/actions/metadata.py index 1c0a5d2dc41..7fb433299a4 100644 --- a/lib/galaxy/tools/actions/metadata.py +++ b/lib/galaxy/tools/actions/metadata.py @@ -40,6 +40,7 @@ class SetMetadataToolAction(ToolAction): produces_real_jobs: bool = False set_output_hid: bool = False + file_source_uri_discovery_complete = True def execute( self, diff --git a/lib/galaxy/tools/actions/model_operations.py b/lib/galaxy/tools/actions/model_operations.py index a640547228c..6497679f8ab 100644 --- a/lib/galaxy/tools/actions/model_operations.py +++ b/lib/galaxy/tools/actions/model_operations.py @@ -40,6 +40,7 @@ log = logging.getLogger(__name__) class ModelOperationToolAction(DefaultToolAction): produces_real_jobs: bool = False + file_source_uri_discovery_complete = True def check_inputs_ready(self, tool, trans, incoming, history, execution_cache=None, collection_info=None): if execution_cache is None: diff --git a/lib/galaxy/tools/actions/upload.py b/lib/galaxy/tools/actions/upload.py index 8d1da22e579..200b3c447d8 100644 --- a/lib/galaxy/tools/actions/upload.py +++ b/lib/galaxy/tools/actions/upload.py @@ -3,7 +3,10 @@ import logging import os from collections.abc import Iterable -from galaxy.exceptions import RequestParameterMissingException +from galaxy.exceptions import ( + RequestParameterInvalidException, + RequestParameterMissingException, +) from galaxy.job_execution.output_collect import copy_collection_metadata_from_target_dict from galaxy.managers.context import ProvidesHistoryContext from galaxy.model import ( @@ -40,6 +43,7 @@ log = logging.getLogger(__name__) class BaseUploadToolAction(ToolAction): produces_real_jobs = True + file_source_uri_discovery_complete = False def execute( self, @@ -88,17 +92,24 @@ class BaseUploadToolAction(ToolAction): class UploadToolAction(BaseUploadToolAction): + file_source_uri_discovery_complete = True + def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: paramfile = param_dict.get("paramfile") if not isinstance(paramfile, str): - return + raise RequestParameterInvalidException("Legacy upload job is missing its paramfile") with open(paramfile) as f: upload_params = json.load(f) + if not isinstance(upload_params, list): + raise RequestParameterInvalidException("Legacy upload paramfile must contain a list") for upload_param in upload_params: + if not isinstance(upload_param, dict): + raise RequestParameterInvalidException("Legacy upload paramfile entries must be objects") if upload_param.get("type") == "url": path = upload_param.get("path") - if isinstance(path, str) and path: - yield path + if not isinstance(path, str) or not path: + raise RequestParameterInvalidException("Legacy URL upload entry is missing its path") + yield path def _setup_job( self, tool, trans: ProvidesHistoryContext, incoming, dataset_upload_inputs, history, preferred_object_store_id @@ -126,6 +137,8 @@ class UploadToolAction(BaseUploadToolAction): class FetchUploadToolAction(BaseUploadToolAction): + file_source_uri_discovery_complete = True + def iter_referenced_file_source_uris(self, param_dict: ToolStateJobInstancePopulatedT) -> Iterable[str]: return iter_fetch_request_urls(param_dict) diff --git a/test/unit/app/jobs/test_job_wrapper.py b/test/unit/app/jobs/test_job_wrapper.py index 94260d24470..690be9b22f5 100644 --- a/test/unit/app/jobs/test_job_wrapper.py +++ b/test/unit/app/jobs/test_job_wrapper.py @@ -156,7 +156,10 @@ class MockTool: self.tmp_target = None self.tool_source = Bunch(to_string=lambda: "") self.inputs = {} - self.tool_action = SimpleNamespace(iter_referenced_file_source_uris=lambda param_dict: ()) + self.tool_action = SimpleNamespace( + has_complete_file_source_uri_discovery=lambda: True, + iter_referenced_file_source_uris=lambda param_dict: (), + ) def params_from_strings(self, param_dict): return param_dict @@ -209,8 +212,11 @@ class MockObjectStore: return None -def _minimal_wrapper(param_dict=None, inputs=None, action_uris=()): - tool_action = SimpleNamespace(iter_referenced_file_source_uris=lambda param_dict: action_uris) +def _minimal_wrapper(param_dict=None, inputs=None, action_uris=(), action_discovery_complete=True): + tool_action = SimpleNamespace( + has_complete_file_source_uri_discovery=lambda: action_discovery_complete, + iter_referenced_file_source_uris=lambda param_dict: action_uris, + ) return SimpleNamespace( tool=SimpleNamespace(inputs=inputs or {}, tool_action=tool_action), get_param_dict=lambda job: param_dict or {}, @@ -244,6 +250,11 @@ def test_referenced_file_source_uris_empty_for_job_without_sources(): assert MinimalJobWrapper._referenced_file_source_uris(_minimal_wrapper(), _job_with_file_source_inputs()) == set() +def test_referenced_file_source_uris_unknown_for_unaudited_action(): + wrapper = _minimal_wrapper(action_discovery_complete=False) + assert MinimalJobWrapper._referenced_file_source_uris(wrapper, _job_with_file_source_inputs()) is None + + def test_referenced_file_source_uris_adds_regular_and_library_input_sources(): regular_src = f"gxuserfiles://{uuid4().hex}/regular.txt" library_src = f"gxuserfiles://{uuid4().hex}/library.txt" @@ -261,3 +272,15 @@ def test_referenced_file_source_uris_ignores_materialized_input_sources(): hda = SimpleNamespace(has_deferred_data=False, dataset=SimpleNamespace(source_uris=[source])) job = _job_with_file_source_inputs(input_datasets=[SimpleNamespace(dataset=hda)]) assert MinimalJobWrapper._referenced_file_source_uris(_minimal_wrapper(), job) == set() + + +def test_fix_output_permissions_does_not_initialize_job_io(): + class WrapperWithoutJobIO: + _job_io = None + + @property + def job_io(self): + raise AssertionError("job_io should not be initialized during failure cleanup") + + wrapper = cast(MinimalJobWrapper, WrapperWithoutJobIO()) + MinimalJobWrapper._fix_output_permissions(wrapper) diff --git a/test/unit/app/tools/test_upload_actions.py b/test/unit/app/tools/test_upload_actions.py index 91903d2ee3c..354d9ff8768 100644 --- a/test/unit/app/tools/test_upload_actions.py +++ b/test/unit/app/tools/test_upload_actions.py @@ -2,6 +2,9 @@ import json import tempfile from typing import Any +import pytest + +from galaxy.exceptions import RequestParameterInvalidException from galaxy.tools.actions.upload import UploadToolAction @@ -22,3 +25,41 @@ def test_upload_action_reports_urls_from_paramfile(): assert list(action.iter_referenced_file_source_uris({"paramfile": paramfile.name})) == [ "https://example.org/input.txt" ] + + +@pytest.mark.parametrize( + ("upload_params", "message"), + [ + ({"type": "url", "path": "https://example.org/input.txt"}, "must contain a list"), + (["not an object"], "entries must be objects"), + ([{"type": "url", "path": None}], "URL upload entry is missing its path"), + ], +) +def test_upload_action_rejects_invalid_paramfile_shape(upload_params, message): + with tempfile.NamedTemporaryFile(mode="w") as paramfile: + json.dump(upload_params, paramfile) + paramfile.flush() + action = ConcreteUploadToolAction() + with pytest.raises(RequestParameterInvalidException, match=message): + list(action.iter_referenced_file_source_uris({"paramfile": paramfile.name})) + + +def test_upload_action_rejects_missing_paramfile_parameter(): + action = ConcreteUploadToolAction() + with pytest.raises(RequestParameterInvalidException, match="missing its paramfile"): + list(action.iter_referenced_file_source_uris({})) + + +def test_upload_action_does_not_hide_missing_paramfile(tmp_path): + action = ConcreteUploadToolAction() + with pytest.raises(FileNotFoundError): + list(action.iter_referenced_file_source_uris({"paramfile": str(tmp_path / "missing.json")})) + + +def test_upload_action_does_not_hide_malformed_paramfile(): + with tempfile.NamedTemporaryFile(mode="w") as paramfile: + paramfile.write("{") + paramfile.flush() + action = ConcreteUploadToolAction() + with pytest.raises(json.JSONDecodeError): + list(action.iter_referenced_file_source_uris({"paramfile": paramfile.name}))