Merge lp:~mwhudson/launchpad/no-codeimportworker-db-access into lp:launchpad

[Michael Hudson-Doyle]

Have the code import worker monitor talk to the internal xml-rpc server rather than directly to the database.

Hi there,

this branch removes the need for the code import worker monitor script to talk directly to the database, and makes it talk to the internal xml-rpc server instead.

In some sense this isn't very exciting, although the diff is rather large. Much of the change is in the worker monitor tests -- all the rest is fairly straightforward I think.

Cheers,
mwh

[Tim Penhey (thumper)]

There is a lot here, but I think it is all pretty good. Very extensive testing, I bet it wasn't a huge amount of fun to write :)

review: Approve

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2010-03-10 17:55:59 +0000
+++ database/schema/security.cfg	2010-03-14 20:28:32 +0000
@@ -586,28 +586,6 @@
 public.validpersonorteamcache           = SELECT
 public.wikiname                         = SELECT, INSERT
 
-[codeimportworker]
-type=user
-public.branch                           = SELECT, UPDATE
-public.branchsubscription               = SELECT
-public.codeimport                       = SELECT, UPDATE
-public.codeimportevent                  = SELECT, INSERT, UPDATE
-public.codeimporteventdata              = SELECT, INSERT
-public.codeimportmachine                = SELECT
-public.codeimportresult                 = SELECT, INSERT, UPDATE
-public.codeimportjob                    = SELECT, INSERT, UPDATE, DELETE
-public.distribution                     = SELECT
-public.distroseries                     = SELECT
-public.emailaddress                     = SELECT
-public.libraryfilealias                 = SELECT, INSERT, UPDATE
-public.libraryfilecontent               = SELECT, INSERT
-public.person                           = SELECT
-public.product                          = SELECT
-public.productseries                    = SELECT
-public.productseriescodeimport          = SELECT
-public.sourcepackagename                = SELECT
-public.teammembership                   = SELECT
-
 [branchscanner]
 type=user
 groups=write, script

=== modified file 'lib/canonical/config/schema-lazr.conf'
--- lib/canonical/config/schema-lazr.conf	2010-02-24 10:18:16 +0000
+++ lib/canonical/config/schema-lazr.conf	2010-03-14 20:28:32 +0000
@@ -472,8 +472,8 @@
 
 
 [codeimportworker]
-# This code is used by the code-import-worker-db which lives in
-# scripts/code-import-worker-db.py and
+# This code is used by the code-import-worker-monitor which lives in
+# scripts/code-import-worker-monitor.py and
 # lib/lp/codehosting/codeimport/worker_monitor.py.
 
 # The interval in seconds the worker should wait between updates to the

=== modified file 'lib/canonical/launchpad/interfaces/librarian.py'
--- lib/canonical/launchpad/interfaces/librarian.py	2009-11-24 15:36:44 +0000
+++ lib/canonical/launchpad/interfaces/librarian.py	2010-03-14 20:28:32 +0000
@@ -150,7 +150,7 @@
         from the Librarian at this time. See LibrarianGarbageCollection.
 
         If restricted is True, the file will be created through the
-        IRestricteLibrarianClient utility.
+        IRestrictedLibrarianClient utility.
         """
 
     def __getitem__(key):

=== modified file 'lib/canonical/launchpad/xmlrpc/faults.py'
--- lib/canonical/launchpad/xmlrpc/faults.py	2010-02-21 22:13:34 +0000
+++ lib/canonical/launchpad/xmlrpc/faults.py	2010-03-14 20:28:32 +0000
@@ -27,6 +27,7 @@
     'NoLinkedBranch',
     'NoSuchBranch',
     'NoSuchBug',
+    'NoSuchCodeImportJob',
     'NoSuchDistribution',
     'NoSuchPackage',
     'NoSuchPerson',
@@ -474,3 +475,12 @@
         self.sourcepackagename = sourcepackagename
         LaunchpadFault.__init__(self, sourcepackagename=sourcepackagename)
 
+
+class NoSuchCodeImportJob(LaunchpadFault):
+    """Raised by `ICodeImportScheduler` methods when a job is not found."""
+
+    error_code = 360
+    msg_template = 'Job %(job_id)d not found.'
+
+    def __init__(self, job_id):
+        LaunchpadFault.__init__(self, job_id=job_id)

=== modified file 'lib/lp/code/interfaces/codeimportscheduler.py'
--- lib/lp/code/interfaces/codeimportscheduler.py	2010-02-22 08:10:03 +0000
+++ lib/lp/code/interfaces/codeimportscheduler.py	2010-03-14 20:28:32 +0000
@@ -16,6 +16,7 @@
 
 from zope.interface import Interface
 
+
 class ICodeImportSchedulerApplication(ILaunchpadApplication):
     """Code import scheduler application root."""
 
@@ -35,3 +36,28 @@
         mark it as having started on said machine and return its id,
         or 0 if there are no jobs pending.
         """
+
+    def getImportDataForJobID(job_id):
+        """Get data about the import with job id `job_id`.
+
+        :return: ``(worker_arguments, branch_url, log_file_name)`` where:
+            * ``worker_arguments`` are the arguments to pass to the code
+              import worker subprocess.
+           * ``branch_url`` is the URL of the import branch (only used to put
+             in OOPS reports)
+           * ``log_file_name`` is the name of the log file to create in the
+             librarian.
+        :raise NoSuchCodeImportJob: if no job with id `job_id` exists.
+        """
+
+    def updateHeartbeat(job_id, log_tail):
+        """Call `ICodeImportJobWorkflow.updateHeartbeat` for job `job_id`.
+
+        :raise NoSuchCodeImportJob: if no job with id `job_id` exists.
+        """
+
+    def finishJobID(job_id, status_name, log_file_alias_url):
+        """Call `ICodeImportJobWorkflow.finishJob` for job `job_id`.
+
+        :raise NoSuchCodeImportJob: if no job with id `job_id` exists.
+        """

=== modified file 'lib/lp/code/model/tests/test_codeimportjob.py'
--- lib/lp/code/model/tests/test_codeimportjob.py	2010-02-24 13:17:01 +0000
+++ lib/lp/code/model/tests/test_codeimportjob.py	2010-03-14 20:28:32 +0000
@@ -14,6 +14,8 @@
 
 from sqlobject.sqlbuilder import SQLConstant
 
+import transaction
+
 from twisted.python.util import mergeFunctionMetadata
 
 from zope.component import getUtility
@@ -702,24 +704,13 @@
         AssertFailureMixin, AssertEventMixin):
     """Unit tests for CodeImportJobWorkflow.finishJob."""
 
-    layer = LaunchpadZopelessLayer
+    layer = LaunchpadFunctionalLayer
 
     def setUp(self):
         super(TestCodeImportJobWorkflowFinishJob, self).setUp()
         login_for_code_imports()
         self.machine = self.factory.makeCodeImportMachine()
         self.machine.setOnline()
