Unit tests and fixes for bugs discovered while testing.

lib/galaxy/jobs/__init__.py

@@ -217,7 +217,10 @@ def __parse_job_conf_xml(self, tree):
         handlers_conf = root.find('handlers')
         self._init_handler_assignment_methods(handlers_conf)
         self._init_handlers(handlers_conf)
-        self._set_default_handler_assignment_methods()
+        if not self.handler_assignment_methods_configured:
+            self._set_default_handler_assignment_methods()
+        else:
+            self.app.application_stack.init_job_handling(self)
         log.info("Job handler assignment methods set to: %s", ', '.join(self.handler_assignment_methods))
         for tag, handlers in [(t, h) for t, h in self.handlers.items() if isinstance(h, list)]:
             log.info("Tag [%s] handlers: %s", tag, ', '.join(handlers))
@@ -341,7 +344,10 @@ def __set_default_job_conf(self):
         # Set the handlers
         self._init_handler_assignment_methods()
         self._init_handlers()
-        self._set_default_handler_assignment_methods()
+        if not self.handler_assignment_methods_configured:
+            self._set_default_handler_assignment_methods()
+        else:
+            self.app.application_stack.init_job_handling(self)
         # Set the destination
         self.default_destination_id = 'local'
         self.destinations['local'] = [JobDestination(id='local', runner='local')]

lib/galaxy/util/facts.py

@@ -10,6 +10,7 @@ class Facts(MutableMapping):
     """A dict-like object that evaluates values at access time."""
 
     def __init__(self, config=None, **kwargs):
+        config = config or {}
         self.__dict__ = {}
         self.__set_defaults(config)
         self.__set_config(config)

lib/galaxy/util/handlers.py

@@ -92,11 +92,6 @@ def _set_default_handler_assignment_methods(self):
             if not self.app.config.track_jobs_in_database and \
                     HANDLER_ASSIGNMENT_METHODS.MEM_SELF not in self.UNSUPPORTED_HANDLER_ASSIGNMENT_METHODS:
                 # DEPRECATED: You should just set mem_self as the only method if you want this
-                # FIXME: does setting MEM_SELF still disable track_jobs_in_database like it used to?
-                # FIXME: test MEM_SELF via all config routes:
-                #  1. no assign config, track_jobs_in_databse = False in config
-                #  2. assign_with='mem-self', track_jobs_in_databse = False in config
-                #  2. assign_with='mem-self', track_jobs_in_databse = True in config
                 log.warning("The `track_jobs_in_database` option is deprecated, please set `%s` as the job"
                             " handler assignment method in the job handler configuration",
                             HANDLER_ASSIGNMENT_METHODS.MEM_SELF)
