[WIP] 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 galaxyproject#592.
Implements galaxyproject#508.

Author: jmchilton

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,25 @@
 """Module contianing the :class:`GalaxyEngine` implementation of :class:`Engine`."""
 
+import abc
 import contextlib
 
+from galaxy.tools.verify.interactor import GalaxyInteractorApi, verify_tool
+
 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 +30,63 @@ 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 = 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,
+                    })
+
+                try:
+                    verify_tool(
+                        tool_id, galaxy_interactor, test_index=test_index, tool_version=tool_version, register_job_data=_register_job_data
+                    )
+                finally:
+                    pass
+
+                # TODO: record pass vs fail
+                # TODO: record timing
+                # TODO: return run_response compatible interface maybe?
+                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 +96,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 +108,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,26 +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,
-                )
-                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,
@@ -123,6 +104,29 @@ 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,
+            )
+            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
     )
 

planemo/galaxy/config.py

@@ -246,8 +246,12 @@ class = StreamHandler
 @contextlib.contextmanager
 def galaxy_config(ctx, runnables, **kwds):
     """Set up a ``GalaxyConfig`` in an auto-cleaned context."""
-    dockerize = kwds.get("dockerize", False)
-    c = docker_galaxy_config if dockerize else local_galaxy_config
+    c = local_galaxy_config
+    if kwds.get("dockerize", False):
+        c = docker_galaxy_config
+    elif kwds.get("external", False):
+        c = external_galaxy_config
+
     with c(ctx, runnables, **kwds) as config:
         yield config
 
@@ -553,6 +557,16 @@ def _shared_galaxy_properties(config_directory, kwds, for_tests):
     return properties
 
 
+@contextlib.contextmanager
+def external_galaxy_config(ctx, runnables, for_tests=False, **kwds):
+    yield BaseGalaxyConfig(
+        galaxy_url=kwds.get("external_galaxy_url", None),
+        master_api_key=_get_master_api_key(kwds),
+        user_api_key=kwds.get("user_api_key", None),
+        runnables=runnables,
+    )
+
+
 def _get_master_api_key(kwds):
     master_api_key = kwds.get("master_api_key", DEFAULT_MASTER_API_KEY)
     return master_api_key
@@ -647,59 +661,24 @@ def startup_command(self, ctx, **kwds):
     def log_contents(self):
         """Retrieve text of log for running Galaxy instance."""
 
-    @abc.abstractproperty
-    def gi(self):
-        """Return an admin bioblend Galaxy instance for API interactions."""
-
-    @abc.abstractproperty
-    def user_gi(self):
-        """Return a user-backed bioblend Galaxy instance for API interactions."""
-
-    @abc.abstractmethod
-    def install_repo(self, *args, **kwds):
-        """Install specified tool shed repository."""
-
-    @abc.abstractproperty
-    def tool_shed_client(self):
-        """Return a admin bioblend tool shed client."""
-
-    @abc.abstractmethod
-    def wait_for_all_installed(self):
-        """Wait for all queued up repositories installs to complete."""
-
-    @abc.abstractmethod
-    def install_workflows(self):
-        """Install all workflows configured with these planemo arguments."""
-
-    @abc.abstractmethod
-    def workflow_id(self, path):
-        """Get installed workflow API ID for input path."""
-
     @abc.abstractmethod
     def cleanup(self):
         """Cleanup allocated resources to run this instance."""
 
 
-class BaseGalaxyConfig(GalaxyConfig):
+class BaseGalaxyConfig(GalaxyInterface):
 
     def __init__(
         self,
-        config_directory,
-        env,
-        test_data_dir,
-        port,
-        server_name,
+        galaxy_url,
         master_api_key,
+        user_api_key,
         runnables,
     ):
-        self.config_directory = config_directory
-        self.env = env
-        self.test_data_dir = test_data_dir
-        self.port = port
-        self.server_name = server_name
+        self.galaxy_url = galaxy_url
         self.master_api_key = master_api_key
+        self._user_api_key = user_api_key
         self.runnables = runnables
-        self._user_api_key = None
         self._workflow_ids = {}
 
     @property
@@ -710,6 +689,7 @@ def gi(self):
     def user_gi(self):
         # TODO: thread-safe
         if self._user_api_key is None:
+            # TODO: respect --galaxy_email - seems like a real bug
             self._user_api_key = user_api_key(self.gi)
 
         return self._gi_for_key(self._user_api_key)
@@ -761,6 +741,31 @@ def workflow_id(self, path):
         return self._workflow_ids[path]
 
 
+class BaseManagedGalaxyConfig(GalaxyConfig):
+
+    def __init__(
+        self,
+        config_directory,
+        env,
+        test_data_dir,
+        port,
+        server_name,
+        master_api_key,
+        runnables,
+    ):
+        galaxy_url = "http://localhost:%d" % port
+        super(BaseManagedGalaxyConfig, self).__(
+            galaxy_url=galaxy_url,
+            master_api_key=master_api_key,
+            user_api_key=None,
+        )
+        self.config_directory = config_directory
+        self.env = env
+        self.test_data_dir = test_data_dir
+        self.port = port
+        self.server_name = server_name
+
+
 class DockerGalaxyConfig(BaseGalaxyConfig):
     """A :class:`GalaxyConfig` description of a Dockerized Galaxy instance."""