-        self.switchDbUser_called = False
-
-    def tearDown(self):
-        super(TestCodeImportJobWorkflowFinishJob, self).tearDown()
-        self.assertTrue(
-            self.switchDbUser_called, "switchDbUser() not called!")
-
-    def switchDbUser(self):
-        self.layer.txn.commit()
-        self.layer.switchDbUser('codeimportworker')
-        self.switchDbUser_called = True
 
     def makeRunningJob(self, code_import=None):
         """Make and return a CodeImportJob object with state==RUNNING.
@@ -740,7 +731,6 @@
         # Calling finishJob with a job whose state is not RUNNING is an error.
         code_import = self.factory.makeCodeImport()
         job = self.factory.makeCodeImportJob(code_import)
-        self.switchDbUser()
         self.assertFailure(
             "The CodeImportJob associated with %s is "
             "PENDING." % code_import.branch.unique_name,
@@ -754,7 +744,6 @@
         # finishJob() deletes the job it is passed.
         running_job = self.makeRunningJob()
         running_job_id = running_job.id
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.SUCCESS, None)
         self.assertEqual(
@@ -765,7 +754,6 @@
         # scheduled appropriately far in the future.
         running_job = self.makeRunningJob()
         code_import = running_job.code_import
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.SUCCESS, None)
         new_job = code_import.import_job
@@ -782,7 +770,6 @@
         # now.
         running_job = self.makeRunningJob()
         code_import = running_job.code_import
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.SUCCESS_PARTIAL, None)
         new_job = code_import.import_job
@@ -800,7 +787,6 @@
             '[e-mail]')
         code_import.updateFromData(
             {'review_status': CodeImportReviewStatus.SUSPENDED}, ddaa)
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.SUCCESS, None)
         self.assertTrue(code_import.import_job is None)
@@ -812,7 +798,6 @@
         # Before calling finishJob() there are no CodeImportResults for the
         # given import...
         self.assertEqual(len(list(code_import.results)), 0)
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.SUCCESS, None)
         # ... and after, there is exactly one.
@@ -841,13 +826,11 @@
         # methods -- e.g. calling requestJob to set requesting_user -- but
         # using removeSecurityProxy and forcing here is expedient.
         setattr(removeSecurityProxy(job), from_field, value)
-        self.switchDbUser()
         result = self.getResultForJob(job)
         self.assertEqual(
             value, getattr(result, to_field),
             "Value %r in job field %r was not passed through to result field"
             " %r." % (value, from_field, to_field))
-        self.layer.switchDbUser('launchpad')
 
     def test_resultObjectFields(self):
         # The CodeImportResult object that finishJob creates contains all the
@@ -898,7 +881,6 @@
         status_jobs = []
         for status in CodeImportResultStatus.items:
             status_jobs.append((status, self.makeRunningJob()))
-        self.switchDbUser()
         for status, job in status_jobs:
             result = self.getResultForJob(job, status)
             self.assertEqual(result.status, status)
@@ -909,12 +891,11 @@
 
         job = self.makeRunningJob()
 
-        self.switchDbUser()
         log_data = 'several\nlines\nof\nlog data'
         log_alias_id = getUtility(ILibrarianClient).addFile(
            'import_log.txt', len(log_data),
            StringIO.StringIO(log_data), 'text/plain')
-        self.layer.txn.commit()
+        transaction.commit()
         log_alias = getUtility(ILibraryFileAliasSet)[log_alias_id]
         result = self.getResultForJob(job, log_alias=log_alias)
 
@@ -927,7 +908,6 @@
         code_import = running_job.code_import
         machine = running_job.machine
         new_events = NewEvents()
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.SUCCESS, None)
         [finish_event] = list(new_events)
@@ -941,7 +921,6 @@
         status_jobs = []
         for status in CodeImportResultStatus.items:
             status_jobs.append((status, self.makeRunningJob()))
-        self.switchDbUser()
         for status, job in status_jobs:
             code_import = job.code_import
             self.assertTrue(code_import.date_last_successful is None)
@@ -958,7 +937,6 @@
         status_jobs = []
         for status in CodeImportResultStatus.items:
             status_jobs.append((status, self.makeRunningJob()))
-        self.switchDbUser()
         for status, job in status_jobs:
             code_import = job.code_import
             self.assertTrue(code_import.date_last_successful is None)
@@ -984,7 +962,6 @@
         self.assertEqual(
             CodeImportReviewStatus.REVIEWED, code_import.review_status)
         running_job = self.makeRunningJob(code_import)
-        self.switchDbUser()
         getUtility(ICodeImportJobWorkflow).finishJob(
             running_job, CodeImportResultStatus.FAILURE, None)
         self.assertEqual(

=== modified file 'lib/lp/code/xmlrpc/codeimportscheduler.py'
--- lib/lp/code/xmlrpc/codeimportscheduler.py	2010-02-19 03:15:04 +0000
+++ lib/lp/code/xmlrpc/codeimportscheduler.py	2010-03-14 20:28:32 +0000
@@ -8,13 +8,20 @@
     'CodeImportSchedulerAPI',
     ]
 
-
-from lp.code.interfaces.codeimportjob import ICodeImportJobSet
-from lp.code.interfaces.codeimportscheduler import ICodeImportScheduler
-from canonical.launchpad.webapp import LaunchpadXMLRPCView
-
 from zope.component import getUtility
 from zope.interface import implements
+from zope.security.proxy import removeSecurityProxy
+
+from canonical.launchpad.interfaces import ILibraryFileAliasSet
+from canonical.launchpad.webapp import canonical_url, LaunchpadXMLRPCView
+from canonical.launchpad.xmlrpc.faults import NoSuchCodeImportJob
+from canonical.launchpad.xmlrpc.helpers import return_fault
+
+from lp.code.enums import CodeImportResultStatus
+from lp.code.interfaces.codeimportjob import (
+    ICodeImportJobSet, ICodeImportJobWorkflow)
+from lp.code.interfaces.codeimportscheduler import ICodeImportScheduler
+from lp.codehosting.codeimport.worker import CodeImportSourceDetails
 
 
 class CodeImportSchedulerAPI(LaunchpadXMLRPCView):
@@ -30,3 +37,57 @@
             return job.id
         else:
             return 0
+
+    def _getJob(self, job_id):
+        job_set = removeSecurityProxy(getUtility(ICodeImportJobSet))
+        job = removeSecurityProxy(job_set.getById(job_id))
+        if job is None:
+            raise NoSuchCodeImportJob(job_id)
+        return job
+
+    # Because you can't use a decorated function as the implementation of a
+    # method exported over XML-RPC, the implementations just thunk to an
+    # implementation wrapped with @return_fault.
+
+    def getImportDataForJobID(self, job_id):
+        """See `ICodeImportScheduler`."""
+        return self._getImportDataForJobID(job_id)
+
+    def updateHeartbeat(self, job_id, log_tail):
+        """See `ICodeImportScheduler`."""
+        return self._updateHeartbeat(job_id, log_tail)
+
+    def finishJobID(self, job_id, status_name, log_file_alias_url):
+        """See `ICodeImportScheduler`."""
+        return self._finishJobID(job_id, status_name, log_file_alias_url)
+
+    @return_fault
+    def _getImportDataForJobID(self, job_id):
+        job = self._getJob(job_id)
+        arguments = CodeImportSourceDetails.fromCodeImport(
+            job.code_import).asArguments()
+        branch = job.code_import.branch
+        branch_url = canonical_url(branch)
+        log_file_name = '%s.log' % branch.unique_name[1:].replace('/', '-')
+        return (arguments, branch_url, log_file_name)
+
+    @return_fault
+    def _updateHeartbeat(self, job_id, log_tail):
+        job = self._getJob(job_id)
+        workflow = removeSecurityProxy(getUtility(ICodeImportJobWorkflow))
+        workflow.updateHeartbeat(job, log_tail)
+        return 0
+
+    @return_fault
+    def _finishJobID(self, job_id, status_name, log_file_alias_url):
+        job = self._getJob(job_id)
+        status = CodeImportResultStatus.items[status_name]
+        workflow = removeSecurityProxy(getUtility(ICodeImportJobWorkflow))
+        if log_file_alias_url:
+            library_file_alias_set = getUtility(ILibraryFileAliasSet)
+            # XXX This is so so so terrible:
+            log_file_alias_id = int(log_file_alias_url.split('/')[-2])
+            log_file_alias = library_file_alias_set[log_file_alias_id]
+        else:
+            log_file_alias = None
+        workflow.finishJob(job, status, log_file_alias)

=== added file 'lib/lp/code/xmlrpc/tests/test_codeimportscheduler.py'
--- lib/lp/code/xmlrpc/tests/test_codeimportscheduler.py	1970-01-01 00:00:00 +0000
+++ lib/lp/code/xmlrpc/tests/test_codeimportscheduler.py	2010-03-14 20:28:32 +0000
@@ -0,0 +1,140 @@
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Test for the methods of `ICodeImportScheduler`."""
+
+__metaclass__ = type
+
+import unittest
+import xmlrpclib
+
+from zope.component import getUtility
+from zope.security.proxy import removeSecurityProxy
+
+from canonical.database.constants import UTC_NOW
+from canonical.launchpad.interfaces import ILaunchpadCelebrities
+from canonical.launchpad.webapp import canonical_url
+from canonical.launchpad.xmlrpc.faults import NoSuchCodeImportJob
+from canonical.launchpad.testing.codeimporthelpers import make_running_import
+from canonical.testing.layers import LaunchpadFunctionalLayer
+
+from lp.code.enums import CodeImportResultStatus
+from lp.code.model.codeimportjob import CodeImportJob
+from lp.code.xmlrpc.codeimportscheduler import CodeImportSchedulerAPI
+from lp.codehosting.codeimport.worker import CodeImportSourceDetails
+from lp.testing import run_with_login, TestCaseWithFactory
+
+
+class TestCodeImportSchedulerAPI(TestCaseWithFactory):
+
+    layer = LaunchpadFunctionalLayer
+
+    def setUp(self):
+        TestCaseWithFactory.setUp(self)
+        self.api = CodeImportSchedulerAPI(None, None)
+        self.machine = self.factory.makeCodeImportMachine(set_online=True)
+        for job in CodeImportJob.select():
+            job.destroySelf()
+
+    def makeCodeImportJob(self, running):
+        person = getUtility(ILaunchpadCelebrities).vcs_imports.teamowner
+        if running:
+            return removeSecurityProxy(run_with_login(person, make_running_import)).import_job
+        else:
+            return run_with_login(person, self.factory.makeCodeImportJob)
+
+    def test_getJobForMachine_no_job_waiting(self):
+        # If no job is waiting getJobForMachine returns 0.
+        job_id = self.api.getJobForMachine(self.machine.hostname, 10)
+        self.assertEqual(0, job_id)
+
+    def test_getJobForMachine_job_waiting(self):
+        # If a job is waiting getJobForMachine returns its id.
+        code_import_job = self.makeCodeImportJob(running=False)
+        job_id = self.api.getJobForMachine(self.machine.hostname, 10)
+        self.assertEqual(code_import_job.id, job_id)
+
+    def test_getImportDataForJobID(self):
+        # getImportDataForJobID returns the worker arguments, branch url and
+        # log file name for an import corresponding to a particular job.
+        code_import_job = self.makeCodeImportJob(running=True)
+        code_import = removeSecurityProxy(code_import_job).code_import
+        code_import_arguments, branch_url, log_file_name = \
+            self.api.getImportDataForJobID(code_import_job.id)
+        import_as_arguments = CodeImportSourceDetails.fromCodeImport(
+            code_import).asArguments()
+        expected_log_file_name = '%s.log' % (
+            code_import.branch.unique_name[1:].replace('/', '-'))
+        self.assertEqual(
+            (import_as_arguments, canonical_url(code_import.branch),
+             expected_log_file_name),
+            (code_import_arguments, branch_url, log_file_name))
+
+    def test_getImportDataForJobID_not_found(self):
+        # getImportDataForJobID returns a NoSuchCodeImportJob fault when there
+        # is no code import job with the given ID.
+        fault = self.api.getImportDataForJobID(-1)
+        self.assertTrue(
+            isinstance(fault, xmlrpclib.Fault),
+            "getImportDataForJobID(-1) returned %r, not a Fault."
+            % (fault,))
+        self.assertEqual(NoSuchCodeImportJob, fault.__class__)
+
+    def test_updateHeartbeat(self):
+        # updateHeartbeat calls the updateHeartbeat job workflow method.
+        code_import_job = self.makeCodeImportJob(running=True)
+        log_tail = self.factory.getUniqueString()
+        self.api.updateHeartbeat(code_import_job.id, log_tail)
+        self.assertSqlAttributeEqualsDate(
+            code_import_job, 'heartbeat', UTC_NOW)
+        self.assertEqual(log_tail, code_import_job.logtail)
+
+    def test_updateHeartbeat_not_found(self):
+        # updateHeartbeat returns a NoSuchCodeImportJob fault when there is no
+        # code import job with the given ID.
+        fault = self.api.updateHeartbeat(-1, '')
+        self.assertTrue(
+            isinstance(fault, xmlrpclib.Fault),
+            "updateHeartbeat(-1, '') returned %r, not a Fault."
+            % (fault,))
+        self.assertEqual(NoSuchCodeImportJob, fault.__class__)
+
+    def test_finishJobID_no_log_file(self):
+        # finishJobID calls the finishJobID job workflow method.  Passing ''
+        # means no log file was uploaded to the librarian.
+        code_import_job = self.makeCodeImportJob(running=True)
+        code_import = code_import_job.code_import
+        self.api.finishJobID(
+            code_import_job.id, CodeImportResultStatus.SUCCESS.name, '')
+        # finishJob does many things, we just check one of them: setting
+        # date_last_successful in the case of success.
+        self.assertSqlAttributeEqualsDate(
+            code_import, 'date_last_successful', UTC_NOW)
+
+    def test_finishJobID_with_log_file(self):
+        # finishJobID calls the finishJobID job workflow method and can parse
+        # a librarian file's http url to figure out its ID.
+        code_import_job = self.makeCodeImportJob(running=True)
+        code_import = code_import_job.code_import
+        log_file_alias = self.factory.makeLibraryFileAlias()
+        self.api.finishJobID(
+            code_import_job.id, CodeImportResultStatus.SUCCESS.name,
+            log_file_alias.http_url)
+        self.assertEqual(
+            log_file_alias, code_import.results[-1].log_file)
+
+    def test_finishJobID_not_found(self):
+        # getImportDataForJobID returns a NoSuchCodeImportJob fault when there
+        # is no code import job with the given ID.
+        fault = self.api.finishJobID(
+            -1, CodeImportResultStatus.SUCCESS.name, '')
+        self.assertTrue(
+            isinstance(fault, xmlrpclib.Fault),
+            "finishJobID(-1, 'SUCCESS', 0) returned %r, not a Fault."
+            % (fault,))
+        self.assertEqual(NoSuchCodeImportJob, fault.__class__)
+
+
+def test_suite():
+    return unittest.TestLoader().loadTestsFromName(__name__)
+

