Properly serialize task kwargs in a nongirder context. Fixes #337

girder_worker/context/nongirder_context.py

@@ -49,7 +49,7 @@ def create_task_job(job_defaults, sender=None, body=None,
             'public': headers.pop('girder_job_public',
                                   job_defaults.get('girder_job_public', '')),
             'args': json.dumps(task_args),
-            'kwargs': task_kwargs,
+            'kwargs': json.dumps(task_kwargs),
             'otherFields': json.dumps(
                 dict(celeryTaskId=headers['id'],
                      celeryParentTaskId=parent_task.request.id,

tests/integration/integration_test_endpoints/server/raw.py

@@ -40,6 +40,8 @@ def __init__(self):
                    self.test_celery_task_revoke_in_queue)
         self.route('POST', ('test_task_chained', ),
                    self.test_celery_task_chained)
+        self.route('POST', ('test_task_chained_with_kwargs', ),
+                   self.test_celery_task_chained_with_kwargs)
         self.route('POST', ('test_task_chained_bad_token_fails', ),
                    self.test_celery_task_chained_bad_token_fails)
 
@@ -212,6 +214,27 @@ def test_celery_task_chained(self, params):
 
         return [job_1, job_2, job_3]
 
+    @access.token
+    @filtermodel(model='job', plugin='jobs')
+    @describeRoute(
+        Description('Test chained celery tasks with kwargs')
+    )
+    def test_celery_task_chained_with_kwargs(self, params):
+        jobModel = ModelImporter.model('job', 'jobs')
+        user = self.getCurrentUser()
+        # F(F(F(6))) --> F(F(8)) --> F(21) --> 10946
+        result = (
+            fibonacci.s(6, foo=1, bar=2) |
+            fibonacci.s(baz=3) |
+            fibonacci.s(last_but_not_least=4)
+        ).delay()
+        result.wait(timeout=10)
+        job_1 = result.job
+        job_2 = jobModel.load(job_1['parentId'], user=user)
+        job_3 = jobModel.load(job_2['parentId'], user=user)
+
+        return [job_1, job_2, job_3]
+
     @access.token
     @filtermodel(model='job', plugin='jobs')
     @describeRoute(

tests/integration/test_celery.py

@@ -128,6 +128,17 @@ def test_celery_chained_tasks(session):
     assert jobs[0]['parentId'] == jobs[1]['_id']
 
 
+def test_celery_chained_tasks_with_kwargs(session):
+    url = 'integration_tests/celery/test_task_chained_with_kwargs'
+    r = session.post(url)
+    assert r.status_code == 200, r.content
+    jobs = r.json()
+    # Check if kwargs are correctly set
+    assert jobs[2]['kwargs'] == dict(foo=1, bar=2)
+    assert jobs[1]['kwargs'] == dict(baz=3)
+    assert jobs[0]['kwargs'] == dict(last_but_not_least=4)
+
+
 def test_celery_chained_task_bad_token_fails(session):
     r = session.post('integration_tests/celery/test_task_chained_bad_token_fails')
     assert r.status_code == 200, r.content