Merge lp:~abentley/launchpad/celery-everywhere-2 into lp:launchpad

[Aaron Bentley]

Support revision mail jobs and translation jobs via Celery.

= Summary =
Support revision mail jobs and translation jobs via Celery.

== Pre-implementation notes ==
None

== Implementation details ==
Add celeryRunJob calls for RosettaUploadJob, RevisionMailJob and RevisionsAddedJob, and test.

Provide pop_notifications task to retrieve the mail notifications from a celeryd.

Rename the feature flag to remove an extra "s" in "classes".

Rename the task queue names to "job" and "branch_write_job", to be more descriptive.

Run jobs with the appropriate database user. To do this, introduce a "config" member on BaseRunnableJob that gives the configuration for that job type.

== Tests ==
bin/test -vt ViaCelery

== Demo and Q/A ==
None

= Launchpad lint =

Checking for conflicts and issues in changed files.

Linting changed files:
  lib/lp/services/job/tests/test_runner.py
  lib/lp/services/job/runner.py
  lib/lp/code/model/tests/test_branchjob.py
  lib/lp/services/job/tests/celery_helpers.py
  lib/lp/codehosting/scanner/tests/test_email.py
  lib/lp/services/features/flags.py
  lib/lp/services/job/celeryconfig.py
  lib/lp/services/job/model/job.py
  lib/lp/services/job/tests/__init__.py
  lib/lp/codehosting/scanner/email.py
  lib/lp/code/model/tests/test_branch.py
  lib/lp/code/model/branchjob.py

[Deryck Hodge (deryck)]

review: Approve

=== modified file 'lib/lp/code/model/branchjob.py'
--- lib/lp/code/model/branchjob.py	2012-04-06 17:28:25 +0000
+++ lib/lp/code/model/branchjob.py	2012-04-10 16:18:21 +0000
@@ -80,7 +80,10 @@
 from lp.code.model.branch import Branch
 from lp.code.model.branchmergeproposal import BranchMergeProposal
 from lp.code.model.revision import RevisionSet
-from lp.codehosting.bzrutils import server
+from lp.codehosting.bzrutils import (
+    read_locked,
+    server,
+    )
 from lp.codehosting.scanner.bzrsync import BzrSync
 from lp.codehosting.vfs import (
     get_ro_server,
@@ -89,6 +92,7 @@
 from lp.codehosting.vfs.branchfs import get_real_branch_path
 from lp.registry.interfaces.productseries import IProductSeriesSet
 from lp.scripts.helpers import TransactionFreeOperation
+from lp.services.config import config
 from lp.services.database.enumcol import EnumCol
 from lp.services.database.lpstorm import IStore
 from lp.services.database.sqlbase import SQLBase
@@ -296,6 +300,8 @@
     class_job_type = BranchJobType.SCAN_BRANCH
     memory_limit = 2 * (1024 ** 3)
 
+    config = config.branchscanner
+
     @classmethod
     def create(cls, branch):
         """See `IBranchScanJobSource`."""
@@ -327,7 +333,9 @@
 
     user_error_types = (NotBranchError,)
 
-    task_queue = 'branch_write'
+    task_queue = 'branch_write_job'
+
+    config = config.upgrade_branches
 
     def getOperationDescription(self):
         return 'upgrading a branch'
@@ -413,6 +421,8 @@
 
     class_job_type = BranchJobType.REVISION_MAIL
 
+    config = config.sendbranchmail
+
     @classmethod
     def create(cls, branch, revno, from_address, body, subject):
         """See `IRevisionMailJobSource`."""
@@ -458,6 +468,8 @@
 
     class_job_type = BranchJobType.REVISIONS_ADDED_MAIL
 
+    config = config.sendbranchmail
+
     @classmethod
     def create(cls, branch, last_scanned_id, last_revision_id,
                from_address):
@@ -519,16 +531,13 @@
         subscriptions = self.branch.getSubscriptionsByLevel(diff_levels)
         if not subscriptions:
             return
-
-        self.bzr_branch.lock_read()
-        try:
-            for revision, revno in self.iterAddedMainline():
-                assert revno is not None
-                mailer = self.getMailerForRevision(
-                    revision, revno, self.generateDiffs())
-                mailer.sendAll()
-        finally:
-            self.bzr_branch.unlock()
+        with server(get_ro_server(), no_replace=True):
+            with read_locked(self.bzr_branch):
+                for revision, revno in self.iterAddedMainline():
+                    assert revno is not None
+                    mailer = self.getMailerForRevision(
+                        revision, revno, self.generateDiffs())
+                    mailer.sendAll()
 
     def getDiffForRevisions(self, from_revision_id, to_revision_id):
         """Generate the diff between from_revision_id and to_revision_id."""
@@ -718,6 +727,8 @@
 
     class_job_type = BranchJobType.ROSETTA_UPLOAD
 
+    config = config.rosettabranches
+
     def __init__(self, branch_job):
         super(RosettaUploadJob, self).__init__(branch_job)
 
@@ -763,7 +774,9 @@
                                        force_translations_upload)
             branch_job = BranchJob(
                 branch, BranchJobType.ROSETTA_UPLOAD, metadata)
