Merge lp:~abentley/launchpad/celery-job-layer into lp:launchpad

[Aaron Bentley]

Run celeryd via a layer in tests.

= Summary =
Run celeryd via a layer in tests.

== Proposed fix ==
None

== Pre-implementation notes ==
Robert confirmed that Layers are still our preferred means of reducing redundant startup/teardown costs.

== Implementation details ==
Implement CeleryJobLayer and CeleryBranchWriteJobLayer to provide celeryd instances for the corresponding queues.

Update all existing tests to use these layers

Extract block_on_job from various tests that retrieve the job's response and then wait.

lp.services.job.tests.celeryd only specifies parameters to lazr.jobrunner.tests.test_celerytask.running, and does not change its behaviour, so it can return running instead of being a contextmanager, itself.

== Tests ==
bin/test --layer=CeleryJobLayer
bin/test --layer=CeleryBranchWriteJobLayer

== 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/codehosting/scanner/tests/test_mergedetection.py
  lib/lp/services/job/runner.py
  lib/lp/code/model/tests/test_branchmergeproposaljobs.py
  lib/lp/code/model/branchmergeproposal.py
  lib/lp/code/model/tests/test_branchjob.py
  lib/lp/services/job/tests/celery_helpers.py
  lib/lp/services/job/celeryjob.py
  lib/lp/codehosting/scanner/tests/test_email.py
  lib/lp/services/job/tests/test_job.py
  lib/lp/services/features/flags.py
  lib/lp/code/model/branchmergeproposaljob.py
  lib/lp/testing/layers.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

[j.c.sackett (jcsackett)]

Aaron--

This looks like a really good cleanup. I think the tests are much cleaner this way. Thanks.

review: Approve

=== modified file 'lib/lp/code/model/tests/test_branch.py'
--- lib/lp/code/model/tests/test_branch.py	2012-04-10 20:24:43 +0000
+++ lib/lp/code/model/tests/test_branch.py	2012-04-13 19:33:20 +0000
@@ -12,7 +12,6 @@
     datetime,
     timedelta,
     )
-import os
 
 from bzrlib.branch import Branch
 from bzrlib.bzrdir import BzrDir
@@ -118,7 +117,7 @@
 from lp.services.database.lpstorm import IStore
 from lp.services.features.testing import FeatureFixture
 from lp.services.job.tests import (
-    celeryd,
+    block_on_job,
     monitor_celery,
     )
 from lp.services.osutils import override_environ