=== modified file 'lib/lp/codehosting/codeimport/dispatcher.py'
--- lib/lp/codehosting/codeimport/dispatcher.py	2010-02-22 05:07:23 +0000
+++ lib/lp/codehosting/codeimport/dispatcher.py	2010-03-14 20:28:32 +0000
@@ -31,7 +31,7 @@
     """
 
     worker_script = os.path.join(
-        config.root, 'scripts', 'code-import-worker-db.py')
+        config.root, 'scripts', 'code-import-worker-monitor.py')
 
     def __init__(self, logger, worker_limit, _sleep=time.sleep):
         """Initialize an instance.

=== modified file 'lib/lp/codehosting/codeimport/tests/test_workermonitor.py'
--- lib/lp/codehosting/codeimport/tests/test_workermonitor.py	2010-03-08 22:26:29 +0000
+++ lib/lp/codehosting/codeimport/tests/test_workermonitor.py	2010-03-14 20:28:32 +0000
@@ -13,27 +13,33 @@
 import StringIO
 import tempfile
 import unittest
+import urllib
 
 from bzrlib.branch import Branch
 from bzrlib.tests import TestCase as BzrTestCase
 
 from twisted.internet import defer, error, protocol, reactor, task
+from twisted.python import failure, log
 from twisted.trial.unittest import TestCase as TrialTestCase
+from twisted.web import xmlrpc
+
+import transaction
 
 from zope.component import getUtility
-from zope.security.proxy import removeSecurityProxy
 
 from canonical.config import config
 from canonical.launchpad.scripts.logger import QuietFakeLogger
+from canonical.launchpad.xmlrpc.faults import NoSuchCodeImportJob
 from canonical.testing.layers import (
-    TwistedLayer, TwistedLaunchpadZopelessLayer)
+    TwistedAppServerLayer, TwistedLaunchpadZopelessLayer, TwistedLayer)
+from canonical.twistedsupport import suppress_stderr
 from canonical.twistedsupport.tests.test_processmonitor import (
     makeFailure, ProcessTestsMixin)
+
 from lp.code.enums import (
     CodeImportResultStatus, CodeImportReviewStatus, RevisionControlSystems)
 from lp.code.interfaces.codeimport import ICodeImportSet
