Stock tools.

lib/galaxy/tools/stock.py

@@ -0,0 +1,33 @@
+"""Reason about stock tools based on ToolSource abstractions."""
+
+from pathlib import Path
+
+from lxml.etree import XMLSyntaxError
+
+# Set GALAXY_INCLUDES_ROOT from tool shed to point this at a Galaxy root
+# (once we are running the tool shed from packages not rooted with Galaxy).
+import galaxy.tools
+from galaxy.tool_util.parser import get_tool_source
+from galaxy.util import galaxy_directory
+from galaxy.util.resources import files
+
+
+def stock_tool_paths():
+    yield from _walk_directory_for_tools(files(galaxy.tools))
+    yield from _walk_directory_for_tools(Path(galaxy_directory()) / "test" / "functional" / "tools")
+
+
+def stock_tool_sources():
+    for stock_tool_path in stock_tool_paths():
+        try:
+            yield get_tool_source(str(stock_tool_path))
+        except XMLSyntaxError:
+            continue
+
+
+def _walk_directory_for_tools(path):
+    if path.is_file() and path.name.endswith(".xml"):
+        yield path
+    elif path.is_dir():
+        for directory in path.iterdir():
+            yield from _walk_directory_for_tools(directory)

lib/galaxy/util/__init__.py

@@ -1738,12 +1738,20 @@ def safe_str_cmp(a, b):
     return rv == 0
 
 
+# never load packages this way (won't work for installed packages),
+# but while we're working on packaging everything this can be a way to point
+# an installed Galaxy at a Galaxy root for things like tools. Ultimately
+# this all needs to be packaged, but we have some very old PRs working on this
+# that are pretty tricky and shouldn't slow current development.
+GALAXY_INCLUDES_ROOT = os.environ.get("GALAXY_INCLUDES_ROOT")
+
+
 #  Don't use this directly, prefer method version that "works" with packaged Galaxy.
-galaxy_root_path = Path(__file__).parent.parent.parent.parent
+galaxy_root_path = Path(GALAXY_INCLUDES_ROOT) if GALAXY_INCLUDES_ROOT else Path(__file__).parent.parent.parent.parent
 
 
 def galaxy_directory() -> str:
-    if in_packages():
+    if in_packages() and not GALAXY_INCLUDES_ROOT:
         # This will work only when running pytest from <galaxy_root>/packages/<package_name>/
         cwd = Path.cwd()
         path = cwd.parent.parent

lib/tool_shed/managers/tools.py

@@ -2,6 +2,7 @@ import os
 import tempfile
 from collections import namedtuple
 from typing import (
+    Dict,
     List,
     Optional,
     Tuple,
@@ -27,6 +28,7 @@ from galaxy.tool_util.parser import (
     get_tool_source,
     ToolSource,
 )
+from galaxy.tools.stock import stock_tool_sources
 from tool_shed.context import (
     ProvidesRepositoriesContext,
     SessionRequestContext,
@@ -36,6 +38,8 @@ from tool_shed.webapp.model import RepositoryMetadata
 from tool_shed.webapp.search.tool_search import ToolSearch
 from .trs import trs_tool_id_to_repository_metadata
 
+STOCK_TOOL_SOURCES: Optional[Dict[str, Dict[str, ToolSource]]] = None
+
 
 def search(trans: SessionRequestContext, q: str, page: int = 1, page_size: int = 10) -> dict:
     """
@@ -114,6 +118,18 @@ def tool_input_models_for(
 
 def tool_source_for(
     trans: ProvidesRepositoriesContext, trs_tool_id: str, tool_version: str, repository_clone_url: Optional[str] = None
+) -> ToolSource:
+    if "~" in trs_tool_id:
+        return _shed_tool_source_for(trans, trs_tool_id, tool_version, repository_clone_url)
+    else:
+        tool_source = _stock_tool_source_for(trs_tool_id, tool_version)
+        if tool_source is None:
+            raise ObjectNotFound()
+        return tool_source
+
+
+def _shed_tool_source_for(
+    trans: ProvidesRepositoriesContext, trs_tool_id: str, tool_version: str, repository_clone_url: Optional[str] = None
 ) -> ToolSource:
     rval = get_repository_metadata_tool_dict(trans, trs_tool_id, tool_version)
     repository_metadata, tool_version_metadata = rval
@@ -133,3 +149,24 @@ def tool_source_for(
         return tool_source
     finally:
         remove_dir(work_dir)
+
+
+def _stock_tool_source_for(tool_id: str, tool_version: str) -> Optional[ToolSource]:
+    _init_stock_tool_sources()
+    assert STOCK_TOOL_SOURCES
+    tool_version_sources = STOCK_TOOL_SOURCES.get(tool_id)
+    if tool_version_sources is None:
+        return None
+    return tool_version_sources.get(tool_version)
+
+
+def _init_stock_tool_sources() -> None:
+    global STOCK_TOOL_SOURCES
+    if STOCK_TOOL_SOURCES is None:
+        STOCK_TOOL_SOURCES = {}
+        for tool_source in stock_tool_sources():
+            tool_id = tool_source.parse_id()
+            tool_version = tool_source.parse_version()
+            if tool_id not in STOCK_TOOL_SOURCES:
+                STOCK_TOOL_SOURCES[tool_id] = {}
+            STOCK_TOOL_SOURCES[tool_id][tool_version] = tool_source

test/unit/app/tools/test_stock.py

@@ -0,0 +1,16 @@
+from galaxy.tools.stock import (
+    stock_tool_paths,
+    stock_tool_sources,
+)
+
+
+def test_stock_tool_paths():
+    file_names = [f.name for f in list(stock_tool_paths())]
+    assert "merge_collection.xml" in file_names
+    assert "meme.xml" in file_names
+    assert "output_auto_format.xml" in file_names
+
+
+def test_stock_tool_sources():
+    tool_source = next(stock_tool_sources())
+    assert tool_source.parse_id()

test/unit/tool_shed/test_tool_source.py

@@ -30,4 +30,9 @@ def test_get_tool(provides_repositories: ProvidesRepositoriesContext, new_reposi
     )
     assert len(cached_bundle.input_models) == 3
 
-    print(RequestToolState.parameter_model_for(cached_bundle).model_json_schema())
+
+def test_stock_bundle(provides_repositories: ProvidesRepositoriesContext):
+    cached_bundle = tool_input_models_cached_for(
+        provides_repositories, "__ZIP_COLLECTION__", "1.0.0", repository_clone_url=None
+    )
+    assert len(cached_bundle.input_models) == 2