@@ -141,6 +140,8 @@
 from lp.testing.factory import LaunchpadObjectFactory
 from lp.testing.layers import (
     AppServerLayer,
+    CeleryBranchWriteJobLayer,
+    CeleryJobLayer,
     DatabaseFunctionalLayer,
     LaunchpadFunctionalLayer,
     LaunchpadZopelessLayer,
@@ -153,8 +154,9 @@
 
 def create_knit(test_case):
     db_branch, tree = test_case.create_branch_and_tree(format='knit')
-    db_branch.branch_format = BranchFormat.BZR_BRANCH_5
-    db_branch.repository_format = RepositoryFormat.BZR_KNIT_1
+    with person_logged_in(db_branch.owner):
+        db_branch.branch_format = BranchFormat.BZR_BRANCH_5
+        db_branch.repository_format = RepositoryFormat.BZR_KNIT_1
     return db_branch, tree
 
 
@@ -307,62 +309,75 @@
 
 class TestBranchJobViaCelery(TestCaseWithFactory):
 
-    layer = ZopelessAppServerLayer
+    layer = CeleryJobLayer
 
     def test_branchChanged_via_celery(self):
         """Running a job via Celery succeeds and emits expected output."""
         # Delay importing anything that uses Celery until RabbitMQLayer is
         # running, so that config.rabbitmq.host is defined when
         # lp.services.job.celeryconfig is loaded.
-        from celery.exceptions import TimeoutError
         self.useFixture(FeatureFixture({
             'jobs.celery.enabled_classes': 'BranchScanJob'}))
-        with celeryd('job') as proc:
-            self.useBzrBranches()
-            db_branch, bzr_tree = self.create_branch_and_tree()
-            bzr_tree.commit(
-                'First commit', rev_id='rev1', committer='[e-mail]')
+        self.useBzrBranches()
+        db_branch, bzr_tree = self.create_branch_and_tree()
+        bzr_tree.commit(
+            'First commit', rev_id='rev1', committer='[e-mail]')
+        with person_logged_in(db_branch.owner):
             db_branch.branchChanged(None, 'rev1', None, None, None)
-            with monitor_celery() as responses:
-                transaction.commit()
-                try:
-                    responses[-1].wait(30)
-                except TimeoutError:
-                    pass
-        self.assertIn(
-            'Updating branch scanner status: 1 revs', proc.stderr.read())
+        with block_on_job():
+            transaction.commit()
         self.assertEqual(db_branch.revision_count, 1)
 
     def test_branchChanged_via_celery_no_enabled(self):
-        """Running a job via Celery succeeds and emits expected output."""
+        """With no feature flag, no task is created."""
         self.useBzrBranches()
         db_branch, bzr_tree = self.create_branch_and_tree()
         bzr_tree.commit(
             'First commit', rev_id='rev1', committer='[e-mail]')
-        db_branch.branchChanged(None, 'rev1', None, None, None)
+        with person_logged_in(db_branch.owner):
+            db_branch.branchChanged(None, 'rev1', None, None, None)
         with monitor_celery() as responses:
             transaction.commit()
             self.assertEqual([], responses)
 
+
+class TestBranchWriteJobViaCelery(TestCaseWithFactory):
+
+    layer = CeleryBranchWriteJobLayer
+
     def test_destroySelf_via_celery(self):
         """Calling destroySelf causes Celery to delete the branch."""
-        from celery.exceptions import TimeoutError
         self.useFixture(FeatureFixture({
             '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)
-            self.assertThat(branch_path, PathExists())
+        self.useBzrBranches()
+        db_branch, tree = self.create_branch_and_tree()
+        branch_path = get_real_branch_path(db_branch.id)
+        self.assertThat(branch_path, PathExists())
+        with person_logged_in(db_branch.owner):
             db_branch.destroySelf()
-            with monitor_celery() as responses:
-                transaction.commit()
-                try:
-                    responses[-1].wait(30)
-                except TimeoutError:
-                    pass
+        with block_on_job():
+            transaction.commit()
         self.assertThat(branch_path, Not(PathExists()))
 
+    def test_requestUpgradeUsesCelery(self):
+        self.useFixture(FeatureFixture({
+            'jobs.celery.enabled_classes': 'BranchUpgradeJob'}))
+        self.useBzrBranches()
+        db_branch, tree = create_knit(self)
+        self.assertEqual(
+            tree.branch.repository._format.get_format_string(),
+            'Bazaar-NG Knit Repository Format 1')
+
+        with person_logged_in(db_branch.owner):
+            db_branch.requestUpgrade(db_branch.owner)
+        with block_on_job():
+            transaction.commit()
+        new_branch = Branch.open(tree.branch.base)
+        self.assertEqual(
+            new_branch.repository._format.get_format_string(),
+            'Bazaar repository format 2a (needs bzr 1.16 or later)\n')
+        self.assertFalse(db_branch.needs_upgrading)
+
 
 class TestBranchRevisionMethods(TestCaseWithFactory):
     """Test the branch methods for adding and removing branch revisions."""
@@ -816,27 +831,6 @@
             jobs,
             [job, ])
 
-    def test_requestUpgradeUsesCelery(self):
-        self.useFixture(FeatureFixture({
-            'jobs.celery.enabled_classes': 'BranchUpgradeJob'}))
-        cwd = os.getcwd()
-        self.useBzrBranches()
-        db_branch, tree = create_knit(self)
-        self.assertEqual(
-            tree.branch.repository._format.get_format_string(),
-            'Bazaar-NG Knit Repository Format 1')
-
-        db_branch.requestUpgrade(db_branch.owner)
-        with monitor_celery() as responses:
-            transaction.commit()
-            with celeryd('branch_write_job', cwd):
-                responses[-1].wait(30)
-        new_branch = Branch.open(tree.branch.base)
-        self.assertEqual(
-            new_branch.repository._format.get_format_string(),
-            'Bazaar repository format 2a (needs bzr 1.16 or later)\n')
-        self.assertFalse(db_branch.needs_upgrading)
-
     def test_requestUpgrade_no_upgrade_needed(self):
         # If a branch doesn't need to be upgraded, requestUpgrade raises an
         # AlreadyLatestFormat.

=== modified file 'lib/lp/code/model/tests/test_branchjob.py'
--- lib/lp/code/model/tests/test_branchjob.py	2012-04-10 20:24:43 +0000
+++ lib/lp/code/model/tests/test_branchjob.py	2012-04-13 19:33:20 +0000
@@ -74,8 +74,7 @@
 from lp.services.job.model.job import Job
 from lp.services.job.runner import JobRunner
 from lp.services.job.tests import (
-    celeryd,
-    monitor_celery,
+    block_on_job,
     )
 from lp.services.osutils import override_environ
 from lp.services.webapp import canonical_url
@@ -88,7 +87,7 @@
     switch_dbuser,
     )
 from lp.testing.layers import (
-    AppServerLayer,
+    CeleryJobLayer,
     DatabaseFunctionalLayer,
     LaunchpadZopelessLayer,
     )
@@ -1234,11 +1233,10 @@
 
 class TestViaCelery(TestCaseWithFactory):
 
-    layer = AppServerLayer
+    layer = CeleryJobLayer
 
     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'
@@ -1247,19 +1245,17 @@
         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):
+        with person_logged_in(db_branch.owner):
+            # wait for branch scan
+            with block_on_job():
                 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)
