Merge lp:~allenap/launchpad/twisted-threading-bug-491870 into lp:launchpad

[Gavin Panella]

In checkwatches, use Twisted to manage threading instead of the threading module.

[Gavin Panella (allenap)]

checkwatches can update bug watches for different remote trackers in parallel. Currently that's implemented using threads via the threading module. This branch continues to use threads, but does so with a Twisted ThreadPool.

To make testing easier, and to make the design more elegant, it's possible to pass in a scheduler to BugWatchUpdate.updateBugTrackers(). Previously the scheduling policy was baked into this method.

Getting Twisted in there also opens the door for using more async code in the future.

[Henning Eggers (henninge)]

Thanks for this improvement!

We talked about some things on IRC:
- Use of lambda is against LP coding style but common in Twisted. We will raise the issue at the reviewers meeting and you can leave it as it is for now. If the style change gets rejected, I ask you to fix this in a follow-up branch. Or wait until after the meeting before landing.
- The intended use of install_signal_handlers to avoid errors in the test should be documented.
- You agreed to add an explicit test for SerialScheduler.

Thanks for your patience with me ... ;-)

review: Approve (code)

[Jonathan Lange (jml)]

On Mon, Feb 8, 2010 at 5:38 PM, Henning Eggers
<email address hidden> wrote:
> Review: Approve code
> Thanks for this improvement!
>
> We talked about some things on IRC:
> - Use of lambda is against LP coding style but common in Twisted. We will raise the issue at the reviewers meeting and you can leave it as it is for now. If the style change gets rejected, I ask you to fix this in a follow-up branch. Or wait until after the meeting before landing.

FWIW, if the reviewers decree that lambda is too hard to read, then
it's probably worth defining a function like this:

  def kerchop(f):
    def callback(ignored, *args, **kwargs):
      return f(*args, **kwargs)
    return mergeFunctionMetadata(f, callback)

And then replacing:
  d.addCallback(lambda ignored: reactor.stop())

with:
  d.addCallback(kerchop(reactor.stop))

The function ought to go somewhere in lp.services, and have a better
name than 'kerchop'.

jml

[Gavin Panella (allenap)]

Hi Henning,

Thank you for the review. I added some more info about install_signal_handlers, and added some tests for SerialScheduler.

I also switched to using the default reactor thread pool instead of creating my own in TwistedThreadScheduler. I felt it was better to be explicit, but two things changed my mind:

 * We're calling reactor.run(). It's ours, we're not sharing it, so no harm in just using its thread pool.

 * The import fascist was complaining about importing deferToThreadPool. Maybe this should be in its module's __all__, maybe not. I'll file a bug for that.

Incremental diff: http://paste.ubuntu.com/372542/

Thanks, Gavin.

[Gavin Panella (allenap)]

> The import fascist was complaining about importing
> deferToThreadPool. Maybe this should be in its module's
> __all__, maybe not. I'll file a bug for that.

http://twistedmatrix.com/trac/ticket/4264

[Jonathan Lange (jml)]

The fascist has a place where you can disable the warning: lib/lp/scripts/utilities/importfascist.py, "valid_imports_not_in_all".

[Gavin Panella (allenap)]

Okay, I'm flip-flopping on this. Back To The Threadpool.

=== modified file 'lib/lp/bugs/doc/checkwatches-cli-switches.txt'
--- lib/lp/bugs/doc/checkwatches-cli-switches.txt	2009-11-17 11:19:25 +0000
+++ lib/lp/bugs/doc/checkwatches-cli-switches.txt	2010-03-15 14:09:31 +0000
@@ -34,7 +34,7 @@
     >>> from lp.testing.factory import LaunchpadObjectFactory
 
     >>> LaunchpadZopelessLayer.switchDbUser('launchpad')
-   >>> login('[e-mail]')
+    >>> login('[e-mail]')
     >>> bug_tracker = LaunchpadObjectFactory().makeBugTracker(
     ...     'http://example.com')
     >>> bug_tracker.active = False

=== modified file 'lib/lp/bugs/doc/externalbugtracker.txt'
--- lib/lp/bugs/doc/externalbugtracker.txt	2010-02-24 03:49:45 +0000
+++ lib/lp/bugs/doc/externalbugtracker.txt	2010-03-15 14:09:31 +0000
@@ -1105,110 +1105,3 @@
 the updateBugTrackers() method. This, too, takes a batch_size parameter, which
 allows it to be passed as a command-line option when the checkwatches script
 is run.
