Merge pull request #4553 from rtfd/humitos/celery/queue-routing

Route task to proper queue
readthedocs · Aug 22, 2018 · 4d8571a · 4d8571a
2 parents 5cefcb8 + 08a5a91
commit 4d8571a
Show file tree

Hide file tree

Showing 2 changed files with 84 additions and 38 deletions.
diff --git a/readthedocs/core/utils/__init__.py b/readthedocs/core/utils/__init__.py
@@ -139,13 +139,12 @@ def prepare_build(
     options['time_limit'] = int(time_limit * 1.2)
 
     update_docs_task = UpdateDocsTask()
-
-    # Py 2.7 doesn't support ``**`` expand syntax twice. We create just one big
-    # kwargs (including the options) for this and expand it just once.
-    # return update_docs_task.si(project.pk, **kwargs, **options)
-    kwargs.update(options)
-
-    return update_docs_task.si(project.pk, **kwargs)
+    return update_docs_task.signature(
+        (project.pk,),
+        kwargs=kwargs,
+        options=options,
+        immutable=True,
+    )
 
 
 def trigger_build(project, version=None, record=True, force=False):

diff --git a/readthedocs/rtd_tests/tests/test_core_utils.py b/readthedocs/rtd_tests/tests/test_core_utils.py
@@ -1,3 +1,4 @@
+# -*- coding: utf-8 -*-
 """Test core util functions"""
 
 from __future__ import absolute_import
@@ -17,62 +18,108 @@ def setUp(self):
         self.project = get(Project, container_time_limit=None)
         self.version = get(Version, project=self.project)
 
+    @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
+    def test_trigger_custom_queue(self, update_docs):
+        """Use a custom queue when routing the task"""
+        self.project.build_queue = 'build03'
+        trigger_build(project=self.project, version=self.version)
+        kwargs = {
+            'version_pk': self.version.pk,
+            'record': True,
+            'force': False,
+            'build_pk': mock.ANY,
+        }
+        options = {
+            'queue': 'build03',
+            'time_limit': 720,
+            'soft_time_limit': 600,
+        }
+        update_docs().signature.assert_has_calls([
+            mock.call(
+                (self.project.pk,),
+                kwargs=kwargs,
+                options=options,
+                immutable=True,
+            ),
+        ])
+        update_docs().signature().apply_async.assert_called()
+
     @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
     def test_trigger_build_time_limit(self, update_docs):
         """Pass of time limit"""
         trigger_build(project=self.project, version=self.version)
-        update_docs().si.assert_has_calls([
+        kwargs = {
+            'version_pk': self.version.pk,
+            'record': True,
+            'force': False,
+            'build_pk': mock.ANY,
+        }
+        options = {
+            'queue': mock.ANY,
+            'time_limit': 720,
+            'soft_time_limit': 600,
+        }
+        update_docs().signature.assert_has_calls([
             mock.call(
-                self.project.pk,
-                time_limit=720,
-                soft_time_limit=600,
-                queue=mock.ANY,
-                force=False,
-                record=True,
-                build_pk=mock.ANY,
-                version_pk=self.version.id,
+                (self.project.pk,),
+                kwargs=kwargs,
+                options=options,
+                immutable=True,
             ),
         ])
-        update_docs().si().apply_async.assert_called()
+        update_docs().signature().apply_async.assert_called()
 
     @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
     def test_trigger_build_invalid_time_limit(self, update_docs):
         """Time limit as string"""
         self.project.container_time_limit = '200s'
         trigger_build(project=self.project, version=self.version)
-        update_docs().si.assert_has_calls([
+        kwargs = {
+            'version_pk': self.version.pk,
+            'record': True,
+            'force': False,
+            'build_pk': mock.ANY,
+        }
+        options = {
+            'queue': mock.ANY,
+            'time_limit': 720,
+            'soft_time_limit': 600,
+        }
+        update_docs().signature.assert_has_calls([
             mock.call(
-                self.project.pk,
-                time_limit=720,
-                soft_time_limit=600,
-                queue=mock.ANY,
-                force=False,
-                record=True,
-                build_pk=mock.ANY,
-                version_pk=self.version.id,
+                (self.project.pk,),
+                kwargs=kwargs,
+                options=options,
+                immutable=True,
             ),
         ])
-        update_docs().si().apply_async.assert_called()
+        update_docs().signature().apply_async.assert_called()
 
     @mock.patch('readthedocs.projects.tasks.UpdateDocsTask')
     def test_trigger_build_rounded_time_limit(self, update_docs):
         """Time limit should round down"""
         self.project.container_time_limit = 3
         trigger_build(project=self.project, version=self.version)
-        update_docs().si.assert_has_calls([
+        kwargs = {
+            'version_pk': self.version.pk,
+            'record': True,
+            'force': False,
+            'build_pk': mock.ANY,
+        }
+        options = {
+            'queue': mock.ANY,
+            'time_limit': 3,
+            'soft_time_limit': 3,
+        }
+        update_docs().signature.assert_has_calls([
             mock.call(
-                self.project.pk,
-                time_limit=3,
-                soft_time_limit=3,
-                queue=mock.ANY,
-                force=False,
-                record=True,
-                build_pk=mock.ANY,
-                version_pk=self.version.id,
+                (self.project.pk,),
+                kwargs=kwargs,
+                options=options,
+                immutable=True,
             ),
         ])
-        update_docs().si().apply_async.assert_called()
-
+        update_docs().signature().apply_async.assert_called()
 
     def test_slugify(self):
         """Test additional slugify"""