-            return cls(branch_job)
+            job = cls(branch_job)
+            job.celeryRunOnCommit()
+            return job
         else:
             return None
 
@@ -889,27 +902,28 @@
 
     def run(self):
         """See `IRosettaUploadJob`."""
-        # This is not called upon job creation because the branch would
-        # neither have been mirrored nor scanned then.
-        self._init_translation_file_lists()
-        # Get the product series that are connected to this branch and
-        # that want to upload translations.
-        productseriesset = getUtility(IProductSeriesSet)
-        productseries = productseriesset.findByTranslationsImportBranch(
-            self.branch, self.force_translations_upload)
-        translation_import_queue = getUtility(ITranslationImportQueue)
-        for series in productseries:
-            approver = TranslationBranchApprover(self.file_names,
-                                                 productseries=series)
-            for iter_info in self._iter_lists_and_uploaders(series):
-                file_names, changed_files, uploader = iter_info
-                for upload_file_name, upload_file_content in changed_files:
-                    if len(upload_file_content) == 0:
-                        continue  # Skip empty files
-                    entry = translation_import_queue.addOrUpdateEntry(
-                        upload_file_name, upload_file_content,
-                        True, uploader, productseries=series)
-                    approver.approve(entry)
+        with server(get_ro_server(), no_replace=True):
+            # This is not called upon job creation because the branch would
+            # neither have been mirrored nor scanned then.
+            self._init_translation_file_lists()
+            # Get the product series that are connected to this branch and
+            # that want to upload translations.
+            productseriesset = getUtility(IProductSeriesSet)
+            productseries = productseriesset.findByTranslationsImportBranch(
+                self.branch, self.force_translations_upload)
+            translation_import_queue = getUtility(ITranslationImportQueue)
+            for series in productseries:
+                approver = TranslationBranchApprover(self.file_names,
+                                                     productseries=series)
+                for iter_info in self._iter_lists_and_uploaders(series):
+                    file_names, changed_files, uploader = iter_info
+                    for upload_file_name, upload_file_content in changed_files:
+                        if len(upload_file_content) == 0:
+                            continue  # Skip empty files
+                        entry = translation_import_queue.addOrUpdateEntry(
+                            upload_file_name, upload_file_content,
+                            True, uploader, productseries=series)
+                        approver.approve(entry)
 
     @staticmethod
     def iterReady():
@@ -949,7 +963,9 @@
 
     class_job_type = BranchJobType.RECLAIM_BRANCH_SPACE
 