-
-Before going further, we must abort the current transaction to avoid
-deadlock; updateBugTrackers() runs updateBugTracker() in a different
-thread.
-
-    >>> transaction.abort()
-
-    >>> from canonical.launchpad.scripts.logger import FakeLogger
-    >>> bug_watch_updater = NonConnectingBugWatchUpdater(
-    ...     transaction, FakeLogger())
-    >>> bug_watch_updater.updateBugTrackers(
-    ...     bug_tracker_names=[standard_bugzilla.name], batch_size=2)
-    DEBUG Using a global batch size of 2
-    INFO Updating 2 watches for 2 bugs on http://example.com
-    initializeRemoteBugDB() called: [u'5', u'6']
-    getRemoteStatus() called: u'5'
-    getRemoteStatus() called: u'6'
-
-    >>> # We should log in again because updateBugTrackers() logs out.
-    >>> login('[e-mail]')
-
-By default, the updateBugTrackers() only spawns one thread, but it can
-spawn as many as required.
-
-    >>> import threading
-
-    >>> class OutputFileForThreads:
-    ...     def __init__(self):
-    ...         self.output = {}
-    ...         self.lock = threading.Lock()
-    ...     def write(self, data):
-    ...         thread_name = threading.currentThread().getName()
-    ...         self.lock.acquire()
-    ...         try:
-    ...             if thread_name in self.output:
-    ...                 self.output[thread_name].append(data)
-    ...             else:
-    ...                 self.output[thread_name] = [data]
-    ...         finally:
-    ...             self.lock.release()
-
-    >>> output_file = OutputFileForThreads()
-
-    >>> class ExternalBugTrackerForThreads(TestExternalBugTracker):
-    ...     def getModifiedRemoteBugs(self, remote_bug_ids, last_checked):
-    ...         print >> output_file, (
-    ...             "getModifiedRemoteBugs(\n"
-    ...             "    remote_bug_ids=%r,\n"
-    ...             "    last_checked=%r)" % (remote_bug_ids, last_checked))
-    ...         return [remote_bug_ids[0], remote_bug_ids[-1]]
-    ...     def getRemoteStatus(self, bug_id):
-    ...         print >> output_file, (
-    ...             "getRemoteStatus(bug_id=%r)" % bug_id)
-    ...         return 'UNKNOWN'
-    ...     def getCurrentDBTime(self):
-    ...         return None
-
-    >>> class BugWatchUpdaterForThreads(BugWatchUpdater):
-    ...     def _getExternalBugTrackersAndWatches(
-    ...         self, bug_trackers, bug_watches):
-    ...         return [(ExternalBugTrackerForThreads(), bug_watches)]
-
-    >>> threaded_bug_watch_updater = BugWatchUpdaterForThreads(
-    ...     transaction, FakeLogger(output_file))
-    >>> threaded_bug_watch_updater.updateBugTrackers(
-    ...     batch_size=5, num_threads=10)
-
-
-    >>> for thread_name in sorted(output_file.output):
-    ...     print '== %s ==' % thread_name
-    ...     print "".join(output_file.output[thread_name]),
-    == MainThread ==
-    DEBUG Using a global batch size of 5
-    DEBUG Skipping updating Ubuntu Bugzilla watches.
-    == auto-generic-string4.example.com ==
-    INFO Updating 5 watches for 5 bugs on http://example.com
-    getRemoteStatus(bug_id=u'1')
-    getRemoteStatus(bug_id=u'101')
-    getRemoteStatus(bug_id=u'5')
-    getRemoteStatus(bug_id=u'6')
-    getRemoteStatus(bug_id=u'7')
-    == debbugs ==
-    INFO Updating 5 watches for 5 bugs on http://example.com
-    getRemoteStatus(bug_id=u'280883')
-    getRemoteStatus(bug_id=u'304014')
-    getRemoteStatus(bug_id=u'308994')
-    getRemoteStatus(bug_id=u'327452')
-    getRemoteStatus(bug_id=u'327549')
-    == email ==
-    DEBUG No watches to update on mailto:[e-mail]
-    == example-bugs ==
-    DEBUG No watches to update on http://bugs.example.com
-    == gnome-bugs ==
-    DEBUG No watches to update on http://bugzilla.gnome.org/
-    == gnome-bugzilla ==
-    INFO Updating 2 watches for 2 bugs on http://example.com
-    getRemoteStatus(bug_id=u'304070')
-    getRemoteStatus(bug_id=u'3224')
-    == mozilla.org ==
-    INFO Updating 4 watches for 3 bugs on http://example.com
-    getRemoteStatus(bug_id=u'123543')
-    getRemoteStatus(bug_id=u'2000')
-    getRemoteStatus(bug_id=u'42')
-    == savannah ==
-    DEBUG No watches to update on http://savannah.gnu.org/
-    == sf ==
-    DEBUG No watches to update on http://sourceforge.net/