-from lp.code.interfaces.codeimportjob import (
-    ICodeImportJobSet, ICodeImportJobWorkflow)
+from lp.code.interfaces.codeimportjob import ICodeImportJobSet
 from lp.code.model.codeimport import CodeImport
 from lp.code.model.codeimportjob import CodeImportJob
 from lp.codehosting import load_optional_plugin
@@ -41,13 +47,12 @@
     CodeImportSourceDetails, CodeImportWorkerExitCode,
     get_default_bazaar_branch_store)
 from lp.codehosting.codeimport.workermonitor import (
-    CodeImportWorkerMonitor, CodeImportWorkerMonitorProtocol, ExitQuietly,
-    read_only_transaction)
+    CodeImportWorkerMonitor, CodeImportWorkerMonitorProtocol, ExitQuietly)
 from lp.codehosting.codeimport.tests.servers import (
     CVSServer, GitServer, MercurialServer, SubversionServer)
 from lp.codehosting.codeimport.tests.test_worker import (
     clean_up_default_stores_for_import)
-from lp.testing import login, logout
+from lp.testing import login, logout, TestCase
 from lp.testing.factory import LaunchpadObjectFactory
 
 
@@ -129,7 +134,45 @@
             self.protocol._tail, 'line 3\nline 4\nline 5\nline 6\n')
 
 