+        series = self.factory.makeProductSeries(branch=db_branch)
+        with block_on_job():
+            RosettaUploadJob.create(
+                commit.db_branch, NULL_REVISION,
+                force_translations_upload=True)
+            transaction.commit()
         queue = getUtility(ITranslationImportQueue)
         entries = list(queue.getAllEntries(target=series))
         self.assertEqual(len(entries), 1)

=== modified file 'lib/lp/code/model/tests/test_branchmergeproposaljobs.py'
--- lib/lp/code/model/tests/test_branchmergeproposaljobs.py	2012-04-13 19:33:19 +0000
+++ lib/lp/code/model/tests/test_branchmergeproposaljobs.py	2012-04-13 19:33:20 +0000
@@ -60,8 +60,7 @@
 from lp.services.job.model.job import Job
 from lp.services.job.runner import JobRunner
 from lp.services.job.tests import (
-    celeryd,
-    monitor_celery,
+    block_on_job,
     pop_remote_notifications,
     )
 from lp.services.osutils import override_environ
@@ -72,7 +71,7 @@
     )
 from lp.testing.dbuser import dbuser
 from lp.testing.layers import (
-    AppServerLayer,
+    CeleryJobLayer,
     LaunchpadZopelessLayer,
     )
 from lp.testing.mail_helpers import pop_notifications
@@ -595,79 +594,67 @@
 
 class TestViaCelery(TestCaseWithFactory):
 
-    layer = AppServerLayer
+    layer = CeleryJobLayer
 
     def test_MergeProposalNeedsReviewEmailJob(self):
         """MergeProposalNeedsReviewEmailJob runs under Celery."""
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes':
              'MergeProposalNeedsReviewEmailJob'}))
-        self.useContext(celeryd('job'))
         bmp = self.factory.makeBranchMergeProposal()
-        with monitor_celery() as responses:
+        with block_on_job():
             MergeProposalNeedsReviewEmailJob.create(bmp)
             transaction.commit()
-        responses[0].wait(30)
         self.assertEqual(2, len(pop_remote_notifications()))
 
     def test_UpdatePreviewDiffJob(self):
         """UpdatePreviewDiffJob runs under Celery."""
-        self.useContext(celeryd('job'))
         self.useBzrBranches(direct_database=True)
         bmp = create_example_merge(self)[0]
         self.factory.makeRevisionsForBranch(bmp.source_branch, count=1)
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes': 'UpdatePreviewDiffJob'}))
-        with monitor_celery() as responses:
+        with block_on_job():
             UpdatePreviewDiffJob.create(bmp)
             transaction.commit()
-            responses[0].wait(30)
         self.assertIsNot(None, bmp.preview_diff)
 
     def test_CodeReviewCommentEmailJob(self):
         """CodeReviewCommentEmailJob runs under Celery."""
         comment = self.factory.makeCodeReviewComment()
-        self.useContext(celeryd('job'))
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes': 'CodeReviewCommentEmailJob'}))
-        with monitor_celery() as responses:
+        with block_on_job():
             CodeReviewCommentEmailJob.create(comment)
             transaction.commit()
-        responses[0].wait(30)
         self.assertEqual(2, len(pop_remote_notifications()))
 
     def test_ReviewRequestedEmailJob(self):
         """ReviewRequestedEmailJob runs under Celery."""
         request = self.factory.makeCodeReviewVoteReference()
-        self.useContext(celeryd('job'))
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes': 'ReviewRequestedEmailJob'}))
-        with monitor_celery() as responses:
+        with block_on_job():
             ReviewRequestedEmailJob.create(request)
             transaction.commit()
-        responses[0].wait(30)
         self.assertEqual(1, len(pop_remote_notifications()))
 
     def test_MergeProposalUpdatedEmailJob(self):
         """MergeProposalUpdatedEmailJob runs under Celery."""
         bmp = self.factory.makeBranchMergeProposal()
-        self.useContext(celeryd('job'))
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes': 'MergeProposalUpdatedEmailJob'}))
-        with monitor_celery() as responses:
+        with block_on_job():
             MergeProposalUpdatedEmailJob.create(
                 bmp, 'change', bmp.registrant)
             transaction.commit()
-        responses[0].wait(30)
         self.assertEqual(2, len(pop_remote_notifications()))
 
     def test_GenerateIncrementalDiffJob(self):
         """GenerateIncrementalDiffJob runs under Celery."""
-        self.useContext(celeryd('job'))
         self.useFixture(FeatureFixture(
             {'jobs.celery.enabled_classes': 'GenerateIncrementalDiffJob'}))
-        with monitor_celery() as responses:
+        with block_on_job():
             job = make_runnable_incremental_diff_job(self)
             transaction.commit()
-        responses[0].wait(30)
         self.assertEqual(JobStatus.COMPLETED, job.status)

=== modified file 'lib/lp/codehosting/scanner/tests/test_email.py'
--- lib/lp/codehosting/scanner/tests/test_email.py	2012-04-13 19:33:19 +0000
+++ lib/lp/codehosting/scanner/tests/test_email.py	2012-04-13 19:33:20 +0000
@@ -30,15 +30,14 @@
 from lp.services.job.runner import JobRunner
 from lp.services.mail import stub
 from lp.services.job.tests import (
-    celeryd,
-    monitor_celery,
+    block_on_job,
     pop_remote_notifications,
     )
 from lp.testing import TestCaseWithFactory
 from lp.testing.dbuser import switch_dbuser
 from lp.testing.layers import (
+    CeleryJobLayer,
     LaunchpadZopelessLayer,
-    ZopelessAppServerLayer,
     )
 
 
@@ -162,12 +161,11 @@
 
 class TestViaCelery(TestCaseWithFactory):
 
-    layer = ZopelessAppServerLayer
+    layer = CeleryJobLayer
 
     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)
@@ -179,9 +177,8 @@
     def test_empty_branch(self):
         """RevisionMailJob for empty branches runs via Celery."""
         db_branch, tree = self.prepare('RevisionMailJob')
-        with monitor_celery() as responses:
+        with block_on_job():
             BzrSync(db_branch).syncBranchAndClose(tree.branch)
-        responses[-1].wait(30)
         self.assertEqual(1, len(pop_remote_notifications()))
 
     def test_uncommit_branch(self):
@@ -189,13 +186,12 @@
         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)
-            pop_remote_notifications()
-            uncommit(tree.branch)
-            bzr_sync.syncBranchAndClose(tree.branch)
-        responses[1].wait(30)
+        with block_on_job():
+            bzr_sync.syncBranchAndClose(tree.branch)
+        pop_remote_notifications()
+        uncommit(tree.branch)
+        with block_on_job():
+            bzr_sync.syncBranchAndClose(tree.branch)
         self.assertEqual(1, len(pop_remote_notifications()))
 
     def test_revisions_added(self):
@@ -206,10 +202,9 @@
         bzr_sync.syncBranchAndClose(tree.branch)
         pop_remote_notifications()
         tree.commit('message2')
