ci/tasks.py: offload testjob post processing to its own task

squad/ci/models.py

@@ -143,17 +143,18 @@ def fetch(self, job_id):
         except DuplicatedTestJob as exception:
             logger.error('Failed to fetch test_job(%d): "%s"' % (test_job.id, str(exception)))
 
-        if test_job.testrun:
-            self.__postprocess_testjob__(test_job)
-
-        # Remove the 'Fetching' job_status only after eventual plugins
-        # are finished, this garantees extra tests and metadata to
-        # be in SQUAD before the build is considered finished
-        test_job.job_status = status
-        test_job.save()
+        if test_job.testrun and test_job.target.enabled_plugins and any(test_job.target.enabled_plugins):
+            # Avoids cyclic import errors
+            from squad.ci.tasks import postprocess_testjob
+            test_job.save()
+            postprocess_testjob.delay(test_job.id, status)
+        else:
+            # Remove the 'Fetching' job_status only after all work is done
+            test_job.job_status = status
+            test_job.save()
 
-        if test_job.testrun:
-            UpdateProjectStatus()(test_job.testrun)
+            if test_job.testrun:
+                UpdateProjectStatus()(test_job.testrun)
 
     def __postprocess_testjob__(self, test_job):
         project = test_job.target

squad/ci/tasks.py

@@ -1,4 +1,5 @@
 from squad.celery import app as celery
+from squad.core.tasks import UpdateProjectStatus
 from squad.ci.models import Backend, TestJob
 from squad.ci.exceptions import SubmissionIssue
 from squad.ci.utils import task_id
@@ -89,3 +90,24 @@ def send_testjob_resubmit_admin_email(job_id, resubmitted_job_id):
     if test_job.target.html_mail:
         message.attach_alternative(html_message, "text/html")
     message.send()
+
+
+@celery.task
+def postprocess_testjob(job_id, status):
+    logger.info("postprocessing  %s" % job_id)
+    testjob = TestJob.objects.get(pk=job_id)
+    backend = testjob.backend
+    backend.__postprocess_testjob__(testjob)
+
+    # Some plugins, like tradefed, can still trigger sub-tasks creating tests,
+    # ending the current thread, thus triggering build-finished events
+    # we need a way to tell the plugins to wait for all sub-tasks to be finished
+
+    # Remove the 'Fetching' job_status only after eventual plugins
+    # are finished, this garantees extra tests and metadata to
+    # be in SQUAD before the build is considered finished
+    testjob.job_status = status
+    testjob.save()
+
+    if testjob.testrun:
+        UpdateProjectStatus()(testjob.testrun)

squad/settings.py

@@ -384,6 +384,7 @@
     'squad.core.tasks.notification.*': {'queue': 'core_notification'},
     'squad.ci.tasks.poll': {'queue': 'ci_poll'},
     'squad.ci.tasks.fetch': {'queue': 'ci_fetch'},
+    'squad.ci.tasks.postprocess_testjob': {'queue': 'ci_fetch_postprocess'},
     'squad.ci.tasks.submit': {'queue': 'ci_quick'},
     'squad.ci.tasks.send_testjob_resubmit_admin_email': {'queue': 'ci_quick'},
 }

test/ci/test_models.py

@@ -35,7 +35,7 @@ class BackendTestBase(TestCase):
 
     def setUp(self):
         self.group = core_models.Group.objects.create(slug='mygroup')
-        self.project = self.group.projects.create(slug='myproject')
+        self.project = self.group.projects.create(slug='myproject', enabled_plugins_list=['linux-log-parser'])
         self.backend = models.Backend.objects.create()
         self.build = self.project.builds.create(version='1')
 
@@ -430,7 +430,7 @@ def test_fetch_sets_fetched_at(self, receive, backend_fetch, backend_job_url):
         test_job.refresh_from_db()
         self.assertIsNotNone(test_job.fetched_at)
 
-    @patch.object(models.Backend, '__postprocess_testjob__')
+    @patch('squad.ci.models.Backend.__postprocess_testjob__')
     @patch('squad.ci.backend.null.Backend.job_url', return_value="http://example.com/123")
     @patch('squad.ci.backend.null.Backend.fetch')
     @patch('squad.ci.models.ReceiveTestRun.__call__')