@@ -262,7 +257,6 @@ def _assign_db_preassign_handler(self, obj, method, configured, index=None, **kw
 
         :returns: str -- A valid job handler ID.
         """
-        # FIXME: test a combo of pool handlers and defined handlers w/ DB_PREASSIGN as the only method
         handler = configured
         if handler is None:
             handler = self.default_handler_id or self.DEFAULT_HANDLER_TAG

lib/galaxy/web/stack/__init__.py

@@ -7,9 +7,8 @@
 import logging
 import os
 
-# The uwsgi module is automatically injected by the parent uwsgi
-# process and only exists that way.  If anything works, this is a
-# uwsgi-managed process.
+# The uwsgi module is automatically injected by the parent uwsgi process and only exists that way.  If anything works,
+# this is a uwsgi-managed process.
 try:
     import uwsgi
 except ImportError:
@@ -348,7 +347,6 @@ def _init_job_handler_subpools(self, job_config, pool):
             # Pools are hierarchical (so that you can have e.g. workflow schedulers use the job handlers pool if no
             # workflow schedulers pool exists), so if a pool for a given tag has already been found higher in the
             # hierarchy, don't add mules from a farm/pool lower in the hierarchy.
-            # FIXME: test
             if tag not in job_config.pool_for_tag:
                 if self.in_pool(farm):
                     job_config.is_handler = True

lib/galaxy/workflow/scheduling_manager.py

@@ -153,7 +153,6 @@ def queue(self, workflow_invocation, request_params):
         try:
             self._assign_handler(workflow_invocation)
         except HandlerAssignmentError:
-            # FIXME: test, raise new exception?
             raise RuntimeError("Unable to set a handler for workflow invocation '%s'" % workflow_invocation.id)
 
         return workflow_invocation
@@ -234,7 +233,10 @@ def __init_handlers(self, config_element):
 
     def __init_handler_assignment_methods(self, config_element=None):
         self._init_handler_assignment_methods(config_element)
-        self._set_default_handler_assignment_methods()
+        if not self.handler_assignment_methods_configured:
+            self._set_default_handler_assignment_methods()
+        else:
+            self.app.application_stack.init_job_handling(self)
         log.info("Workflow scheduling handler assignment method(s): %s", ', '.join(self.handler_assignment_methods))
         for tag, handlers in [(t, h) for t, h in self.handlers.items() if isinstance(h, list)]:
             log.info("Tag [%s] handlers: %s", tag, ', '.join(handlers))

test/unit/jobs/handler_template_job_conf.xml

@@ -0,0 +1,12 @@
+<?xml version="1.0"?>
+<job_conf>
+    <plugins>
+        <plugin id="local" type="runner" load="galaxy.jobs.runners.local:LocalJobRunner" workers="4"/>
+    </plugins>
+    <handlers{assign_with}>
+        {handlers}
+    </handlers>
+    <destinations>
+        <destination id="local" runner="local"/>
+    </destinations>
+</job_conf>

test/unit/jobs/test_job_configuration.py

@@ -4,15 +4,18 @@
 import tempfile
 import unittest
 
+import mock
+
 from galaxy.jobs import JobConfiguration
 from galaxy.util import bunch
-from galaxy.web.stack import ApplicationStack
+from galaxy.web.stack import ApplicationStack, UWSGIApplicationStack
 
 # File would be slightly more readable if contents were embedded directly, but
 # there are advantages to testing the documentation/examples.
 SIMPLE_JOB_CONF = os.path.join(os.path.dirname(__file__), "..", "..", "..", "config", "job_conf.xml.sample_basic")
 ADVANCED_JOB_CONF = os.path.join(os.path.dirname(__file__), "..", "..", "..", "config", "job_conf.xml.sample_advanced")
 CONDITIONAL_RUNNER_JOB_CONF = os.path.join(os.path.dirname(__file__), "conditional_runners_job_conf.xml")
+HANDLER_TEMPLATE_JOB_CONF = os.path.join(os.path.dirname(__file__), "handler_template_job_conf.xml")
 
 
 class JobConfXmlParserTestCase(unittest.TestCase):
@@ -29,8 +32,11 @@ def setUp(self):
             server_name="main",
         )
         self.__write_config_from(SIMPLE_JOB_CONF)
-        self.app = bunch.Bunch(config=self.config, job_metrics=MockJobMetrics(), application_stack=ApplicationStack())
+        self.__app = None
+        self.__application_stack = None
         self.__job_configuration = None
+        self.__job_configuration_base_pools = None
+        self.__uwsgi_opt = None
 
     def tearDown(self):
         shutil.rmtree(self.temp_directory)
@@ -51,11 +57,6 @@ def test_configuration_of_tasks(self):
         assert len(task_runners) == 1
         assert task_runners[0]["workers"] == 5
 
-    def test_if_no_handlers_implict_db_self(self):
-        assert self.job_config.default_handler_id is None
-        assert self.job_config.handlers == {}
-        assert self.job_config.handler_assignment_methods == ['db-self']
-
     def test_explicit_handler_default(self):
         self.__with_advanced_config()
         assert self.job_config.default_handler_id == "handlers"
@@ -65,6 +66,69 @@ def test_handler_tag_parsing(self):
         assert "handler0" in self.job_config.handlers["handlers"]
         assert "handler1" in self.job_config.handlers["handlers"]
 
+    def test_implict_db_self_handler_assign(self):
+        assert self.job_config.handler_assignment_methods == ['db-self']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers == {}
+
+    def test_implicit_db_assign_handler_assign_with_explicit_handlers(self):
+        self.__with_handlers_config(handlers=[{'id': 'handler0'}, {'id': 'handler1'}])
+        assert self.job_config.handler_assignment_methods == ['db-preassign']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['handler0', 'handler1']
+
+    def test_implict_uwsgi_mule_message_handler_assign(self):
+        self.__with_uwsgi_application_stack(mule='lib/galaxy/main.py', farm='job-handlers:1')
+        assert self.job_config.handler_assignment_methods == ['uwsgi-mule-message']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['main.job-handlers.1']
+
+    def test_implict_uwsgi_mule_message_handler_assign_with_explicit_handlers(self):
+        self.__with_handlers_config(handlers=[{'id': 'handler0'}, {'id': 'handler1'}])
+        self.__with_uwsgi_application_stack(mule='lib/galaxy/main.py', farm='job-handlers:1')
+        assert self.job_config.handler_assignment_methods == ['uwsgi-mule-message', 'db-preassign']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['handler0', 'handler1', 'main.job-handlers.1']
+
+    def test_explicit_mem_self_handler_assign(self):
+        self.__with_handlers_config(assign_with='mem-self')
+        assert self.job_config.handler_assignment_methods == ['mem-self']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers == {}
+        assert not self.config.track_jobs_in_database
+
+    def test_explicit_mem_self_handler_assign_with_handlers(self):
+        self.__with_handlers_config(assign_with='mem-self', handlers=[{'id': 'handler0'}])
+        assert self.job_config.handler_assignment_methods == ['mem-self']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['handler0']
+
+    def test_explicit_db_preassign_handler_assign_with_uwsgi(self):
+        self.__with_handlers_config(assign_with='db-preassign', handlers=[{'id': 'handler0'}])
+        self.__with_uwsgi_application_stack(mule='lib/galaxy/main.py', farm='job-handlers:1')
+        assert self.job_config.handler_assignment_methods == ['db-preassign']
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['handler0', 'main.job-handlers.1']
+
+    def test_uwsgi_farms_as_handler_tags(self):
+        self.__with_uwsgi_application_stack(
+            mule=['lib/galaxy/main.py'] * 2,
+            farm=['job-handlers:1', 'job-handlers.foo:2']
+        )
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['main.job-handlers.1']
+        assert self.job_config.handlers['foo'] == ['main.job-handlers.foo.1']
+
+    def test_uwsgi_overlapping_pools(self):
+        self.__with_handlers_config(base_pools=('workflow-schedulers', 'job-handlers'))
+        self.__with_uwsgi_application_stack(
+            mule=['lib/galaxy/main.py'] * 3,
+            farm=['job-handlers:1', 'workflow-schedulers:2', 'job-handlers.foo:3']
+        )
+        assert self.job_config.default_handler_id is None
+        assert self.job_config.handlers['_default_'] == ['main.workflow-schedulers.1']
+        assert self.job_config.handlers['foo'] == ['main.job-handlers.foo.1']
+
     def test_load_simple_destination(self):
         local_dest = self.job_config.destinations["local"][0]
         assert local_dest.id == "local"
@@ -165,17 +229,57 @@ def test_conditional_runners_from_environ(self):
 
     # TODO: Add job metrics parsing test.
 
+    @property
+    def app(self):
+        if not self.__app:
+            self.__app = bunch.Bunch(
+                config=self.config,
+                job_metrics=MockJobMetrics(),
+                application_stack=self.application_stack
+            )
+        return self.__app
+
+    @property
+    def application_stack(self):
+        if not self.__application_stack:
+            self.__application_stack = ApplicationStack()
+        return self.__application_stack
+
     @property
     def job_config(self):
         if not self.__job_configuration:
-            self.__job_configuration = JobConfiguration(self.app)
+            base_handler_pools = self.__job_configuration_base_pools or JobConfiguration.DEFAULT_BASE_HANDLER_POOLS
+            mock_uwsgi = mock.Mock()
+            mock_uwsgi.mule_id = lambda: 1
+            with mock.patch('galaxy.web.stack.uwsgi', mock_uwsgi), \
+                    mock.patch('galaxy.web.stack.uwsgi.opt', self.__uwsgi_opt), \
+                    mock.patch('galaxy.jobs.JobConfiguration.DEFAULT_BASE_HANDLER_POOLS', base_handler_pools):
+                self.__job_configuration = JobConfiguration(self.app)
         return self.__job_configuration
 
+    def __with_uwsgi_application_stack(self, **uwsgi_opt):
+        self.__uwsgi_opt = uwsgi_opt
+        self.__application_stack = UWSGIApplicationStack()
+
     def __with_advanced_config(self):
         self.__write_config_from(ADVANCED_JOB_CONF)
 
-    def __write_config_from(self, path):
-        self.__write_config(open(path, "r").read())
+    def __with_handlers_config(self, assign_with=None, handlers=None, base_pools=None):
+        handlers = handlers or []
+        template = {
+            'assign_with': ' assign_with="%s"' % assign_with if assign_with is not None else '',
+            'handlers': '\n'.join([
+                '<handler id="{id}"/>'.format(
+                    id=x['id'],
+                    tags=' tags="%s"' % x['tags'] if 'tags' in x else ''
+                ) for x in handlers]),
+        }
+        self.__job_configuration_base_pools = base_pools
+        self.__write_config_from(HANDLER_TEMPLATE_JOB_CONF, template=template)
+
+    def __write_config_from(self, path, template=None):
+        template = template or {}
+        self.__write_config(open(path, "r").read().format(**template))
 
     def __write_config(self, contents):
         with open(os.path.join(self.temp_directory, "job_conf.xml"), "w") as f: