Implement external Galaxy engine.

This work is focused on "test" and "run" commands for Galaxy tools and workflows, though it should enable other combinations of artifacts and commands such as serving workflows and various CWL artifact operations against that fork of Galaxy.

The tool piece of this requires an unreleased version of galaxy-lib (galaxyproject/galaxy-lib#91) and unmerged Galaxy modifications to expose a APIs for external tool testing (galaxyproject/galaxy#5545). Workflows are oddly enough likely closer to working with this WIP.

Implements #592.
Implements #508.

Author: jmchilton

planemo/commands/cmd_test.py

@@ -65,9 +65,9 @@ def cli(ctx, paths, **kwds):
     """
     runnables = for_paths(paths)
     enable_beta_test = any([r.type not in [RunnableType.galaxy_tool, RunnableType.directory] for r in runnables])
-    enable_beta_test = enable_beta_test or not is_galaxy_engine(**kwds)
+    enable_beta_test = enable_beta_test or kwds.get("engine", "galaxy") != "engine"
     if enable_beta_test:
-        info("Enable beta testing mode to test artifact that isn't a Galaxy tool.")
+        info("Enable beta testing mode for testing.")
         with engine_context(ctx, **kwds) as engine:
             test_data = engine.test(runnables)
             return_value = handle_reports_and_summary(ctx, test_data.structured_data, kwds=kwds)

planemo/engine/factory.py

@@ -3,25 +3,31 @@
 import contextlib
 
 from .cwltool import CwlToolEngine
-from .galaxy import DockerizedGalaxyEngine
-from .galaxy import GalaxyEngine
+from .galaxy import (
+    DockerizedManagedGalaxyEngine,
+    ExternalGalaxyEngine,
+    LocalManagedGalaxyEngine,
+)
+
 
 UNKNOWN_ENGINE_TYPE_MESSAGE = "Unknown engine type specified [%s]."
 
 
 def is_galaxy_engine(**kwds):
     """Return True iff the engine configured is :class:`GalaxyEngine`."""
     engine_type_str = kwds.get("engine", "galaxy")
-    return engine_type_str in ["galaxy", "docker_galaxy"]
+    return engine_type_str in ["galaxy", "docker_galaxy", "external_galaxy"]
 
 
 def build_engine(ctx, **kwds):
     """Build an engine from the supplied planemo configuration."""
     engine_type_str = kwds.get("engine", "galaxy")
     if engine_type_str == "galaxy":
-        engine_type = GalaxyEngine
+        engine_type = LocalManagedGalaxyEngine
     elif engine_type_str == "docker_galaxy":
-        engine_type = DockerizedGalaxyEngine
+        engine_type = DockerizedManagedGalaxyEngine
+    elif engine_type_str == "external_galaxy":
+        engine_type = ExternalGalaxyEngine
     elif engine_type_str == "cwltool":
         engine_type = CwlToolEngine
     else:

planemo/engine/galaxy.py

@@ -1,19 +1,26 @@
 """Module contianing the :class:`GalaxyEngine` implementation of :class:`Engine`."""
+from __future__ import absolute_import
 
+import abc
 import contextlib
 
+from galaxy.tools.verify import interactor
+
 from planemo.galaxy.activity import execute
+from planemo.galaxy.config import external_galaxy_config
 from planemo.galaxy.serve import serve_daemon
 from planemo.runnable import RunnableType
 from .interface import BaseEngine
 
 
 class GalaxyEngine(BaseEngine):
-    """An :class:`Engine` implementation backed by Galaxy.
+    """An :class:`Engine` implementation backed by a managed Galaxy.
 
     More information on Galaxy can be found at http://galaxyproject.org/.
     """
 
+    __metaclass__ = abc.ABCMeta
+
     handled_runnable_types = [
         RunnableType.cwl_tool,
         RunnableType.cwl_workflow,
@@ -24,14 +31,71 @@ class GalaxyEngine(BaseEngine):
     def _run(self, runnable, job_path):
         """Run CWL job in Galaxy."""
         self._ctx.vlog("Serving artifact [%s] with Galaxy." % (runnable,))
-        with self.serve_runnables([runnable]) as config:
+        with self.ensure_runnables_served([runnable]) as config:
             self._ctx.vlog("Running job path [%s]" % job_path)
             run_response = execute(self._ctx, config, runnable, job_path, **self._kwds)
 
         return run_response
 
+    @abc.abstractmethod
+    def ensure_runnables_served(self, runnables):
+        """Use a context manager and describe Galaxy instance with runnables being served."""
+
+    def _run_test_case(self, test_case):
+        if hasattr(test_case, "job_path"):
+            # Simple file-based job path.
+            super(GalaxyEngine, self)._run_test_case(test_case)
+        else:
+            with self.ensure_runnables_served([test_case.runnable]) as config:
+                galaxy_interactor_kwds = {
+                    "galaxy_url": config.galaxy_url,
+                    "master_api_key": config.master_api_key,
+                    "api_key": config.user_api_key,
+                    "keep_outputs_dir": "",  # TODO: this...
+                }
+                tool_id = test_case.tool_id
+                test_index = test_case.test_index
+                tool_version = test_case.tool_version
+                galaxy_interactor = interactor.GalaxyInteractorApi(**galaxy_interactor_kwds)
+
+                test_results = []
+
+                def _register_job_data(job_data):
+                    test_results.append({
+                        'id': tool_id + "-" + str(test_index),
+                        'has_data': True,
+                        'data': job_data,
+                    })
+
+                verbose = self._ctx.verbose
+                try:
+                    if verbose:
+                        # TODO: this is pretty hacky, it'd be better to send a stream
+                        # and capture the output information somehow.
+                        interactor.VERBOSE_GALAXY_ERRORS = True
+
+                    interactor.verify_tool(
+                        tool_id,
+                        galaxy_interactor,
+                        test_index=test_index,
+                        tool_version=tool_version,
+                        register_job_data=_register_job_data,
+                        quiet=not verbose,
+                    )
+                except Exception:
+                    pass
+
+                return test_results[0]
+
+
+class LocalManagedGalaxyEngine(BaseEngine):
+    """An :class:`Engine` implementation backed by a managed Galaxy.
+
+    More information on Galaxy can be found at http://galaxyproject.org/.
+    """
+
     @contextlib.contextmanager
-    def serve_runnables(self, runnables):
+    def ensure_runnables_served(self, runnables):
         # TODO: define an interface for this - not everything in config would make sense for a
         # pre-existing Galaxy interface.
         with serve_daemon(self._ctx, runnables, **self._serve_kwds()) as config:
@@ -41,7 +105,7 @@ def _serve_kwds(self):
         return self._kwds.copy()
 
 
-class DockerizedGalaxyEngine(GalaxyEngine):
+class DockerizedManagedGalaxyEngine(LocalManagedGalaxyEngine):
     """An :class:`Engine` implementation backed by Galaxy running in Docker.
 
     More information on Galaxy can be found at http://galaxyproject.org/.
@@ -53,7 +117,20 @@ def _serve_kwds(self):
         return serve_kwds
 
 
+class ExternalGalaxyEngine(GalaxyEngine):
+    """An :class:`Engine` implementation backed by an external Galaxy instance.
+    """
+
+    @contextlib.contextmanager
+    def ensure_runnables_served(self, runnables):
+        # TODO: ensure tools are available
+        with external_galaxy_config(self._ctx, runnables, **self._kwds) as config:
+            config.install_workflows()
+            yield config
+
+
 __all__ = (
-    "GalaxyEngine",
-    "DockerizedGalaxyEngine",
+    "DockerizedManagedGalaxyEngine",
+    "ExternalGalaxyEngine",
+    "LocalManagedGalaxyEngine",
 )

planemo/engine/interface.py

@@ -95,27 +95,7 @@ def _collect_test_results(self, test_cases):
             self._ctx.vlog(
                 "Running tests %s" % test_case
             )
-            runnable = test_case.runnable
-            job_path = test_case.job_path
-            tmp_path = None
-            if job_path is None:
-                job = test_case.job
-                f = tempfile.NamedTemporaryFile(
-                    dir=test_case.tests_directory,
-                    suffix=".json",
-                    prefix="plnmotmptestjob",
-                    delete=False,
-                    mode="w+",
-                )
-                tmp_path = f.name
-                job_path = tmp_path
-                json.dump(job, f)
-                f.close()
-            try:
-                run_response = self._run(runnable, job_path)
-            finally:
-                if tmp_path:
-                    os.remove(tmp_path)
+            run_response = self._run_test_case(test_case)
             self._ctx.vlog(
                 "Test case [%s] resulted in run response [%s]",
                 test_case,
@@ -124,6 +104,30 @@ def _collect_test_results(self, test_cases):
             test_results.append((test_case, run_response))
         return test_results
 
+    def _run_test_case(self, test_case):
+        runnable = test_case.runnable
+        job_path = test_case.job_path
+        tmp_path = None
+        if job_path is None:
+            job = test_case.job
+            f = tempfile.NamedTemporaryFile(
+                dir=test_case.tests_directory,
+                suffix=".json",
+                prefix="plnmotmptestjob",
+                delete=False,
+                mode="w+",
+            )
+            tmp_path = f.name
+            job_path = tmp_path
+            json.dump(job, f)
+            f.close()
+        try:
+            run_response = self._run(runnable, job_path)
+        finally:
+            if tmp_path:
+                os.remove(tmp_path)
+        return run_response
+
     def _process_test_results(self, test_results):
         for (test_case, run_response) in test_results:
             pass

planemo/galaxy/api.py

@@ -7,13 +7,18 @@
 DEFAULT_MASTER_API_KEY = "test_key"
 
 
-def gi(port, key=None):
+def gi(port=None, url=None, key=None):
     """Return a bioblend ``GalaxyInstance`` for Galaxy on this port."""
     ensure_module()
     if key is None:
         key = DEFAULT_MASTER_API_KEY
+    if port is None:
+        url = url
+    else:
+        url = "http://localhost:%d" % int(port)
+
     return galaxy.GalaxyInstance(
-        url="http://localhost:%d" % int(port),
+        url=url,
         key=key
     )