-class TestWorkerMonitorUnit(TrialTestCase):
+class FakeCodeImportScheduleEndpointProxy:
+    """A fake implementation of a proxy to `ICodeImportScheduler`.
+
+    The constructor takes a dictionary mapping job ids to information that
+    should be returned by getImportDataForJobID and the exception to raise if
+    getImportDataForJobID is called with a job id not in the passed-in
+    dictionary, defaulting to a fault with the same code as
+    NoSuchCodeImportJob (because the class of the exception is lost when you
+    go through XML-RPC serialization).
+    """
+
+    def __init__(self, jobs_dict, no_such_job_exception=None):
+        self.calls = []
+        self.jobs_dict = jobs_dict
+        if no_such_job_exception is None:
+            no_such_job_exception = xmlrpc.Fault(
+                faultCode=NoSuchCodeImportJob.error_code, faultString='')
+        self.no_such_job_exception = no_such_job_exception
+
+    def callRemote(self, method_name, *args):
+        method = getattr(self, '_remote_%s' % method_name, self._default)
+        deferred = defer.maybeDeferred(method, *args)
+        def append_to_log(pass_through):
+            self.calls.append((method_name,) + tuple(args))
+            return pass_through
+        deferred.addCallback(append_to_log)
+        return deferred
+
+    def _default(self, *args):
+        return None
+
+    def _remote_getImportDataForJobID(self, job_id):
+        if job_id in self.jobs_dict:
+            return self.jobs_dict[job_id]
+        else:
+            raise self.no_such_job_exception
+
+
+class TestWorkerMonitorUnit(TrialTestCase, TestCase):
     """Unit tests for most of the `CodeImportWorkerMonitor` class.
 
     We have to pay attention to the fact that several of the methods of the
@@ -140,164 +183,172 @@
 
     layer = TwistedLaunchpadZopelessLayer
 
+    # This works around a clash between the TrialTestCase and our TestCase.
+    skip = None
+
     class WorkerMonitor(CodeImportWorkerMonitor):
         """A subclass of CodeImportWorkerMonitor that stubs logging OOPSes."""
 
         def _logOopsFromFailure(self, failure):
-            self._failures.append(failure)
-
-    def getResultsForOurCodeImport(self):
-        """Return the `CodeImportResult`s for the `CodeImport` we created.
-        """
-        code_import = getUtility(ICodeImportSet).get(self.code_import_id)
-        return code_import.results
-
-    def getOneResultForOurCodeImport(self):
-        """Return the only `CodeImportResult` for the `CodeImport` we created.
-
-        This method fails the test if there is more than one
-        `CodeImportResult` for this `CodeImport`.
-        """
-        results = list(self.getResultsForOurCodeImport())
-        self.failUnlessEqual(len(results), 1)
-        return results[0]
+            log.err(failure)
 
     def assertOopsesLogged(self, exc_types):
-        self.assertEqual(len(exc_types), len(self.worker_monitor._failures))
-        for failure, exc_type in zip(self.worker_monitor._failures,
-                                     exc_types):
-            self.assert_(failure.check(exc_type))
-
-    def setUp(self):
-        login('[e-mail]')
-        self.factory = LaunchpadObjectFactory()
-        job = self.factory.makeCodeImportJob()
-        self.code_import_id = job.code_import.id
-        getUtility(ICodeImportJobWorkflow).startJob(
-            job, self.factory.makeCodeImportMachine(set_online=True))
-        self.job_id = job.id
-        self.worker_monitor = self.WorkerMonitor(job.id, QuietFakeLogger())
-        self.worker_monitor._failures = []
-        self.layer.txn.commit()
-        self.layer.switchDbUser('codeimportworker')
-
-    def tearDown(self):
-        logout()
-
-    def test_getJob(self):
-        # getJob() returns the job whose id we passed to the constructor.
-        return self.assertEqual(
-            self.worker_monitor.getJob().id, self.job_id)
-
-    def test_getJobWhenJobDeleted(self):
-        # If the job has been deleted, getJob sets _call_finish_job to False
-        # and raises ExitQuietly.
-        job = self.worker_monitor.getJob()
-        removeSecurityProxy(job).destroySelf()
-        self.assertRaises(ExitQuietly, self.worker_monitor.getJob)
-        self.assertNot(self.worker_monitor._call_finish_job)
-
-    def test_getSourceDetails(self):
-        # getSourceDetails extracts the details from the CodeImport database
-        # object.
-        @read_only_transaction
-        def check_source_details(details):
-            job = self.worker_monitor.getJob()
-            self.assertEqual(
-                details.url, job.code_import.url)
-            self.assertEqual(
-                details.cvs_root, job.code_import.cvs_root)
-            self.assertEqual(
-                details.cvs_module, job.code_import.cvs_module)
-        return self.worker_monitor.getSourceDetails().addCallback(
-            check_source_details)
+        failures = self.flushLoggedErrors()
+        self.assertEqual(len(exc_types), len(failures))
+        for fail, exc_type in zip(failures, exc_types):
+            self.assert_(fail.check(exc_type))
+
+    def makeWorkerMonitorWithJob(self, job_id=1, job_data=()):
+        return self.WorkerMonitor(
+            job_id, QuietFakeLogger(),
+            FakeCodeImportScheduleEndpointProxy({job_id: job_data}))
+
+    def makeWorkerMonitorWithoutJob(self, exception=None):
+        return self.WorkerMonitor(
+            1, QuietFakeLogger(),
+            FakeCodeImportScheduleEndpointProxy({}, exception))
+
+    def test_getWorkerArguments(self):
+        # getWorkerArguments returns a deferred that fires with the
+        # 'arguments' part of what getImportDataForJobID returns.
+        args = [self.factory.getUniqueString(),
+                self.factory.getUniqueString()]
+        worker_monitor = self.makeWorkerMonitorWithJob(1, (args, 1, 2))
+        return worker_monitor.getWorkerArguments().addCallback(
+            self.assertEqual, args)
+
+    def test_getWorkerArguments_sets_branch_url_and_logfilename(self):
+        # getWorkerArguments sets the _branch_url (for use in oops reports)
+        # and _log_file_name (for upload to the librarian) attributes on the
+        # WorkerMonitor from the data returned by getImportDataForJobID.
+        branch_url = self.factory.getUniqueString()
+        log_file_name = self.factory.getUniqueString()
+        worker_monitor = self.makeWorkerMonitorWithJob(
+            1, (['a'], branch_url, log_file_name))
+        def check_branch_log(ignored):
+            # Looking at the _ attributes here is in slightly poor taste, but
+            # much much easier than them by logging and parsing an oops, etc.
+            self.assertEqual(
+                (branch_url, log_file_name),
+                (worker_monitor._branch_url, worker_monitor._log_file_name))
+        return worker_monitor.getWorkerArguments().addCallback(
+            check_branch_log)
+
+    def test_getWorkerArguments_job_not_found_raises_exit_quietly(self):
+        # When getImportDataForJobID signals a fault indicating that
+        # getWorkerArguments didn't find the supplied job, getWorkerArguments
+        # translates this to an 'ExitQuietly' exception.
+        worker_monitor = self.makeWorkerMonitorWithoutJob()
+        return self.assertFailure(
+            worker_monitor.getWorkerArguments(), ExitQuietly)
+
+    def test_getWorkerArguments_endpoint_failure_raises(self):
+        # When getImportDataForJobID raises an arbitrary exception, it is not
+        # handled in any special way by getWorkerArguments.
+        worker_monitor = self.makeWorkerMonitorWithoutJob(
+            exception=ZeroDivisionError())
+        return self.assertFailure(
+            worker_monitor.getWorkerArguments(), ZeroDivisionError)
+
+    def test_getWorkerArguments_arbitrary_fault_raises(self):
+        # When getImportDataForJobID signals an arbitrary fault, it is not
+        # handled in any special way by getWorkerArguments.
+        worker_monitor = self.makeWorkerMonitorWithoutJob(
+            exception=xmlrpc.Fault(1, ''))
+        return self.assertFailure(
+            worker_monitor.getWorkerArguments(), xmlrpc.Fault)
 
     def test_updateHeartbeat(self):
-        # The worker monitor's updateHeartbeat method calls the
-        # updateHeartbeat job workflow method.
-        @read_only_transaction
+        # updateHeartbeat calls the updateHeartbeat XML-RPC method.
+        log_tail = self.factory.getUniqueString()
+        job_id = self.factory.getUniqueInteger()
+        worker_monitor = self.makeWorkerMonitorWithJob(job_id)
         def check_updated_details(result):
-            job = self.worker_monitor.getJob()
-            self.assertEqual(job.logtail, 'log tail')
-        return self.worker_monitor.updateHeartbeat('log tail').addCallback(
+            self.assertEqual(
+                [('updateHeartbeat', job_id, log_tail)],
+                worker_monitor.codeimport_endpoint.calls)
+        return worker_monitor.updateHeartbeat(log_tail).addCallback(
             check_updated_details)
 
-    def test_finishJobCallsFinishJob(self):
-        # The worker monitor's finishJob method calls the
-        # finishJob job workflow method.
-        @read_only_transaction
+    def test_finishJob_calls_finishJobID_empty_log_file(self):
+        # When the log file is empty, finishJob calls finishJobID with the
+        # name of the status enum and an empty string to indicate that no log
+        # file was uplaoded to the librarian.
+        job_id = self.factory.getUniqueInteger()
+        worker_monitor = self.makeWorkerMonitorWithJob(job_id)
+        self.assertEqual(worker_monitor._log_file.tell(), 0)
         def check_finishJob_called(result):
-            # We take as indication that finishJob was called that a
-            # CodeImportResult was created.
             self.assertEqual(
-                len(list(self.getResultsForOurCodeImport())), 1)
-        return self.worker_monitor.finishJob(
+                [('finishJobID', job_id, 'SUCCESS', '')],
+                worker_monitor.codeimport_endpoint.calls)
+        return worker_monitor.finishJob(
             CodeImportResultStatus.SUCCESS).addCallback(
             check_finishJob_called)
 
-    def test_finishJobDoesntUploadEmptyFileToLibrarian(self):
-        # The worker monitor's finishJob method does not try to upload an
-        # empty log file to the librarian.
-        self.assertEqual(self.worker_monitor._log_file.tell(), 0)
-        @read_only_transaction
-        def check_no_file_uploaded(result):
-            result = self.getOneResultForOurCodeImport()
-            self.assertIdentical(result.log_file, None)
-        return self.worker_monitor.finishJob(
-            CodeImportResultStatus.SUCCESS).addCallback(
-            check_no_file_uploaded)
-
-    def test_finishJobUploadsNonEmptyFileToLibrarian(self):
-        # The worker monitor's finishJob method uploads the log file to the
-        # librarian.
-        self.worker_monitor._log_file.write('some text')
-        @read_only_transaction
+    def test_finishJob_uploads_nonempty_file_to_librarian(self):
+        # finishJob method uploads the log file to the librarian and calls the
+        # finishJobID XML-RPC method with the url of that file.
+        self.layer.force_dirty_database()
+        log_text = self.factory.getUniqueString()
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        worker_monitor._log_file.write(log_text)
         def check_file_uploaded(result):
-            result = self.getOneResultForOurCodeImport()
-            self.assertNotIdentical(result.log_file, None)
-            self.assertEqual(result.log_file.read(), 'some text')
-        return self.worker_monitor.finishJob(
+            transaction.abort()
+            url = worker_monitor.codeimport_endpoint.calls[0][3]
+            text = urllib.urlopen(url).read()
+            self.assertEqual(log_text, text)
+        return worker_monitor.finishJob(
             CodeImportResultStatus.SUCCESS).addCallback(
             check_file_uploaded)
 
-    def test_finishJobStillCreatesResultWhenLibrarianUploadFails(self):
-        # If the upload to the librarian fails for any reason, the
-        # worker monitor still calls the finishJob workflow method,
-        # but an OOPS is logged to indicate there was a problem.
+    @suppress_stderr
+    def test_finishJob_still_calls_finishJobID_if_upload_fails(self):
+        # If the upload to the librarian fails for any reason, the worker
+        # monitor still calls the finishJobID XML-RPC method, but logs an
+        # error to indicate there was a problem.
+
         # Write some text so that we try to upload the log.
-        self.worker_monitor._log_file.write('some text')
+        job_id = self.factory.getUniqueInteger()
+        worker_monitor = self.makeWorkerMonitorWithJob(job_id)
+        worker_monitor._log_file.write('some text')
+
         # Make _createLibrarianFileAlias fail in a distinctive way.
-        self.worker_monitor._createLibrarianFileAlias = lambda *args: 1/0
-        def check_oops_logged_and_result_created(ignored):
-            self.assertOopsesLogged([ZeroDivisionError])
+        worker_monitor._createLibrarianFileAlias = lambda *args: 1/0
+        def check_finishJob_called(result):
             self.assertEqual(
-                len(list(self.getResultsForOurCodeImport())), 1)
-        return self.worker_monitor.finishJob(
+                [('finishJobID', job_id, 'SUCCESS', '')],
+                worker_monitor.codeimport_endpoint.calls)
+            errors = self.flushLoggedErrors(ZeroDivisionError)
+            self.assertEqual(1, len(errors))
+        return worker_monitor.finishJob(
             CodeImportResultStatus.SUCCESS).addCallback(
-            check_oops_logged_and_result_created)
+            check_finishJob_called)
 
-    def patchOutFinishJob(self):
+    def patchOutFinishJob(self, worker_monitor):
         calls = []
         def finishJob(status):
             calls.append(status)
             return defer.succeed(None)
-        self.worker_monitor.finishJob = finishJob
+        worker_monitor.finishJob = finishJob
         return calls
 
     def test_callFinishJobCallsFinishJobSuccess(self):
         # callFinishJob calls finishJob with CodeImportResultStatus.SUCCESS if
         # its argument is not a Failure.
-        calls = self.patchOutFinishJob()
-        self.worker_monitor.callFinishJob(None)
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        calls = self.patchOutFinishJob(worker_monitor)
+        worker_monitor.callFinishJob(None)
         self.assertEqual(calls, [CodeImportResultStatus.SUCCESS])
 
+    @suppress_stderr
     def test_callFinishJobCallsFinishJobFailure(self):
         # callFinishJob calls finishJob with CodeImportResultStatus.FAILURE
         # and swallows the failure if its argument indicates that the
         # subprocess exited with an exit code of
         # CodeImportWorkerExitCode.FAILURE.
-        calls = self.patchOutFinishJob()
-        ret = self.worker_monitor.callFinishJob(
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        calls = self.patchOutFinishJob(worker_monitor)
+        ret = worker_monitor.callFinishJob(
             makeFailure(
                 error.ProcessTerminated,
                 exitCode=CodeImportWorkerExitCode.FAILURE))
@@ -311,8 +362,9 @@
         # If the argument to callFinishJob indicates that the subprocess
         # exited with a code of CodeImportWorkerExitCode.SUCCESS_NOCHANGE, it
         # calls finishJob with a status of SUCCESS_NOCHANGE.
-        calls = self.patchOutFinishJob()
-        ret = self.worker_monitor.callFinishJob(
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        calls = self.patchOutFinishJob(worker_monitor)
+        ret = worker_monitor.callFinishJob(
             makeFailure(
                 error.ProcessTerminated,
                 exitCode=CodeImportWorkerExitCode.SUCCESS_NOCHANGE))
@@ -322,12 +374,14 @@
         # callFinishJob did not swallow the error, this will fail the test.
         return ret
 
+    @suppress_stderr
     def test_callFinishJobCallsFinishJobArbitraryFailure(self):
         # If the argument to callFinishJob indicates that there was some other
         # failure that had nothing to do with the subprocess, it records
         # failure.
-        calls = self.patchOutFinishJob()
-        ret = self.worker_monitor.callFinishJob(makeFailure(RuntimeError))
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        calls = self.patchOutFinishJob(worker_monitor)
+        ret = worker_monitor.callFinishJob(makeFailure(RuntimeError))
         self.assertEqual(calls, [CodeImportResultStatus.FAILURE])
         self.assertOopsesLogged([RuntimeError])
         # We return the deferred that callFinishJob returns -- if
@@ -338,8 +392,9 @@
         # If the argument to callFinishJob indicates that the subprocess
         # exited with a code of CodeImportWorkerExitCode.SUCCESS_PARTIAL, it
         # calls finishJob with a status of SUCCESS_PARTIAL.
-        calls = self.patchOutFinishJob()
-        ret = self.worker_monitor.callFinishJob(
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        calls = self.patchOutFinishJob(worker_monitor)
+        ret = worker_monitor.callFinishJob(
             makeFailure(
                 error.ProcessTerminated,
                 exitCode=CodeImportWorkerExitCode.SUCCESS_PARTIAL))
@@ -349,23 +404,31 @@
         # callFinishJob did not swallow the error, this will fail the test.
         return ret
 
+    @suppress_stderr
     def test_callFinishJobLogsTracebackOnFailure(self):
         # When callFinishJob is called with a failure, it dumps the traceback
         # of the failure into the log file.
-        ret = self.worker_monitor.callFinishJob(makeFailure(RuntimeError))
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        ret = worker_monitor.callFinishJob(makeFailure(RuntimeError))
         def check_log_file(ignored):
-            self.worker_monitor._log_file.seek(0)
-            log_text = self.worker_monitor._log_file.read()
-            self.assertIn('RuntimeError', log_text)
+            failures = self.flushLoggedErrors(RuntimeError)
+            self.assertEqual(1, len(failures))
+            fail = failures[0]
+            traceback_file = StringIO.StringIO()
+            fail.printTraceback(traceback_file)
+            worker_monitor._log_file.seek(0)
+            log_text = worker_monitor._log_file.read()
+            self.assertIn(traceback_file.read(), log_text)
         return ret.addCallback(check_log_file)
 
     def test_callFinishJobRespects_call_finish_job(self):
         # callFinishJob does not call finishJob if _call_finish_job is False.
         # This is to support exiting without fuss when the job we are working
         # on is deleted in the web UI.
-        calls = self.patchOutFinishJob()
-        self.worker_monitor._call_finish_job = False
-        self.worker_monitor.callFinishJob(None)
+        worker_monitor = self.makeWorkerMonitorWithJob()
+        calls = self.patchOutFinishJob(worker_monitor)
+        worker_monitor._call_finish_job = False
+        worker_monitor.callFinishJob(None)
         self.assertEqual(calls, [])
 
 
@@ -376,14 +439,29 @@
     # This works around a clash between the TrialTestCase and the BzrTestCase.
     skip = None
 
+    layer = TwistedLayer
+
     class WorkerMonitor(CodeImportWorkerMonitor):
         """See `CodeImportWorkerMonitor`.
 
         Override _launchProcess to return a deferred that we can