-    task_queue = 'branch_write'
+    task_queue = 'branch_write_job'
+
+    config = config.reclaimbranchspace
 
     def __repr__(self):
         return '<RECLAIM_BRANCH_SPACE branch job (%(id)s) for %(branch)s>' % {

=== modified file 'lib/lp/code/model/tests/test_branch.py'
--- lib/lp/code/model/tests/test_branch.py	2012-04-06 17:28:25 +0000
+++ lib/lp/code/model/tests/test_branch.py	2012-04-10 16:18:21 +0000
@@ -316,8 +316,8 @@
         # lp.services.job.celeryconfig is loaded.
         from celery.exceptions import TimeoutError
         self.useFixture(FeatureFixture({
-            'jobs.celery.enabled_classses': 'BranchScanJob'}))
-        with celeryd('standard') as proc:
+            'jobs.celery.enabled_classes': 'BranchScanJob'}))
+        with celeryd('job') as proc:
             self.useBzrBranches()
             db_branch, bzr_tree = self.create_branch_and_tree()
             bzr_tree.commit(
@@ -348,8 +348,8 @@
         """Calling destroySelf causes Celery to delete the branch."""
         from celery.exceptions import TimeoutError
         self.useFixture(FeatureFixture({
-            'jobs.celery.enabled_classses': 'ReclaimBranchSpaceJob'}))
-        with celeryd('branch_write'):
+            'jobs.celery.enabled_classes': 'ReclaimBranchSpaceJob'}))
+        with celeryd('branch_write_job'):
             self.useBzrBranches()
             db_branch, tree = self.create_branch_and_tree()
             branch_path = get_real_branch_path(db_branch.id)
@@ -818,7 +818,7 @@
 
     def test_requestUpgradeUsesCelery(self):
         self.useFixture(FeatureFixture({
-            'jobs.celery.enabled_classses': 'BranchUpgradeJob'}))
+            'jobs.celery.enabled_classes': 'BranchUpgradeJob'}))
         cwd = os.getcwd()
         self.useBzrBranches()
         db_branch, tree = create_knit(self)
@@ -829,7 +829,7 @@
         db_branch.requestUpgrade(db_branch.owner)
         with monitor_celery() as responses:
             transaction.commit()
-            with celeryd('branch_write', cwd):
+            with celeryd('branch_write_job', cwd):
                 responses[-1].wait(30)
         new_branch = Branch.open(tree.branch.base)
         self.assertEqual(

=== modified file 'lib/lp/code/model/tests/test_branchjob.py'
--- lib/lp/code/model/tests/test_branchjob.py	2012-04-06 17:28:25 +0000
+++ lib/lp/code/model/tests/test_branchjob.py	2012-04-10 16:18:21 +0000
@@ -60,6 +60,7 @@
     RosettaUploadJob,
     )
 from lp.code.model.branchrevision import BranchRevision
+from lp.code.model.directbranchcommit import DirectBranchCommit
 from lp.code.model.tests.test_branch import create_knit
 from lp.code.model.revision import RevisionSet
 from lp.codehosting.vfs import branch_id_to_path
@@ -67,18 +68,27 @@
 from lp.services.config import config
 from lp.services.database.constants import UTC_NOW
 from lp.services.database.lpstorm import IMasterStore
+from lp.services.features.testing import FeatureFixture
 from lp.services.identity.interfaces.emailaddress import EmailAddressStatus
 from lp.services.job.interfaces.job import JobStatus
 from lp.services.job.model.job import Job
 from lp.services.job.runner import JobRunner
+from lp.services.job.tests import (
+    celeryd,
+    monitor_celery,
+    )
 from lp.services.osutils import override_environ
 from lp.services.webapp import canonical_url
-from lp.testing import TestCaseWithFactory
+from lp.testing import (
+    person_logged_in,
+    TestCaseWithFactory,
+    )
 from lp.testing.dbuser import (
     dbuser,
     switch_dbuser,
     )
 from lp.testing.layers import (
+    AppServerLayer,
     DatabaseFunctionalLayer,
     LaunchpadZopelessLayer,
     )
@@ -1222,6 +1232,41 @@
         self.assertEqual([], unfinished_jobs)
 
 