=== modified file 'lib/lp/bugs/scripts/checkwatches.py'
--- lib/lp/bugs/scripts/checkwatches.py	2010-02-19 12:05:10 +0000
+++ lib/lp/bugs/scripts/checkwatches.py	2010-03-15 14:09:31 +0000
@@ -6,7 +6,6 @@
 
 from copy import copy
 from datetime import datetime, timedelta
-import Queue as queue
 import socket
 import sys
 import threading
@@ -14,6 +13,11 @@
 
 import pytz
 
+from twisted.internet import reactor
+from twisted.internet.defer import DeferredList
+from twisted.internet.threads import deferToThreadPool
+from twisted.python.threadpool import ThreadPool
+
 from zope.component import getUtility
 from zope.event import notify
 
@@ -271,44 +275,28 @@
         self._logout()
 
     def updateBugTrackers(
-        self, bug_tracker_names=None, batch_size=None, num_threads=1):
+        self, bug_tracker_names=None, batch_size=None, scheduler=None):
         """Update all the bug trackers that have watches pending.
 
         If bug tracker names are specified in bug_tracker_names only
         those bug trackers will be checked.
 
-        The updates are run in threads, so that long running updates
-        don't block progress. However, by default the number of
-        threads is 1, to help with testing.
+        A custom scheduler can be passed in. This should inherit from
+        `BaseScheduler`. If no scheduler is given, `SerialScheduler`
+        will be used, which simply runs the jobs in order.
         """
         self.log.debug("Using a global batch size of %s" % batch_size)
 
-        # Put all the work on the queue. This is simpler than drip-feeding the
-        # queue, and avoids a situation where a worker thread exits because
-        # there's no work left and the feeding thread hasn't been scheduled to
-        # add work to the queue.
-        work = queue.Queue()
+        # Default to using the very simple SerialScheduler.
+        if scheduler is None:
+            scheduler = SerialScheduler()
+
+        # Schedule all the jobs to run.
         for updater in self._bugTrackerUpdaters(bug_tracker_names):