-        callback/errback as we choose.
+        callback/errback as we choose.  Passing ``has_job=False`` to the
+        constructor will cause getWorkerArguments() to raise ExitQuietly (this
+        bit is tested above).
         """
 
-        def _launchProcess(self, source_details):
+        def __init__(self, process_deferred, has_job=True):
+            if has_job:
+                job_data = {1: ([], '', '')}
+            else:
+                job_data = {}
+            CodeImportWorkerMonitor.__init__(
+                self, 1, QuietFakeLogger(),
+                FakeCodeImportScheduleEndpointProxy(job_data))
+            self.result_status = None
+            self.process_deferred = process_deferred
+
+        def _launchProcess(self, worker_arguments):
             return self.process_deferred
 
         def finishJob(self, status):
@@ -391,60 +469,47 @@
             self.result_status = status
             return defer.succeed(None)
 
-    layer = TwistedLaunchpadZopelessLayer
-
-    def setUp(self):
-        self.factory = LaunchpadObjectFactory()
-        login('[e-mail]')
-        job = self.factory.makeCodeImportJob()
-        self.code_import_id = job.code_import.id
-        getUtility(ICodeImportJobWorkflow).startJob(
-            job, self.factory.makeCodeImportMachine(set_online=True))
-        self.job_id = job.id
-        self.worker_monitor = self.WorkerMonitor(job.id, QuietFakeLogger())
-        self.worker_monitor.result_status = None
-        self.layer.txn.commit()
-        self.layer.switchDbUser('codeimportworker')
-
-    def tearDown(self):
-        logout()
-
-    @read_only_transaction
-    def assertFinishJobCalledWithStatus(self, ignored, status):
-        """Assert that finishJob was called with the given status."""
-        self.assertEqual(self.worker_monitor.result_status, status)
+    def assertFinishJobCalledWithStatus(self, ignored, worker_monitor, status):
+        """Assert that finishJob was called with the given status."""
+        self.assertEqual(worker_monitor.result_status, status)
+
+    def assertFinishJobNotCalled(self, ignored, worker_monitor):
+        """Assert that finishJob was called with the given status."""
+        self.assertFinishJobCalledWithStatus(ignored, worker_monitor, None)
 
     def test_success(self):
         # In the successful case, finishJob is called with
         # CodeImportResultStatus.SUCCESS.
-        self.worker_monitor.process_deferred = defer.succeed(None)
-        return self.worker_monitor.run().addCallback(
-            self.assertFinishJobCalledWithStatus,
+        worker_monitor = self.WorkerMonitor(defer.succeed(None))
+        return worker_monitor.run().addCallback(
+            self.assertFinishJobCalledWithStatus, worker_monitor,
             CodeImportResultStatus.SUCCESS)
 
     def test_failure(self):
         # If the process deferred is fired with a failure, finishJob is called
         # with CodeImportResultStatus.FAILURE, but the call to run() still
         # succeeds.
-        self.worker_monitor.process_deferred = defer.fail(RuntimeError())
-        return self.worker_monitor.run().addCallback(
-            self.assertFinishJobCalledWithStatus,
+        worker_monitor = self.WorkerMonitor(defer.fail(RuntimeError()))
+        return worker_monitor.run().addCallback(
+            self.assertFinishJobCalledWithStatus, worker_monitor,
             CodeImportResultStatus.FAILURE)
 
     def test_quiet_exit(self):
         # If the process deferred fails with ExitQuietly, the call to run()
-        # succeeds.
-        self.worker_monitor.process_deferred = defer.fail(ExitQuietly())
-        return self.worker_monitor.run()
+        # succeeds, and finishJob is not called at all.
+        worker_monitor = self.WorkerMonitor(
+            defer.succeed(None), has_job=False)
+        return worker_monitor.run().addCallback(
+            self.assertFinishJobNotCalled, worker_monitor)
 
     def test_quiet_exit_from_finishJob(self):
         # If finishJob fails with ExitQuietly, the call to run() still
         # succeeds.
-        self.worker_monitor.process_deferred = defer.succeed(None)
+        worker_monitor = self.WorkerMonitor(defer.succeed(None))
         def finishJob(reason):
             raise ExitQuietly
-        self.worker_monitor.finishJob = finishJob
-        return self.worker_monitor.run()
+        worker_monitor.finishJob = finishJob
+        return worker_monitor.run()
 
 
 def nuke_codeimport_sample_data():
@@ -488,7 +553,7 @@
 
 class TestWorkerMonitorIntegration(TrialTestCase, BzrTestCase):
 
-    layer = TwistedLaunchpadZopelessLayer
+    layer = TwistedAppServerLayer
 
     # This works around a clash between the TrialTestCase and the BzrTestCase.
     skip = None
@@ -599,7 +664,6 @@
         self.assertEqual(
             self.foreign_commit_count, len(branch.revision_history()))
 
-    @read_only_transaction
     def assertImported(self, ignored, code_import_id):
         """Assert that the `CodeImport` of the given id was imported."""
         # In the in-memory tests, check that resetTimeout on the
@@ -617,8 +681,9 @@
 
         This implementation does it in-process.
         """