+class TestViaCelery(TestCaseWithFactory):
+
+    layer = AppServerLayer
+
+    def test_RosettaUploadJob(self):
+        """Ensure RosettaUploadJob can run under Celery."""
+        self.useContext(celeryd('job'))
+        self.useBzrBranches(direct_database=True)
+        self.useFixture(FeatureFixture({
+            'jobs.celery.enabled_classes': 'BranchScanJob RosettaUploadJob'
+        }))
+        db_branch = self.factory.makeAnyBranch()
+        self.createBzrBranch(db_branch)
+        commit = DirectBranchCommit(db_branch, no_race_check=True)
+        commit.writeFile('foo.pot', 'gibberish')
+        with monitor_celery() as responses:
+            with person_logged_in(db_branch.owner):
+                commit.commit('message')
+                transaction.commit()
+                # Wait for branch scan to complete.
+                responses[0].wait(30)
+                series = self.factory.makeProductSeries(branch=db_branch)
+                RosettaUploadJob.create(
+                    commit.db_branch, NULL_REVISION,
+                    force_translations_upload=True)
+                transaction.commit()
+        # Wait for RosettaUploadJob to complete
+        responses[1].wait(30)
+        queue = getUtility(ITranslationImportQueue)
+        entries = list(queue.getAllEntries(target=series))
+        self.assertEqual(len(entries), 1)
+        entry = entries[0]
+        self.assertEqual('foo.pot', entry.path)
+
+
 class TestReclaimBranchSpaceJob(TestCaseWithFactory):
 
     layer = LaunchpadZopelessLayer

=== modified file 'lib/lp/codehosting/scanner/email.py'
--- lib/lp/codehosting/scanner/email.py	2012-01-01 02:58:52 +0000
+++ lib/lp/codehosting/scanner/email.py	2012-04-10 16:18:21 +0000
@@ -46,10 +46,11 @@
     # No diff is associated with the removed email.
     subject = "[Branch %s] %s removed" % (
         revisions_removed.db_branch.unique_name, count)
-    getUtility(IRevisionMailJobSource).create(
+    job = getUtility(IRevisionMailJobSource).create(
         revisions_removed.db_branch, revno='removed',
         from_address=config.canonical.noreply_from_address,
         body=contents, subject=subject)
+    job.celeryRunOnCommit()
 
 
 def queue_tip_changed_email_jobs(tip_changed):
@@ -66,11 +67,12 @@
                    revisions)
         subject = "[Branch %s] %s" % (
             tip_changed.db_branch.unique_name, revisions)