-            work.put(updater)
-
-        # This will be run once in each worker thread.
-        def do_work():
-            while True:
-                try:
-                    job = work.get(block=False)
-                except queue.Empty:
-                    break
-                else:
-                    job(batch_size)
-
-        # Start and join the worker threads.
-        threads = []
-        for run in xrange(num_threads):
-            thread = threading.Thread(target=do_work)
-            thread.start()
-            threads.append(thread)
-        for thread in threads:
-            thread.join()
+            scheduler.schedule(updater, batch_size)
+
+        # Run all the jobs.
+        scheduler.run()
 
     def updateBugTracker(self, bug_tracker, batch_size):
         """Updates the given bug trackers's bug watches.
@@ -1165,6 +1153,67 @@
         self.log.error("%s (%s)" % (message, oops_info.oopsid))
 
 
+class BaseScheduler:
+    """Run jobs according to a policy."""
+
+    def schedule(self, func, *args, **kwargs):
+        """Add a job to be run."""
+        raise NotImplementedError(self.schedule)
+
+    def run(self):
+        """Run the jobs."""
+        raise NotImplementedError(self.run)
+
+
+class SerialScheduler(BaseScheduler):
+    """Run jobs in order, one at a time."""
+
+    def __init__(self):
+        self._jobs = []
+
+    def schedule(self, func, *args, **kwargs):
+        self._jobs.append((func, args, kwargs))
+
+    def run(self):
+        jobs, self._jobs = self._jobs[:], []
+        for (func, args, kwargs) in jobs:
+            func(*args, **kwargs)
+
+
+class TwistedThreadScheduler(BaseScheduler):
+    """Run jobs in threads, chaperoned by Twisted."""
+
+    def __init__(self, num_threads, install_signal_handlers=True):
+        """Create a new `TwistedThreadScheduler`.
+
+        :param num_threads: The number of threads to allocate to the
+          thread pool.
+        :type num_threads: int
+
+        :param install_signal_handlers: Whether the Twisted reactor
+          should install signal handlers or not. This is intented for
+          testing - set to False to avoid layer violations - but may
+          be useful in other situations.
+        :type install_signal_handlers: bool
+        """
+        self._thread_pool = ThreadPool(0, num_threads)
+        self._install_signal_handlers = install_signal_handlers
+        self._jobs = []
+
+    def schedule(self, func, *args, **kwargs):
+        self._jobs.append(
+            deferToThreadPool(
+                reactor, self._thread_pool, func, *args, **kwargs))
+
+    def run(self):
+        jobs, self._jobs = self._jobs[:], []
+        jobs_done = DeferredList(jobs)
+        jobs_done.addBoth(lambda ignore: self._thread_pool.stop())
+        jobs_done.addBoth(lambda ignore: reactor.stop())
+        reactor.callWhenRunning(self._thread_pool.start)
+        reactor.run(self._install_signal_handlers)
+
+
 class CheckWatchesCronScript(LaunchpadCronScript):
 
     def add_my_options(self):
@@ -1203,9 +1252,16 @@
         else:
             # Otherwise we just update those watches that need updating,
             # and we let the BugWatchUpdater decide which those are.
+            if self.options.jobs <= 1:
+                # Use the default scheduler.
+                scheduler = None
+            else:
+                # Run jobs in parallel.
+                scheduler = TwistedThreadScheduler(self.options.jobs)
             updater.updateBugTrackers(
-                self.options.bug_trackers, self.options.batch_size,
-                self.options.jobs)
+                self.options.bug_trackers,
+                self.options.batch_size,
+                scheduler)
 
         run_time = time.time() - start_time
         self.logger.info("Time for this run: %.3f seconds." % run_time)

=== modified file 'lib/lp/bugs/scripts/tests/test_checkwatches.py'
--- lib/lp/bugs/scripts/tests/test_checkwatches.py	2010-01-05 12:19:17 +0000
+++ lib/lp/bugs/scripts/tests/test_checkwatches.py	2010-03-15 14:09:31 +0000
@@ -2,9 +2,13 @@
 # GNU Affero General Public License version 3 (see the file LICENSE).
 """Checkwatches unit tests."""
 
+from __future__ import with_statement
+
 __metaclass__ = type
 
+import threading
 import unittest
+
 import transaction
 
 from zope.component import getUtility
@@ -19,11 +23,13 @@
 from canonical.testing import LaunchpadZopelessLayer
 
 from lp.bugs.externalbugtracker.bugzilla import BugzillaAPI
+from lp.bugs.interfaces.bugtracker import IBugTrackerSet
 from lp.bugs.scripts import checkwatches
-from lp.bugs.scripts.checkwatches import CheckWatchesErrorUtility
+from lp.bugs.scripts.checkwatches import (
+    BugWatchUpdater, CheckWatchesErrorUtility, TwistedThreadScheduler)
 from lp.bugs.tests.externalbugtracker import (
-    TestBugzillaAPIXMLRPCTransport, new_bugtracker)
-from lp.testing import TestCaseWithFactory
+    TestBugzillaAPIXMLRPCTransport, TestExternalBugTracker, new_bugtracker)
+from lp.testing import TestCaseWithFactory, ZopeTestInSubProcess
 
 
 def always_BugzillaAPI_get_external_bugtracker(bugtracker):
@@ -202,5 +208,171 @@
             self.bugtask_with_question.status.title))
 
 
+class TestSchedulerBase:
+
+    def test_args_and_kwargs(self):
+        def func(name, aptitude):
+            self.failUnlessEqual("Robin Hood", name)
+            self.failUnlessEqual("Riding through the glen", aptitude)
+        # Positional args specified when adding a job are passed to
+        # the job function at run time.
+        self.scheduler.schedule(
+            func, "Robin Hood", "Riding through the glen")
+        # Keyword args specified when adding a job are passed to the
+        # job function at run time.
+        self.scheduler.schedule(
+            func, name="Robin Hood", aptitude="Riding through the glen")
+        # Positional and keyword args can both be specified.
+        self.scheduler.schedule(
+            func, "Robin Hood", aptitude="Riding through the glen")
+        # Run everything.
+        self.scheduler.run()
+
+
+class TestSerialScheduler(TestSchedulerBase, unittest.TestCase):
+    """Test SerialScheduler."""
+
+    def setUp(self):
+        self.scheduler = checkwatches.SerialScheduler()
+
+    def test_ordering(self):
+        # The numbers list will be emptied in the order we add jobs to
+        # the scheduler.
+        numbers = [1, 2, 3]
+        # Remove 3 and check.
+        self.scheduler.schedule(
+            list.remove, numbers, 3)
+        self.scheduler.schedule(
+            lambda: self.failUnlessEqual([1, 2], numbers))
+        # Remove 1 and check.
+        self.scheduler.schedule(
+            list.remove, numbers, 1)
+        self.scheduler.schedule(
+            lambda: self.failUnlessEqual([2], numbers))
+        # Remove 2 and check.
+        self.scheduler.schedule(
+            list.remove, numbers, 2)
+        self.scheduler.schedule(
+            lambda: self.failUnlessEqual([], numbers))
+        # Run the scheduler.
+        self.scheduler.run()
+
+
+class TestTwistedThreadScheduler(TestSchedulerBase, unittest.TestCase):
+    """Test TwistedThreadScheduler.
+
+    By default, updateBugTrackers() runs jobs serially, but a
+    different scheduling policy can be plugged in. One such policy,
+    for running several jobs in parallel, is TwistedThreadScheduler.
+    """
+
+    def setUp(self):
+        self.scheduler = checkwatches.TwistedThreadScheduler(
+            num_threads=5, install_signal_handlers=False)
+
+
+class OutputFileForThreads:
+    """Collates writes according to thread name."""
+
+    def __init__(self):
+        self.output = {}
+        self.lock = threading.Lock()
+
+    def write(self, data):
+        thread_name = threading.currentThread().getName()
+        with self.lock:
+            if thread_name in self.output:
+                self.output[thread_name].append(data)
+            else:
+                self.output[thread_name] = [data]
+
+
+class ExternalBugTrackerForThreads(TestExternalBugTracker):
+    """Fake which records interesting activity to a file."""
+
+    def __init__(self, output_file):
+        super(ExternalBugTrackerForThreads, self).__init__()
+        self.output_file = output_file
+
+    def getRemoteStatus(self, bug_id):
+        self.output_file.write("getRemoteStatus(bug_id=%r)" % bug_id)
+        return 'UNKNOWN'
+
+    def getCurrentDBTime(self):
+        return None
+
+
+class BugWatchUpdaterForThreads(BugWatchUpdater):
+    """Fake updater.
+
+    Plumbs an `ExternalBugTrackerForThreads` into a given output file,
+    which is expected to be an instance of `OutputFileForThreads`, and
+    suppresses normal log activity.
+    """
+
+    def __init__(self, output_file):
+        logger = QuietFakeLogger()
+        super(BugWatchUpdaterForThreads, self).__init__(transaction, logger)
+        self.output_file = output_file
+
+    def _getExternalBugTrackersAndWatches(self, bug_trackers, bug_watches):
+        return [(ExternalBugTrackerForThreads(self.output_file), bug_watches)]
+
+
+class TestTwistedThreadSchedulerInPlace(
+    ZopeTestInSubProcess, TestCaseWithFactory):
+    """Test TwistedThreadScheduler in place.
+
+    As in, driving as much of the bug watch machinery as is possible
+    without making external connections.
+    """
+
+    layer = LaunchpadZopelessLayer
+
+    def test(self):
+        # Prepare test data.
+        self.owner = self.factory.makePerson()
+        self.trackers = [
+            getUtility(IBugTrackerSet).ensureBugTracker(
+                "http://butterscotch.example.com", self.owner,
+                BugTrackerType.BUGZILLA, name="butterscotch"),
+            getUtility(IBugTrackerSet).ensureBugTracker(
+                "http://strawberry.example.com", self.owner,
+                BugTrackerType.BUGZILLA, name="strawberry"),
+            ]
+        self.bug = self.factory.makeBug(owner=self.owner)
+        for tracker in self.trackers:
+            for num in (1, 2, 3):
+                self.factory.makeBugWatch(
+                    "%s-%d" % (tracker.name, num),
+                    tracker, self.bug, self.owner)
+        # Commit so that threads all see the same database state.
+        transaction.commit()
+        # Prepare the updater with the Twisted scheduler.
+        output_file = OutputFileForThreads()
+        threaded_bug_watch_updater = BugWatchUpdaterForThreads(output_file)
+        threaded_bug_watch_scheduler = TwistedThreadScheduler(
+            num_threads=10, install_signal_handlers=False)
+        threaded_bug_watch_updater.updateBugTrackers(
+            bug_tracker_names=[tracker.name for tracker in self.trackers],
+            batch_size=5, scheduler=threaded_bug_watch_scheduler)
+        # The thread names should match the tracker names.
+        self.assertEqual(
+            ['butterscotch', 'strawberry'], sorted(output_file.output))
+        # Check that getRemoteStatus() was called.
+        self.assertEqual(
+            ["getRemoteStatus(bug_id=u'butterscotch-1')",
+             "getRemoteStatus(bug_id=u'butterscotch-2')",
+             "getRemoteStatus(bug_id=u'butterscotch-3')"],
+            output_file.output['butterscotch']
+            )
+        self.assertEqual(
+            ["getRemoteStatus(bug_id=u'strawberry-1')",
+             "getRemoteStatus(bug_id=u'strawberry-2')",
+             "getRemoteStatus(bug_id=u'strawberry-3')"],
+            output_file.output['strawberry']
+            )
+
+
 def test_suite():
     return unittest.TestLoader().loadTestsFromName(__name__)

=== modified file 'lib/lp/scripts/utilities/importfascist.py'
--- lib/lp/scripts/utilities/importfascist.py	2010-02-23 11:53:24 +0000
+++ lib/lp/scripts/utilities/importfascist.py	2010-03-15 14:09:31 +0000
@@ -62,6 +62,7 @@
     'openid.fetchers': set(['Urllib2Fetcher']),
     'storm.database': set(['STATE_DISCONNECTED']),
     'textwrap': set(['dedent']),
+    'twisted.internet.threads': set(['deferToThreadPool']),
     'zope.component': set(
         ['adapter',
          'ComponentLookupError',

=== modified file 'lib/lp/services/job/runner.py'
--- lib/lp/services/job/runner.py	2010-02-25 12:07:15 +0000
+++ lib/lp/services/job/runner.py	2010-03-15 14:09:31 +0000
@@ -401,9 +401,10 @@
 class JobCronScript(LaunchpadCronScript):
     """Base class for scripts that run jobs."""
 
-    def __init__(self, runner_class=JobRunner):
+    def __init__(self, runner_class=JobRunner, test_args=None):
         self.dbuser = getattr(config, self.config_name).dbuser
-        super(JobCronScript, self).__init__(self.config_name, self.dbuser)
+        super(JobCronScript, self).__init__(
+            self.config_name, self.dbuser, test_args)
         self.runner_class = runner_class
 
     def main(self):

=== modified file 'lib/lp/services/job/tests/test_runner.py'
--- lib/lp/services/job/tests/test_runner.py	2010-02-24 19:52:01 +0000
+++ lib/lp/services/job/tests/test_runner.py	2010-03-15 14:09:31 +0000
@@ -11,23 +11,24 @@
 from unittest import TestLoader
 
 import transaction
-from canonical.testing import LaunchpadZopelessLayer
+
 from zope.component import getUtility
 from zope.error.interfaces import IErrorReportingUtility
 from zope.interface import implements
 
+from canonical.launchpad.webapp import errorlog
+from canonical.launchpad.webapp.interfaces import (
+    DEFAULT_FLAVOR, IStoreSelector, MAIN_STORE)
+from canonical.testing import LaunchpadZopelessLayer
+
 from lp.code.interfaces.branchmergeproposal import (
-    IUpdatePreviewDiffJobSource,)
-from lp.testing.mail_helpers import pop_notifications
-from lp.services.job.runner import (
-    JobCronScript, JobRunner, BaseRunnableJob, TwistedJobRunner
-)
+    IUpdatePreviewDiffJobSource)
 from lp.services.job.interfaces.job import JobStatus, IRunnableJob
 from lp.services.job.model.job import Job
-from lp.testing import TestCaseWithFactory
-from canonical.launchpad.webapp import errorlog
-from canonical.launchpad.webapp.interfaces import (
-    IStoreSelector, MAIN_STORE, DEFAULT_FLAVOR)
+from lp.services.job.runner import (
+    BaseRunnableJob, JobCronScript, JobRunner, TwistedJobRunner)
+from lp.testing import TestCaseWithFactory, ZopeTestInSubProcess
+from lp.testing.mail_helpers import pop_notifications
 
 
 class NullJob(BaseRunnableJob):
@@ -300,7 +301,7 @@
         self.entries.append(input)
 
 
-class TestTwistedJobRunner(TestCaseWithFactory):
+class TestTwistedJobRunner(ZopeTestInSubProcess, TestCaseWithFactory):
 
     layer = LaunchpadZopelessLayer
 
@@ -325,7 +326,7 @@
         self.assertIn('Job ran too long.', oops.value)
 
 
-class TestJobCronScript(TestCaseWithFactory):
+class TestJobCronScript(ZopeTestInSubProcess, TestCaseWithFactory):
 
     layer = LaunchpadZopelessLayer
 
@@ -351,7 +352,8 @@
             source_interface = IUpdatePreviewDiffJobSource
 
             def __init__(self):
-                super(JobCronScriptSubclass, self).__init__(DummyRunner)
+                super(JobCronScriptSubclass, self).__init__(
+                    DummyRunner, test_args=[])
                 self.logger = ListLogger()
 
         old_errorlog = errorlog.globalErrorUtility

=== modified file 'lib/lp/testing/__init__.py'
--- lib/lp/testing/__init__.py	2010-03-01 03:06:02 +0000
+++ lib/lp/testing/__init__.py	2010-03-15 14:09:31 +0000
@@ -29,6 +29,7 @@
     'validate_mock_class',
     'WindmillTestCase',
     'with_anonymous_login',
+    'ZopeTestInSubProcess',
     ]
 
 import copy
@@ -38,6 +39,8 @@
 from pprint import pformat
 import shutil
 import subprocess
+import subunit
+import sys
 import tempfile
 import time
 
@@ -62,6 +65,7 @@
 from zope.interface.verify import verifyClass, verifyObject
 from zope.security.proxy import (
     isinstance as zope_isinstance, removeSecurityProxy)
+from zope.testing.testrunner.runner import TestResult as ZopeTestResult
 
 from canonical.launchpad.webapp import errorlog
 from canonical.config import config
@@ -586,6 +590,57 @@
         self.client.open(url=u'http://launchpad.dev:8085')
 
 
+class ZopeTestInSubProcess:
+    """Run tests in a sub-process, respecting Zope idiosyncrasies.
+
+    Use this as a mixin with an interesting `TestCase` to isolate
+    tests with side-effects. Each and every test *method* in the test
+    case is run in a new, forked, sub-process. This will slow down
+    your tests, so use it sparingly. However, when you need to, for
+    example, start the Twisted reactor (which cannot currently be
+    safely stopped and restarted in process) it is invaluable.
+
+    This is basically a reimplementation of subunit's
+    `IsolatedTestCase` or `IsolatedTestSuite`, but adjusted to work
+    with Zope. In particular, Zope's TestResult object is responsible
+    for calling testSetUp() and testTearDown() on the selected layer.
+    """
+
+    def run(self, result):
+        assert isinstance(result, ZopeTestResult), (
+            "result must be a Zope result object, not %r." % (result,))
+        pread, pwrite = os.pipe()
+        pid = os.fork()
+        if pid == 0:
+            # Child.
+            os.close(pread)
+            fdwrite = os.fdopen(pwrite, 'w', 1)
+            # Send results to both the Zope result object (so that
+            # layer setup and teardown are done properly, etc.) and to
+            # the subunit stream client so that the parent process can
+            # obtain the result.
+            result = testtools.MultiTestResult(
+                result, subunit.TestProtocolClient(fdwrite))
+            super(ZopeTestInSubProcess, self).run(result)
+            fdwrite.flush()
+            sys.stdout.flush()
+            sys.stderr.flush()
+            # Exit hard.
+            os._exit(0)
+        else:
+            # Parent.
+            os.close(pwrite)
+            fdread = os.fdopen(pread, 'rU')
+            # Accept the result from the child process. Skip all the
+            # Zope-specific result stuff by passing a super() of the
+            # result.
+            result = super(ZopeTestResult, result)
+            protocol = subunit.TestProtocolServer(result)
+            protocol.readFrom(fdread)
+            fdread.close()
+            os.waitpid(pid, 0)
+
+
 def capture_events(callable_obj, *args, **kwargs):
     """Capture the events emitted by a callable.
 