-        self.layer.switchDbUser('codeimportworker')
-        monitor = CIWorkerMonitorForTesting(job_id, QuietFakeLogger())
+        monitor = CIWorkerMonitorForTesting(
+            job_id, QuietFakeLogger(),
+            xmlrpc.Proxy(config.codeimportdispatcher.codeimportscheduler_url))
         deferred = monitor.run()
         def save_protocol_object(result):
             """Save the process protocol object.
@@ -704,7 +769,7 @@
         This implementation does it in a child process.
         """
         script_path = os.path.join(
-            config.root, 'scripts', 'code-import-worker-db.py')
+            config.root, 'scripts', 'code-import-worker-monitor.py')
         process_end_deferred = defer.Deferred()
         # The "childFDs={0:0, 1:1, 2:2}" means that any output from the script
         # goes to the test runner's console rather than to pipes that noone is

=== modified file 'lib/lp/codehosting/codeimport/workermonitor.py'
--- lib/lp/codehosting/codeimport/workermonitor.py	2010-02-17 04:28:48 +0000
+++ lib/lp/codehosting/codeimport/workermonitor.py	2010-03-14 20:28:32 +0000
@@ -14,26 +14,19 @@
 
 from twisted.internet import defer, error, reactor, task
 from twisted.python import failure
-from twisted.python.util import mergeFunctionMetadata
+from twisted.web import xmlrpc
 
 from zope.component import getUtility
 
 from canonical.config import config
-from canonical.database.sqlbase import begin, commit, rollback
-from canonical.launchpad.interfaces import ILibraryFileAliasSet
-from canonical.launchpad.webapp.interaction import Participation
-from canonical.launchpad.webapp import canonical_url
-from canonical.twistedsupport import defer_to_thread
+from canonical.launchpad.xmlrpc.faults import NoSuchCodeImportJob
+from canonical.librarian.interfaces import IFileUploadClient
 from canonical.twistedsupport.loggingsupport import (
     log_oops_from_failure)
 from canonical.twistedsupport.processmonitor import (
     ProcessMonitorProtocolWithTimeout)
 from lp.code.enums import CodeImportResultStatus
-from lp.code.interfaces.codeimportjob import (
-    ICodeImportJobSet, ICodeImportJobWorkflow)
-from lp.codehosting.codeimport.worker import (
-    CodeImportSourceDetails, CodeImportWorkerExitCode)
-from lp.testing import login, logout, ANONYMOUS
+from lp.codehosting.codeimport.worker import CodeImportWorkerExitCode
 
 
 class CodeImportWorkerMonitorProtocol(ProcessMonitorProtocolWithTimeout):
@@ -109,49 +102,6 @@
         self._looping_call.stop()
 
 
-def read_only_transaction(function):
-    """Wrap 'function' in a transaction and Zope session.
-
-    The transaction is always aborted."""
-    def transacted(*args, **kwargs):
-        begin()
-        # XXX gary 20-Oct-2008 bug 285808
-        # We should reconsider using a ftest helper for production code. For
-        # now, we explicitly keep the code from using a test request by using
-        # a basic participation.
-        login(ANONYMOUS, Participation())
-        try:
-            return function(*args, **kwargs)
-        finally:
-            logout()
-            rollback()
-    return mergeFunctionMetadata(function, transacted)
-
-
-def writing_transaction(function):
-    """Wrap 'function' in a transaction and Zope session.
-
-    The transaction is committed if 'function' returns normally and
-    aborted if it raises an exception."""
-    def transacted(*args, **kwargs):
-        begin()
-        # XXX gary 20-Oct-2008 bug 285808
-        # We should reconsider using a ftest helper for production code. For
-        # now, we explicitly keep the code from using a test request by using
-        # a basic participation.
-        login(ANONYMOUS, Participation())
-        try:
-            ret = function(*args, **kwargs)
-        except:
-            logout()
-            rollback()
-            raise
-        logout()
-        commit()
-        return ret
-    return mergeFunctionMetadata(function, transacted)
-
-
 class ExitQuietly(Exception):
     """Raised to indicate that we should abort and exit without fuss.
 
@@ -171,113 +121,105 @@
     path_to_script = os.path.join(
         config.root, 'scripts', 'code-import-worker.py')
 
-    def __init__(self, job_id, logger):
+    def __init__(self, job_id, logger, codeimport_endpoint):
         """Construct an instance.
 
         :param job_id: The ID of the CodeImportJob we are to work on.
         :param logger: A `Logger` object.
         """
+        self._job_id = job_id
         self._logger = logger
-        self._job_id = job_id
+        self.codeimport_endpoint = codeimport_endpoint
         self._call_finish_job = True
         self._log_file = tempfile.TemporaryFile()
-        self._source_details = None
-        self._code_import_id = None
         self._branch_url = None
+        self._log_file_name = 'no-name-set.txt'
 
     def _logOopsFromFailure(self, failure):
         request = log_oops_from_failure(
-            failure, code_import_job_id=self._job_id,
-            code_import_id=self._code_import_id, URL=self._branch_url)
+            failure, code_import_job_id=self._job_id, URL=self._branch_url)
         self._logger.info(
             "Logged OOPS id %s: %s: %s",
             request.oopsid, failure.type.__name__, failure.value)
 
-    def getJob(self):
-        """Fetch the `CodeImportJob` object we are working on from the DB.
-
-        Only call this from defer_to_thread-ed methods!
-
-        :raises ExitQuietly: if the job is not found.
-        """
-        job = getUtility(ICodeImportJobSet).getById(self._job_id)
-        if job is None:
-            self._logger.info(
-                "Job %d not found, exiting quietly.", self._job_id)
+    def _trap_nosuchcodeimportjob(self, failure):
+        failure.trap(xmlrpc.Fault)
+        if failure.value.faultCode == NoSuchCodeImportJob.error_code:
             self._call_finish_job = False
             raise ExitQuietly
         else:
-            return job
-
-    @defer_to_thread
-    @read_only_transaction
-    def getSourceDetails(self):
-        """Get a `CodeImportSourceDetails` for the job we are working on."""
-        code_import = self.getJob().code_import
-        source_details = CodeImportSourceDetails.fromCodeImport(code_import)
-        self._logger.info(
-            'Found source details: %s', source_details.asArguments())
-        self._branch_url = canonical_url(code_import.branch)
-        self._code_import_id = code_import.id
-        return source_details
-
-    @defer_to_thread
-    @writing_transaction
+            raise failure.value
+
+    def getWorkerArguments(self):
+        """Get arguments for the worker for the import we are working on.
+
+        This also sets the _branch_url and _log_file_name attributes for use
+        in the _logOopsFromFailure and finishJob methods respectively.
+        """
+        deferred = self.codeimport_endpoint.callRemote(
+            'getImportDataForJobID', self._job_id)
+        def _processResult(result):
+            code_import_arguments, branch_url, log_file_name = result
+            self._branch_url = branch_url
+            self._log_file_name = log_file_name
+            self._logger.info(
+                'Found source details: %s', code_import_arguments)
+            return code_import_arguments
+        return deferred.addCallbacks(_processResult, self._trap_nosuchcodeimportjob)
+
     def updateHeartbeat(self, tail):
         """Call the updateHeartbeat method for the job we are working on."""
         self._logger.debug("Updating heartbeat.")
-        getUtility(ICodeImportJobWorkflow).updateHeartbeat(
-            self.getJob(), tail)
+        deferred = self.codeimport_endpoint.callRemote(
+            'updateHeartbeat', self._job_id, tail)
+        return deferred.addErrback(self._trap_nosuchcodeimportjob)
 
     def _createLibrarianFileAlias(self, name, size, file, contentType):
-        """Call `ILibraryFileAliasSet.create` with the given parameters.
+        """Call `IFileUploadClient.remoteAddFile` with the given parameters.
 
-        This is a separate method that exists only to be patched in
-        tests.
+        This is a separate method that exists only to be patched in tests.
         """