-        getUtility(IRevisionMailJobSource).create(
+        job = getUtility(IRevisionMailJobSource).create(
             tip_changed.db_branch, 'initial',
             config.canonical.noreply_from_address, message, subject)
     else:
-        getUtility(IRevisionsAddedJobSource).create(
+        job = getUtility(IRevisionsAddedJobSource).create(
             tip_changed.db_branch, tip_changed.db_branch.last_scanned_id,
             tip_changed.bzr_branch.last_revision(),
             config.canonical.noreply_from_address)
+    job.celeryRunOnCommit()

=== modified file 'lib/lp/codehosting/scanner/tests/test_email.py'
--- lib/lp/codehosting/scanner/tests/test_email.py	2012-01-01 02:58:52 +0000
+++ lib/lp/codehosting/scanner/tests/test_email.py	2012-04-10 16:18:21 +0000
@@ -7,6 +7,7 @@
 
 import email
 
+from bzrlib.uncommit import uncommit
 from zope.component import getUtility
 from zope.event import notify
 
@@ -20,13 +21,34 @@
     IRevisionsAddedJobSource,
     )
 from lp.code.model.branchjob import RevisionMailJob
+from lp.codehosting.scanner.bzrsync import BzrSync
 from lp.codehosting.scanner import events
 from lp.codehosting.scanner.tests.test_bzrsync import BzrSyncTestCase
 from lp.registry.interfaces.person import IPersonSet
+from lp.services.config import config
+from lp.services.features.testing import FeatureFixture
 from lp.services.job.runner import JobRunner
 from lp.services.mail import stub
+from lp.services.job.tests import (
+    celeryd,
+    monitor_celery,
+    )
 from lp.testing import TestCaseWithFactory
-from lp.testing.layers import LaunchpadZopelessLayer
+from lp.testing.dbuser import switch_dbuser
+from lp.testing.layers import (
+    LaunchpadZopelessLayer,
+    ZopelessAppServerLayer,
+    )
+
+
+def add_subscriber(branch):
+    test_user = getUtility(IPersonSet).getByEmail('[e-mail]')
+    branch.subscribe(
+        test_user,
+        BranchSubscriptionNotificationLevel.FULL,
+        BranchSubscriptionDiffSize.FIVEKLINES,
+        CodeReviewNotificationLevel.NOEMAIL,
+        test_user)
 
 
 class TestBzrSyncEmail(BzrSyncTestCase):
@@ -39,13 +61,7 @@
     def makeDatabaseBranch(self):
         branch = BzrSyncTestCase.makeDatabaseBranch(self)
         LaunchpadZopelessLayer.txn.begin()
-        test_user = getUtility(IPersonSet).getByEmail('[e-mail]')
-        branch.subscribe(
-            test_user,
-            BranchSubscriptionNotificationLevel.FULL,
-            BranchSubscriptionDiffSize.FIVEKLINES,
-            CodeReviewNotificationLevel.NOEMAIL,
-            test_user)
+        add_subscriber(branch)
         LaunchpadZopelessLayer.txn.commit()
         return branch
 
@@ -130,12 +146,12 @@
         recommit_email_msg = email.message_from_string(recommit_email[2])
         recommit_email_body = recommit_email_msg.get_payload()[0].get_payload(
             decode=True)
-        subject =  '[Branch %s] Rev 1: second' % self.db_branch.unique_name
+        subject = '[Branch %s] Rev 1: second' % self.db_branch.unique_name
         self.assertEmailHeadersEqual(subject, recommit_email_msg['Subject'])
         body_bits = [
             'revno: 1',
             'committer: %s' % author,
-            'branch nick: %s'  % self.bzr_branch.nick,
+            'branch nick: %s' % self.bzr_branch.nick,
             'message:\n  second',
             'added:\n  hello.txt',
             ]
@@ -143,6 +159,63 @@
             self.assertTextIn(bit, recommit_email_body)
 
 
+class TestViaCelery(TestCaseWithFactory):
+
+    layer = ZopelessAppServerLayer
+
+    @staticmethod
+    def pop_notifications():
+        from lp.services.job.tests.celery_helpers import pop_notifications
+        return pop_notifications.delay().get(30)
+
+    def prepare(self, job_name):
+        self.useFixture(FeatureFixture(
+            {'jobs.celery.enabled_classes': job_name}))
+        self.useContext(celeryd('job'))
+        self.useBzrBranches(direct_database=True)
+        db_branch, tree = self.create_branch_and_tree()
+        add_subscriber(db_branch)
+        switch_dbuser(config.branchscanner.dbuser)
+        # Needed for feature flag teardown
+        self.addCleanup(switch_dbuser, config.launchpad.dbuser)
+        return db_branch, tree
+
+    def test_empty_branch(self):
+        """RevisionMailJob for empty branches runs via Celery."""
+        db_branch, tree = self.prepare('RevisionMailJob')
+        with monitor_celery() as responses:
+            BzrSync(db_branch).syncBranchAndClose(tree.branch)
+        responses[-1].wait(30)
+        self.assertEqual(1, len(self.pop_notifications()))
+
+    def test_uncommit_branch(self):
+        """RevisionMailJob for removed revisions runs via Celery."""
+        db_branch, tree = self.prepare('RevisionMailJob')
+        tree.commit('message')
+        bzr_sync = BzrSync(db_branch)
+        with monitor_celery() as responses:
+            bzr_sync.syncBranchAndClose(tree.branch)
+            responses[0].wait(30)
+            self.pop_notifications()
+            uncommit(tree.branch)
+            bzr_sync.syncBranchAndClose(tree.branch)
+        responses[1].wait(30)
+        self.assertEqual(1, len(self.pop_notifications()))
+
+    def test_revisions_added(self):
+        """RevisionMailJob for removed revisions runs via Celery."""
+        db_branch, tree = self.prepare('RevisionsAddedJob')
+        tree.commit('message')
+        bzr_sync = BzrSync(db_branch)
+        bzr_sync.syncBranchAndClose(tree.branch)
+        self.pop_notifications()
+        tree.commit('message2')
+        with monitor_celery() as responses:
+            bzr_sync.syncBranchAndClose(tree.branch)
+            responses[-1].wait(30)
+            self.assertEqual(1, len(self.pop_notifications()))
+
+
 class TestScanBranches(TestCaseWithFactory):
 
     layer = LaunchpadZopelessLayer

=== modified file 'lib/lp/services/features/flags.py'
--- lib/lp/services/features/flags.py	2012-04-05 13:05:04 +0000
+++ lib/lp/services/features/flags.py	2012-04-10 16:18:21 +0000
@@ -125,7 +125,7 @@
      '',
      '',
      ''),