=== added file 'lib/lp/testing/tests/test_zope_test_in_subprocess.py'
--- lib/lp/testing/tests/test_zope_test_in_subprocess.py	1970-01-01 00:00:00 +0000
+++ lib/lp/testing/tests/test_zope_test_in_subprocess.py	2010-03-15 14:09:31 +0000
@@ -0,0 +1,131 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test `lp.testing.ZopeTestInSubProcess`.
+
+How does it do this?
+
+A `TestCase`, mixed-in with `ZopeTestInSubProcess`, is run by the Zope
+test runner. This test case sets its own layer, to keep track of the
+PIDs when certain methods are called. It also records pids for its own
+methods. Assertions are made as these methods are called to ensure that
+they are running in the correct process - the parent or the child.
+
+Recording of the PIDs is handled using the `record_pid` decorator.
+"""
+
+__metaclass__ = type
+
+import functools
+import os
+import unittest
+
+from lp.testing import ZopeTestInSubProcess
+
+
+def record_pid(method):
+    """Decorator that records the pid at method invocation.
+
+    Will probably only DTRT with class methods or bound instance
+    methods.
+    """
+    @functools.wraps(method)
+    def wrapper(self, *args, **kwargs):
+        setattr(self, 'pid_in_%s' % method.__name__, os.getpid())
+        return method(self, *args, **kwargs)
+    return wrapper
+
+
+class TestZopeTestInSubProcessLayer:
+    """Helper to test `ZopeTestInSubProcess`.
+
+    Asserts that layers are set up and torn down in the expected way,
+    namely that setUp() and tearDown() are called in the parent
+    process, and testSetUp() and testTearDown() are called in the
+    child process.
+
+    The assertions for tearDown() and testTearDown() must be done here
+    because the test case runs before these methods are called. In the
+    interests of symmetry and clarity, the assertions for setUp() and
+    testSetUp() are done here too.
+
+    This layer expects to be *instantiated*, which is not the norm for
+    Zope layers. See `TestZopeTestInSubProcess` for its use.
+    """
+
+    @record_pid
+    def __init__(self):
+        # These are needed to satisfy the requirements of the
+        # byzantine Zope layer machinery.
+        self.__name__ = self.__class__.__name__
+        self.__bases__ = self.__class__.__bases__
+
+    @record_pid
+    def setUp(self):
+        # Runs in the parent process.
+        assert self.pid_in___init__ == self.pid_in_setUp, (
+            "layer.setUp() not called in parent process.")
+
+    @record_pid
+    def testSetUp(self):
+        # Runs in the child process.
+        assert self.pid_in___init__ != self.pid_in_testSetUp, (
+            "layer.testSetUp() called in parent process.")
+
+    @record_pid
+    def testTearDown(self):
+        # Runs in the child process.
+        assert self.pid_in_testSetUp == self.pid_in_testTearDown, (
+            "layer.testTearDown() not called in same process as testSetUp().")
+
+    @record_pid
+    def tearDown(self):
+        # Runs in the parent process.
+        assert self.pid_in___init__ == self.pid_in_tearDown, (
+            "layer.tearDown() not called in parent process.")
+
+
+class TestZopeTestInSubProcess(ZopeTestInSubProcess, unittest.TestCase):
+    """Test `ZopeTestInSubProcess`.
+
+    Assert that setUp(), test() and tearDown() are called in the child
+    process.
+
+    Sets its own layer attribute. This layer is then responsible for
+    recording the PID at interesting moments. Specifically,
+    layer.testSetUp() must be called in the same process as
+    test.setUp().
+    """
+
+    @record_pid
+    def __init__(self, method_name='runTest'):
+        # Runs in the parent process.
+        super(TestZopeTestInSubProcess, self).__init__(method_name)
+        self.layer = TestZopeTestInSubProcessLayer()
+
+    @record_pid
+    def setUp(self):
+        # Runs in the child process.
+        super(TestZopeTestInSubProcess, self).setUp()
+        self.failUnlessEqual(
+            self.layer.pid_in_testSetUp, self.pid_in_setUp,
+            "setUp() not called in same process as layer.testSetUp().")
+
+    @record_pid
+    def test(self):
+        # Runs in the child process.
+        self.failUnlessEqual(
+            self.pid_in_setUp, self.pid_in_test,
+            "test method not run in same process as setUp().")
+
+    @record_pid
+    def tearDown(self):
+        # Runs in the child process.
+        super(TestZopeTestInSubProcess, self).tearDown()
+        self.failUnlessEqual(
+            self.pid_in_setUp, self.pid_in_tearDown,
+            "tearDown() not run in same process as setUp().")
+
+
+def test_suite():
+    return unittest.TestLoader().loadTestsFromName(__name__)