-        return getUtility(ILibraryFileAliasSet).create(
+        # This blocks, but never mind: nothing else is going on in the process
+        # by this point.  We could dispatch to a thread if we felt like it, or
+        # even come up with an asynchronous implementation of the librarian
+        # protocol (it's not very complicated).
+        return getUtility(IFileUploadClient).remoteAddFile(
             name, size, file, contentType)
 
-    @defer_to_thread
-    @writing_transaction
     def finishJob(self, status):
-        """Call the finishJob method for the job we are working on.
+        """Call the finishJobID method for the job we are working on.
 
         This method uploads the log file to the librarian first.
         """
-        job = self.getJob()
         log_file_size = self._log_file.tell()
         if log_file_size > 0:
             self._log_file.seek(0)
-            branch = job.code_import.branch
-            log_file_name = '%s-%s-%s-log.txt' % (
-                branch.owner.name, branch.product.name, branch.name)
             try:
-                log_file_alias = self._createLibrarianFileAlias(
-                    log_file_name, log_file_size, self._log_file,
+                log_file_alias_url = self._createLibrarianFileAlias(
+                    self._log_file_name, log_file_size, self._log_file,
                     'text/plain')
                 self._logger.info(
-                    "Uploaded logs to librarian %s.", log_file_alias.getURL())
+                    "Uploaded logs to librarian %s.", log_file_alias_url)
             except:
                 self._logger.error("Upload to librarian failed.")
                 self._logOopsFromFailure(failure.Failure())
-                log_file_alias = None
+                log_file_alias_url = ''
         else:
-            log_file_alias = None
-        getUtility(ICodeImportJobWorkflow).finishJob(
-            job, status, log_file_alias)
+            log_file_alias_url = ''
+        return self.codeimport_endpoint.callRemote(
+            'finishJobID', self._job_id, status.name, log_file_alias_url
+            ).addErrback(self._trap_nosuchcodeimportjob)
 
     def _makeProcessProtocol(self, deferred):
         """Make an `CodeImportWorkerMonitorProtocol` for a subprocess."""
         return CodeImportWorkerMonitorProtocol(deferred, self, self._log_file)
 
-    def _launchProcess(self, source_details):
+    def _launchProcess(self, worker_arguments):
         """Launch the code-import-worker.py child process."""
         deferred = defer.Deferred()
         protocol = self._makeProcessProtocol(deferred)
         interpreter = '%s/bin/py' % config.root
-        command = [interpreter, self.path_to_script]
-        command.extend(source_details.asArguments())
+        command = [interpreter, self.path_to_script] + worker_arguments
         self._logger.info(
             "Launching worker child process %s.", command)
         reactor.spawnProcess(
@@ -286,7 +228,7 @@
 
     def run(self):
         """Perform the import."""
-        return self.getSourceDetails().addCallback(
+        return self.getWorkerArguments().addCallback(
             self._launchProcess).addBoth(
             self.callFinishJob).addErrback(
             self._silenceQuietExit)
@@ -297,6 +239,11 @@
         return None
 
     def _reasonToStatus(self, reason):
+        """Translate the 'reason' for process exit into a result status.
+
+        Different exit codes are presumed by Twisted to be errors, but are
+        different kinds of success for us.
+        """
         if isinstance(reason, failure.Failure):
             if reason.check(error.ProcessTerminated):
                 if reason.value.exitCode == \

=== modified file 'lib/lp/codehosting/inmemory.py'
--- lib/lp/codehosting/inmemory.py	2010-02-24 04:24:01 +0000
+++ lib/lp/codehosting/inmemory.py	2010-03-14 20:28:32 +0000
@@ -62,12 +62,22 @@
         expected_value = kwargs[attribute]
         branch = self._object_set.get(branch_id)
         if branch is None:
-            return None
+            return FakeResult(None)
         if expected_value is getattr(branch, attribute):
-            return branch
+            return FakeResult(branch)
         return None
 
 
+class FakeResult:
+    """As with FakeStore, just enough of a result to pass tests."""
+
+    def __init__(self, branch):
+        self._branch = branch
+
+    def one(self):
+        return self._branch
+
+
 class FakeDatabaseObject:
     """Base class for fake database objects."""
 

=== modified file 'lib/lp/testing/__init__.py'
--- lib/lp/testing/__init__.py	2010-02-17 20:54:36 +0000
+++ lib/lp/testing/__init__.py	2010-03-14 20:28:32 +0000
@@ -294,7 +294,7 @@
         sql_class = type(sql_object)
         store = Store.of(sql_object)
         found_object = store.find(
-            sql_class, **({'id': sql_object.id, attribute_name: date}))
+            sql_class, **({'id': sql_object.id, attribute_name: date})).one()
         if found_object is None:
             self.fail(
                 "Expected %s to be %s, but it was %s."

=== renamed file 'scripts/code-import-worker-db.py' => 'scripts/code-import-worker-monitor.py'
--- scripts/code-import-worker-db.py	2009-10-13 14:38:07 +0000
+++ scripts/code-import-worker-monitor.py	2010-03-14 20:28:32 +0000
@@ -22,11 +22,14 @@
 
 from twisted.internet import defer, reactor
 from twisted.python import log
+from twisted.web import xmlrpc
+
+from canonical.config import config
+from canonical.twistedsupport.loggingsupport import set_up_oops_reporting
 
 from lp.codehosting.codeimport.workermonitor import (
     CodeImportWorkerMonitor)
 from lp.services.scripts.base import LaunchpadScript
-from canonical.twistedsupport.loggingsupport import set_up_oops_reporting
 
 
 class CodeImportWorker(LaunchpadScript):
@@ -35,22 +38,30 @@
         LaunchpadScript.__init__(self, name, dbuser, test_args)
         set_up_oops_reporting(name, mangle_stdout=True)
 
+    def _init_db(self, implicit_begin, isolation):
+        # This script doesn't access the database.
+        pass
+
     def main(self):
+        arg, = self.args
+        job_id = int(arg)
         # XXX: MichaelHudson 2008-05-07 bug=227586: Setting up the component
         # architecture overrides $GNUPGHOME to something stupid.
         os.environ['GNUPGHOME'] = ''
-        reactor.callWhenRunning(self._run_reactor)
+        reactor.callWhenRunning(self._do_import, job_id)
         reactor.run()
 
-    def _run_reactor(self):
-        defer.maybeDeferred(self._main).addErrback(
+    def _do_import(self, job_id):
+        defer.maybeDeferred(self._main, job_id).addErrback(
             log.err).addCallback(
             lambda ignored: reactor.stop())
 
-    def _main(self):
-        arg, = self.args
-        return CodeImportWorkerMonitor(int(arg), self.logger).run()
+    def _main(self, job_id):
+        worker = CodeImportWorkerMonitor(
+            job_id, self.logger,
+            xmlrpc.Proxy(config.codeimportdispatcher.codeimportscheduler_url))
+        return worker.run()
 
 if __name__ == '__main__':
-    script = CodeImportWorker('codeimportworker', dbuser='codeimportworker')
+    script = CodeImportWorker('codeimportworker')
     script.run()

=== modified file 'scripts/code-import-worker.py'
--- scripts/code-import-worker.py	2010-02-04 01:12:05 +0000
+++ scripts/code-import-worker.py	2010-03-14 20:28:32 +0000
@@ -8,7 +8,7 @@
 By 'processing a code import' we mean importing or updating code from a
 remote, non-Bazaar, repository.
 
-This script is usually run by the code-import-worker-db.py script that
+This script is usually run by the code-import-worker-monitor.py script that
 communicates progress and results to the database.
 """