-    ('jobs.celery.enabled_classses',
+    ('jobs.celery.enabled_classes',
      'space delimited',
      'Names of Job classes that should be run via celery',
      'No jobs run via celery',

=== modified file 'lib/lp/services/job/celeryconfig.py'
--- lib/lp/services/job/celeryconfig.py	2012-04-06 17:28:25 +0000
+++ lib/lp/services/job/celeryconfig.py	2012-04-10 16:18:21 +0000
@@ -8,9 +8,9 @@
 CELERY_IMPORTS = ("lp.services.job.celeryjob", )
 CELERY_RESULT_BACKEND = "amqp"
 CELERY_QUEUES = {
-    "branch_write": {"binding_key": "branch_write"},
-    "standard": {"binding_key": "standard"},
+    "branch_write_job": {"binding_key": "branch_write_job"},
+    "job": {"binding_key": "job"},
 }
-CELERY_DEFAULT_EXCHANGE = "standard"
-CELERY_DEFAULT_QUEUE = "standard"
+CELERY_DEFAULT_EXCHANGE = "job"
+CELERY_DEFAULT_QUEUE = "job"
 CELERY_CREATE_MISSING_QUEUES = False

=== modified file 'lib/lp/services/job/model/job.py'
--- lib/lp/services/job/model/job.py	2012-03-21 20:39:49 +0000
+++ lib/lp/services/job/model/job.py	2012-04-10 16:18:21 +0000
@@ -33,7 +33,9 @@
     Int,
     Reference,
     )
+from storm.zope.interfaces import IZStorm
 import transaction
+from zope.component import getUtility
 from zope.interface import implements
 
 from lp.services.config import dbconfig
@@ -253,17 +255,33 @@
 
     needs_init = True
 
-    @classmethod
-    def get(cls, job_id):
-        if cls.needs_init:
-            scripts.execute_zcml_for_scripts(use_web_security=False)
-            cls.needs_init = False
-
-        dbconfig.override(
-            dbuser='branchscanner', isolation_level='read_committed')
+    @staticmethod
+    def getDerived(job_id):
+        """Return the derived branch job associated with the job id."""
         from lp.code.model.branchjob import (
             BranchJob,
             )
         store = IStore(BranchJob)
         branch_job = store.find(BranchJob, BranchJob.job == job_id).one()
-        return branch_job.makeDerived()
+        if branch_job is None:
+            raise ValueError('No BranchJob with job=%s.' % job_id)
+        return branch_job.makeDerived(), store
+
+    @classmethod
+    def switchDBUser(cls, job_id):
+        """Switch to the DB user associated with this Job ID."""
+        derived, store = cls.getDerived(job_id)
+        dbconfig.override(
+            dbuser=derived.config.dbuser, isolation_level='read_committed')
+        transaction.abort()
+        getUtility(IZStorm).remove(store)
+        store.close()
+
+    @classmethod
+    def get(cls, job_id):
+        transaction.abort()
+        if cls.needs_init:
+            scripts.execute_zcml_for_scripts(use_web_security=False)
+            cls.needs_init = False
+        cls.switchDBUser(job_id)
+        return cls.getDerived(job_id)[0]

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2012-04-06 17:28:25 +0000
+++ lib/lp/services/job/runner.py	2012-04-10 16:18:21 +0000
@@ -103,7 +103,7 @@
 
     retry_error_types = ()
 
-    task_queue = 'standard'
+    task_queue = 'job'
 
     celery_responses = None
 
@@ -190,23 +190,22 @@
         return oops_config.create(
             context=dict(exc_info=info))
 
-    def runViaCelery(self):
+    def runViaCelery(self, ignore_result=False):
         """Request that this job be run via celery."""
         # Avoid importing from lp.services.job.celeryjob where not needed, to
         # avoid configuring Celery when Rabbit is not configured.
         from lp.services.job.celeryjob import CeleryRunJob
-        ignore_result = bool(BaseRunnableJob.celery_responses is None)
-        response = CeleryRunJob.apply_async(
+        return CeleryRunJob.apply_async(
             (self.job_id,), queue=self.task_queue,
             ignore_result=ignore_result)
-        if not ignore_result:
-            BaseRunnableJob.celery_responses.append(response)
-        return response
 
     def celeryCommitHook(self, succeeded):
         """Hook function to call when a commit completes."""
         if succeeded:
-            self.runViaCelery()
+            ignore_result = bool(BaseRunnableJob.celery_responses is None)
+            response = self.runViaCelery(ignore_result)
+            if not ignore_result:
+                BaseRunnableJob.celery_responses.append(response)
 
     def celeryRunOnCommit(self):
         """Configure transaction so that commit runs this job via Celery."""
@@ -623,7 +622,7 @@
 
     The name of a BaseRunnableJob must be specified.
     """
-    flag = getFeatureFlag('jobs.celery.enabled_classses')
+    flag = getFeatureFlag('jobs.celery.enabled_classes')
     if flag is None:
         return False
     return class_name in flag.split(' ')

=== modified file 'lib/lp/services/job/tests/__init__.py'
--- lib/lp/services/job/tests/__init__.py	2012-04-06 17:28:25 +0000
+++ lib/lp/services/job/tests/__init__.py	2012-04-10 16:18:21 +0000
@@ -32,6 +32,7 @@
         '--concurrency', '1',
         '--loglevel', 'INFO',
         '--queues', queue,
+        '--include', 'lp.services.job.tests.celery_helpers',
     )
     with running('bin/celeryd', cmd_args, cwd=cwd) as proc:
         # Wait for celeryd startup to complete.

=== added file 'lib/lp/services/job/tests/celery_helpers.py'
--- lib/lp/services/job/tests/celery_helpers.py	1970-01-01 00:00:00 +0000
+++ lib/lp/services/job/tests/celery_helpers.py	2012-04-10 16:18:21 +0000
@@ -0,0 +1,19 @@
+# Copyright 2012 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+__metaclass__ = type
+
+__all__ = ['pop_notifications']
+
+# Force the correct celeryconfig to be used.
+import lp.services.job.celeryjob
+# Quiet lint unused import warning.
+lp.services.job.celeryjob
+
+from celery.task import task
+
+
+@task
+def pop_notifications():
+    from lp.testing.mail_helpers import pop_notifications
+    return pop_notifications()

=== modified file 'lib/lp/services/job/tests/test_runner.py'
--- lib/lp/services/job/tests/test_runner.py	2012-04-03 16:06:43 +0000
+++ lib/lp/services/job/tests/test_runner.py	2012-04-10 16:18:21 +0000
@@ -720,19 +720,19 @@
     def test_matching_flag(self):
         """A matching flag returns True."""
         self.useFixture(FeatureFixture(
-            {'jobs.celery.enabled_classses': 'foo bar'}))
+            {'jobs.celery.enabled_classes': 'foo bar'}))
         self.assertTrue(celery_enabled('foo'))
         self.assertTrue(celery_enabled('bar'))
 
     def test_non_matching_flag(self):
         """A non-matching flag returns false."""
         self.useFixture(FeatureFixture(
-            {'jobs.celery.enabled_classses': 'foo bar'}))
+            {'jobs.celery.enabled_classes': 'foo bar'}))
         self.assertFalse(celery_enabled('baz'))
         self.assertTrue(celery_enabled('bar'))
 
     def test_substring(self):
         """A substring of an enabled class does not match."""
         self.useFixture(FeatureFixture(
-            {'jobs.celery.enabled_classses': 'foobar'}))
+            {'jobs.celery.enabled_classes': 'foobar'}))
         self.assertFalse(celery_enabled('bar'))