-        with monitor_celery() as responses:
+        with block_on_job():
             bzr_sync.syncBranchAndClose(tree.branch)
-            responses[-1].wait(30)
-            self.assertEqual(1, len(pop_remote_notifications()))
+        self.assertEqual(1, len(pop_remote_notifications()))
 
 
 class TestScanBranches(TestCaseWithFactory):

=== modified file 'lib/lp/services/job/model/job.py'
--- lib/lp/services/job/model/job.py	2012-04-13 19:33:19 +0000
+++ lib/lp/services/job/model/job.py	2012-04-13 19:33:20 +0000
@@ -277,15 +277,20 @@
 
         return base_job.makeDerived(), store
 
+    @staticmethod
+    def clearStore(store):
+        transaction.abort()
+        getUtility(IZStorm).remove(store)
+        store.close()
+
     @classmethod
     def switchDBUser(cls, job_id):
         """Switch to the DB user associated with this Job ID."""
+        cls.clearStore(IStore(Job))
         derived, store = cls.getDerived(job_id)
         dbconfig.override(
             dbuser=derived.config.dbuser, isolation_level='read_committed')
-        transaction.abort()
-        getUtility(IZStorm).remove(store)
-        store.close()
+        cls.clearStore(store)
 
     @classmethod
     def get(cls, job_id):

=== modified file 'lib/lp/services/job/tests/__init__.py'
--- lib/lp/services/job/tests/__init__.py	2012-04-13 19:33:19 +0000
+++ lib/lp/services/job/tests/__init__.py	2012-04-13 19:33:20 +0000
@@ -4,6 +4,7 @@
 __metaclass__ = type
 
 __all__ = [
+    'block_on_job',
     'celeryd',
     'monitor_celery',
     'pop_remote_notifications',
@@ -15,7 +16,6 @@
 from lp.services.job.runner import BaseRunnableJob
 
 
-@contextmanager
 def celeryd(queue, cwd=None):
     """Return a ContextManager for a celeryd instance.
 
@@ -35,9 +35,7 @@
         '--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.
-        yield proc
+    return running('bin/celeryd', cmd_args, cwd=cwd)
 
 
 @contextmanager
@@ -52,6 +50,13 @@
         BaseRunnableJob.celery_responses = old_responses
 
 
+@contextmanager
+def block_on_job():
+    with monitor_celery() as responses:
+        yield
+    responses[-1].wait(30)
+
+
 def pop_remote_notifications():
     """Pop the notifications from a celeryd worker."""
     from lp.services.job.tests.celery_helpers import pop_notifications

=== modified file 'lib/lp/testing/layers.py'
--- lib/lp/testing/layers.py	2012-03-22 12:13:15 +0000
+++ lib/lp/testing/layers.py	2012-04-13 19:33:20 +0000
@@ -111,6 +111,7 @@
     ConfigUseFixture,
     )
 from lp.services.database.sqlbase import session_store
+from lp.services.job.tests import celeryd
 from lp.services.googlesearch.tests.googleserviceharness import (
     GoogleServiceTestSetup,
     )
@@ -1859,6 +1860,42 @@
         LayerProcessController.postTestInvariants()
 
 
+class CeleryJobLayer(AppServerLayer):
+    """Layer for tests that run jobs via Celery."""
+
+    celeryd = None
+
+    @classmethod
+    @profiled
+    def setUp(cls):
+        cls.celeryd = celeryd('job')
+        cls.celeryd.__enter__()
+
+    @classmethod
+    @profiled
+    def tearDown(cls):
+        cls.celeryd.__exit__(None, None, None)
+        cls.celeryd = None
+
+
+class CeleryBranchWriteJobLayer(AppServerLayer):
+    """Layer for tests that run jobs which write to branches via Celery."""
+
+    celeryd = None
+
+    @classmethod
+    @profiled
+    def setUp(cls):
+        cls.celeryd = celeryd('branch_write_job')
+        cls.celeryd.__enter__()
+
+    @classmethod
+    @profiled
+    def tearDown(cls):
+        cls.celeryd.__exit__(None, None, None)
+        cls.celeryd = None
+
+
 class ZopelessAppServerLayer(LaunchpadZopelessLayer):
     """Layer for tests that run in the zopeless environment with an appserver.
     """