Merge pull request #962 from scitran/rules-enqueue

Use central job enqueue logic for jobs spawned by rules

Author: nagem

api/jobs/gears.py

@@ -124,6 +124,21 @@ def validate_gear_config(gear, config_):
             })
     return True
 
+def fill_gear_default_values(gear, config_):
+    """
+    Given a gear and a config map, fill any missing keys using defaults from the gear's config
+    """
+
+    if config_ is None:
+        config_ = {}
+
+    for k,v in gear['gear'].get('config', {}).iteritems():
+        if 'default' in v:
+            config_.setdefault(k, v['default'])
+
+    return config_
+
+
 def insert_gear(doc):
     gear_tools.validate_manifest(doc['gear'])
 

api/jobs/handlers.py

@@ -335,7 +335,7 @@ def get_config(self, _id):
         # Detect if config is old- or new-style.
         # TODO: remove this logic with a DB upgrade, ref database.py's reserved upgrade section.
 
-        if c.get('config') is not None and c.get('inputs') is not None:
+        if 'config' in c and c.get('inputs') is not None:
             # New behavior
 
             # API keys are only returned in-flight, when the job is running, and not persisted to the job object.

api/jobs/queue.py

@@ -9,7 +9,7 @@
 
 from .. import config
 from .jobs import Job
-from .gears import get_gear, validate_gear_config
+from .gears import get_gear, validate_gear_config, fill_gear_default_values
 from ..validators import InputValidationException
 from ..dao.containerutil import create_filereference_from_dictionary, create_containerreference_from_dictionary, create_containerreference_from_filereference
 
@@ -156,7 +156,7 @@ def enqueue_job(job_map, origin, perm_check_uid=None):
         if gear.get('gear', {}).get('custom', {}).get('flywheel', {}).get('invalid', False):
             raise InputValidationException('Gear marked as invalid, will not run!')
 
-        config_ = job_map.get('config', {})
+        config_ = fill_gear_default_values(gear, job_map.get('config', {}))
         validate_gear_config(gear, config_)
 
         # Translate maps to FileReferences

api/jobs/rules.py

@@ -1,10 +1,12 @@
 import fnmatch
 
 from .. import config
+from ..types import Origin
 from ..dao.containerutil import FileReference
 
 from . import gears
 from .jobs import Job
+from .queue import Queue
 
 log = config.log
 
@@ -235,11 +237,16 @@ def create_jobs(db, container_before, container_after, container_type):
 
 
     spawned_jobs = []
+    origin ={
+        'type': str(Origin.system),
+        'id': None
+    }
 
     for pj in potential_jobs:
-        pj['job'].insert()
-        spawned_jobs.append(pj['rule']['alg'])
+        job_map = pj['job'].map()
+        Queue.enqueue_job(job_map, origin)
 
+        spawned_jobs.append(pj['rule']['alg'])
 
     return spawned_jobs
 

test/unit_tests/python/test_gear_util.py

@@ -0,0 +1,56 @@
+
+import copy
+from api.jobs import gears
+
+# DISCUSS: this basically asserts that the log helper doesn't throw, which is of non-zero but questionable value.
+# Could instead be marked for pytest et. al to ignore coverage? Desirability? Compatibility?
+def test_fill_defaults():
+
+    gear_config = {
+        'key_one':      {'default': 1},
+        'key_two':      {'default': 2},
+        'key_three':    {'default': 3},
+        'key_no_de':    {}
+    }
+
+    gear = {
+        'gear': {
+            'config': gear_config
+        }
+    }
+
+    # test sending in complete config does not change
+    config = {
+        'key_one': 4,
+        'key_two': 5,
+        'key_three': 6
+    }
+
+    result = gears.fill_gear_default_values(gear, config)
+    assert result['key_one'] == 4
+    assert result['key_two'] == 5
+    assert result['key_three'] == 6
+
+    # test sending in empty config
+    result = gears.fill_gear_default_values(gear, {})
+    assert result['key_one'] == 1
+    assert result['key_two'] == 2
+    assert result['key_three'] == 3
+
+    # test sending in None config
+    result = gears.fill_gear_default_values(gear, None)
+    assert result['key_one'] == 1
+    assert result['key_two'] == 2
+    assert result['key_three'] == 3
+
+    # test sending in semi-complete config
+    config = {
+        'key_one': None,
+        'key_two': []
+        #'key_three': 6 # missing
+    }
+
+    result = gears.fill_gear_default_values(gear, config)
+    assert result['key_one'] == None
+    assert result['key_two'] == []
+    assert result['key_three'] == 3