Merge pull request #1312 from mvdbeek/fix_for_paths

Fix ``for_paths`` when path is directory of tools

planemo/runnable.py

@@ -15,7 +15,6 @@
     List,
     NamedTuple,
     Optional,
-    overload,
     Union,
 )
 from urllib.parse import urlparse
@@ -31,7 +30,6 @@
     looks_like_a_tool_xml,
 )
 from galaxy.tool_util.parser import get_tool_source
-from typing_extensions import Literal
 
 from planemo.exit_codes import (
     EXIT_CODE_UNKNOWN_FILE_TYPE,
@@ -181,38 +179,24 @@ def workflows_from_dockstore_yaml(path):
     return workflows
 
 
-def workflow_dir_runnables(path: str, return_all: bool = False) -> Optional[Union[Runnable, List[Runnable]]]:
+def workflow_dir_runnables(path: str) -> List[Runnable]:
     dockstore_path = os.path.join(path, DOCKSTORE_REGISTRY_CONF)
     if os.path.exists(dockstore_path):
-        runnables = [
+        return [
             Runnable(str(path), RunnableType.galaxy_workflow) for path in workflows_from_dockstore_yaml(dockstore_path)
         ]
-        if return_all:
-            return runnables
-        else:
-            return runnables[0]
-    return None
+    return []
 
 
 def tool_dir_runnables(path: str) -> List[Runnable]:
-    return [for_path(p) for (p, _) in yield_tool_sources_on_paths(ctx=None, paths=[path])]
-
-
-@overload
-def for_path(path: str, return_all: Literal[False] = False) -> Runnable:
-    ...
+    return for_paths(tool_path for tool_path, _ in yield_tool_sources_on_paths(ctx=None, paths=[path]))
 
 
-@overload
-def for_path(path: str, return_all: bool = False) -> Union[Runnable, List[Runnable]]:
-    pass
-
-
-def for_path(path: str, return_all: bool = False) -> Union[Runnable, List[Runnable]]:
+def for_path(path: str) -> Union[Runnable, List[Runnable]]:
     """Produce a class:`Runnable` for supplied path."""
     runnable_type = None
     if os.path.isdir(path):
-        runnable = workflow_dir_runnables(path, return_all=return_all) or tool_dir_runnables(path)
+        runnable = workflow_dir_runnables(path) or tool_dir_runnables(path)
         if runnable:
             return runnable
         runnable_type = RunnableType.directory
@@ -247,7 +231,14 @@ def for_path(path: str, return_all: bool = False) -> Union[Runnable, List[Runnab
 
 def for_paths(paths: Iterable[str]) -> List[Runnable]:
     """Return a specialized list of Runnable objects for paths."""
-    return [for_path(path) for path in paths]
+    runnables = []
+    for path in paths:
+        runnables_for_path = for_path(path)
+        if isinstance(runnables_for_path, list):
+            runnables.extend(runnables_for_path)
+        else:
+            runnables.append(runnables_for_path)
+    return runnables
 
 
 def for_uri(uri: str) -> Runnable:

planemo/runnable_resolve.py

@@ -12,15 +12,15 @@
 )
 
 
-def for_runnable_identifier(ctx, runnable_identifier, kwds, return_all=False):
+def for_runnable_identifier(ctx, runnable_identifier, kwds):
     """Convert URI, path, or alias into Runnable."""
     # could be a URI, path, or alias
     current_profile = kwds.get("profile")
     runnable_identifier = translate_alias(ctx, runnable_identifier, current_profile)
     if not runnable_identifier.startswith(GALAXY_WORKFLOWS_PREFIX):
         runnable_identifier = uri_to_path(ctx, runnable_identifier)
     if os.path.exists(runnable_identifier):
-        runnable = for_path(runnable_identifier, return_all=return_all)
+        runnable = for_path(runnable_identifier)
     else:  # assume galaxy workflow or tool id
         if "/repos/" in runnable_identifier:
             runnable_identifier = f"{GALAXY_TOOLS_PREFIX}{runnable_identifier}"
@@ -34,7 +34,7 @@ def for_runnable_identifiers(ctx, runnable_identifiers, kwds):
     """Convert lists of URIs, paths, and/or aliases into Runnables."""
     runnables = []
     for r in runnable_identifiers:
-        runnable = for_runnable_identifier(ctx, r, kwds, return_all=True)
+        runnable = for_runnable_identifier(ctx, r, kwds)
         if isinstance(runnable, list):
             runnables.extend(runnable)
         else:

planemo/workflow_lint.py

@@ -120,7 +120,7 @@ def structure(path, lint_context):
 
 # misspell for pytest
 def _lint_tsts(path: str, lint_context: WorkflowLintContext) -> None:
-    runnables = for_path(path, return_all=True)
+    runnables = for_path(path)
     if not isinstance(runnables, list):
         runnables = [runnables]
     for runnable in runnables:

tests/test_cmd_autoupdate.py

@@ -2,12 +2,14 @@
 import json
 import os
 import tempfile
+from contextlib import contextmanager
 
 import yaml
 
 from .test_utils import CliTestCase
 
 
+@contextmanager
 def create_tmp_test_tool_file(tool_version):
     """
     Note that to ensure this test is stable, we use packages that have
@@ -25,9 +27,12 @@ def create_tmp_test_tool_file(tool_version):
     </requirements>
 </tool>
     """
-    t = tempfile.NamedTemporaryFile(suffix=".xml", delete=False, mode="w")
-    t.write(xml_str)
-    return t.name
+    with tempfile.TemporaryDirectory() as tempdir, tempfile.NamedTemporaryFile(
+        suffix=".xml", mode="w", dir=tempdir
+    ) as t:
+        t.write(xml_str)
+        t.flush()
+        yield t.name
 
 
 class CmdAutoupdateTestCase(CliTestCase):
@@ -38,17 +43,15 @@ def setUp(self):
 
     def test_autoupdate_dry_run(self):
         """Test autoupdate command with dry run flag."""
-        with self._isolate():
-            xmlfile = create_tmp_test_tool_file("0.6.0")
+        with self._isolate(), create_tmp_test_tool_file("0.6.0") as xmlfile:
             autoupdate_command = ["autoupdate", xmlfile, "--conda_channels", "bioconda", "--dry-run"]
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
             assert f"Update required to {xmlfile}!" in result.output
             assert "Tool main requirement has version 0.6.0, newest conda version is 0.7.3" in result.output
 
     def test_autoupdate(self):
         """Test autoupdate command."""
-        with self._isolate():
-            xmlfile = create_tmp_test_tool_file("0.6.0")
+        with self._isolate(), create_tmp_test_tool_file("0.6.0") as xmlfile:
             autoupdate_command = ["autoupdate", xmlfile, "--conda_channels", "bioconda"]
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
             assert f'Updating {xmlfile.split("/")[-1]} from version 0.6.0 to 0.7.3' in result.output
@@ -57,10 +60,21 @@ def test_autoupdate(self):
                 xmlfile_contents = f.read()
             assert "2017.11.9" in xmlfile_contents
 
+    def test_autoupdate_directory(self):
+        """Test autoupdate command."""
+        with self._isolate(), create_tmp_test_tool_file("0.6.0") as xmlfile:
+            xml_directory = os.path.dirname(xmlfile)
+            autoupdate_command = ["autoupdate", xml_directory, "--conda_channels", "bioconda"]
+            result = self._runner.invoke(self._cli.planemo, autoupdate_command)
+            assert f'Updating {xmlfile.split("/")[-1]} from version 0.6.0 to 0.7.3' in result.output
+            assert f"Tool {xmlfile} successfully updated." in result.output
+            with open(xmlfile) as f:
+                xmlfile_contents = f.read()
+            assert "2017.11.9" in xmlfile_contents
+
     def test_autoupdate_no_update_needed(self):
         """Test autoupdate command when no update is needed."""
-        with self._isolate():
-            xmlfile = create_tmp_test_tool_file("0.7.3")
+        with self._isolate(), create_tmp_test_tool_file("0.7.3") as xmlfile:
             autoupdate_command = ["autoupdate", xmlfile, "--conda_channels", "bioconda"]
             result = self._runner.invoke(self._cli.planemo, autoupdate_command)
             assert f"No updates required or made to {xmlfile}." in result.output