Merge lp:~wgrant/launchpad/celery-tweaks into lp:launchpad

[William Grant]

Two celery fixes: reschedule accepted PackageCopyJobs, and fix mail delivery.

Two celery fixes:

 - PackageCopyJobs need to be requeued in celery once they're resumed by acceptFromQueue.
 - celery Zope initialisation was calling execute_zcml_for_scripts but not set_immediate_mail_delivery, so it was using dev-configured Zope transactional appserver mail.

[Steve Kowalik (stevenk)]

review: Approve (code)

=== modified file 'lib/lp/services/job/celeryjob.py'
--- lib/lp/services/job/celeryjob.py	2012-08-02 15:21:50 +0000
+++ lib/lp/services/job/celeryjob.py	2013-06-04 05:56:27 +0000
@@ -34,6 +34,7 @@
     install_feature_controller,
     make_script_feature_controller,
     )
+from lp.services.mail.sendmail import set_immediate_mail_delivery
 from lp.services.job.model.job import (
     Job,
     UniversalJobSource,
@@ -139,6 +140,7 @@
         return
     transaction.abort()
     scripts.execute_zcml_for_scripts(use_web_security=False)
+    set_immediate_mail_delivery(True)
     needs_zcml = False
 
 
 
=== modified file 'lib/lp/soyuz/model/queue.py'
--- lib/lp/soyuz/model/queue.py	2013-05-31 02:51:23 +0000
+++ lib/lp/soyuz/model/queue.py	2013-06-04 05:56:27 +0000
@@ -542,6 +542,7 @@
         self.setAccepted()
         job = PlainPackageCopyJob.get(self.package_copy_job_id)
         job.resume()
+        job.celeryRunOnCommit()
         # The copy job will send emails as appropriate.  We don't
         # need to worry about closing bugs from syncs, although we
         # should probably give karma but that needs more work to
 
=== modified file 'lib/lp/soyuz/tests/test_packagecopyjob.py'
--- lib/lp/soyuz/tests/test_packagecopyjob.py	2013-05-24 01:02:43 +0000
+++ lib/lp/soyuz/tests/test_packagecopyjob.py	2013-06-04 05:56:27 +0000
@@ -59,7 +59,7 @@
 from lp.soyuz.model.queue import PackageUpload
 from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
 from lp.testing import (
-    celebrity_logged_in,
+    admin_logged_in,
     person_logged_in,
     run_script,
     TestCaseWithFactory,
@@ -87,7 +87,7 @@
 def create_proper_job(factory):
     """Create a job that will complete successfully."""
     publisher = SoyuzTestPublisher()
-    with celebrity_logged_in('admin'):
+    with admin_logged_in():
         publisher.prepareBreezyAutotest()
     distroseries = publisher.breezy_autotest
 
@@ -1642,6 +1642,51 @@
         emails = pop_remote_notifications()
         self.assertEqual(1, len(emails))
 
+    def test_resume_from_queue(self):
+        # Accepting a suspended copy from the queue sends it back
+        # through celery.
+        self.useFixture(FeatureFixture({
+            'jobs.celery.enabled_classes': 'PlainPackageCopyJob',
+        }))
+
+        source_pub = self.factory.makeSourcePackagePublishingHistory(
+            component=u"main", status=PackagePublishingStatus.PUBLISHED)
+        target_series = self.factory.makeDistroSeries()
+        getUtility(ISourcePackageFormatSelectionSet).add(
+            target_series, SourcePackageFormat.FORMAT_1_0)
+        requester = self.factory.makePerson()
+        with person_logged_in(target_series.main_archive.owner):
+            target_series.main_archive.newComponentUploader(requester, u"main")
+        job = getUtility(IPlainPackageCopyJobSource).create(
+            package_name=source_pub.source_package_name,
+            package_version=source_pub.source_package_version,
+            source_archive=source_pub.archive,
+            target_archive=target_series.main_archive,
+            target_distroseries=target_series,
+            target_pocket=PackagePublishingPocket.PROPOSED,
+            include_binaries=False, requester=requester)
+
+        # Run the job once. There's no ancestry so it will be suspended
+        # and added to the queue.
+        with block_on_job(self):
+            transaction.commit()
+        self.assertEqual(JobStatus.SUSPENDED, job.status)
+
+        # Approve its queue entry and rerun to completion.
+        pu = getUtility(IPackageUploadSet).getByPackageCopyJobIDs(
+            [job.id]).one()
+        with admin_logged_in():
+            pu.acceptFromQueue()
+        self.assertEqual(JobStatus.WAITING, job.status)
+
+        with block_on_job(self):
+            transaction.commit()
+        self.assertEqual(JobStatus.COMPLETED, job.status)
+        self.assertEqual(
+            1,
+            target_series.main_archive.getPublishedSources(
+                name=source_pub.source_package_name).count())
+
 
 class TestPlainPackageCopyJobPermissions(TestCaseWithFactory):