Merge lp:~stevenk/launchpad/dsdj-runner into lp:launchpad

[Steve Kowalik]

[r=henninge][bug=739997] Implement the job runner for DSDJs.

Add "the business end" to DistroSeriesDifferenceJob is the short description.

The longer description is implement a run() method for DSDJ, add a *lot* of tests for it, create a database user, add relevant permissions for it, implement a cronscript that will run said jobs, also create DSDJs when an SPPH is marked for deletion, fix some bugs I noticed during implementation and perform some clean up.

(Some of the clean up wasn't strictly necessary, but I was the first implementer of DistributionJob, so I still think things should be 'just so'.)

[Henning Eggers (henninge)]

-----BEGIN PGP SIGNED MESSAGE-----
Hash: SHA1

Hi Steven,
thank you for this branch. It looks all very well done and thoroughly tested.
That's good. ;-)

I have two issues:
- - You keep using IMasterStore when I think IStore would be totally sufficient.
Any particular reason for that?
- - There is a helper method called canonical.launchpad.scripts.tests.run_script
that encapsulates a lot of what you are doing manually.

 review approve code

Cheers,
Henning

Am 21.03.2011 08:49, schrieb Steve Kowalik:
> === modified file 'lib/lp/soyuz/tests/test_distroseriesdifferencejob.py'
> --- lib/lp/soyuz/tests/test_distroseriesdifferencejob.py 2011-03-11 13:35:48 +0000
> +++ lib/lp/soyuz/tests/test_distroseriesdifferencejob.py 2011-03-21 07:49:35 +0000
> @@ -5,28 +5,43 @@
>
> __metaclass__ = type
>
> +import os
> +import subprocess
> +import sys

New line here between "system" and "3rd party" imports. But maybe that section
is going away anyway ...

> import transaction
> from psycopg2 import ProgrammingError
> from zope.component import getUtility
> from zope.interface.verify import verifyObject
>
> +from canonical.config import config
> +from canonical.launchpad.interfaces.lpstorm import IMasterStore
> from canonical.testing.layers import (
> LaunchpadZopelessLayer,
> ZopelessDatabaseLayer,
> )
> +from lp.registry.enum import (
> + DistroSeriesDifferenceStatus,
> + DistroSeriesDifferenceType,
> + )
> +from lp.registry.model.distroseriesdifference import DistroSeriesDifference
> from lp.services.features.testing import FeatureFixture
> from lp.services.job.interfaces.job import JobStatus
> -from lp.soyuz.interfaces.distroseriesdifferencejob import (
> +from lp.soyuz.enums import PackagePublishingStatus
> +from lp.soyuz.interfaces.distributionjob import (
> IDistroSeriesDifferenceJobSource,
> )
> from lp.soyuz.model.distroseriesdifferencejob import (
> create_job,
> + DistroSeriesDifferenceJob,
> FEATURE_FLAG_ENABLE_MODULE,
> find_waiting_jobs,
> make_metadata,
> may_require_job,
> )
> -from lp.testing import TestCaseWithFactory
> +from lp.testing import (
> + person_logged_in,
> + TestCaseWithFactory,
> + )
>
>
> class TestDistroSeriesDifferenceJobSource(TestCaseWithFactory):
[...]
> @@ -163,6 +187,246 @@
> self.getJobSource().createForPackagePublication(distroseries, package)
> self.assertContentEqual([], find_waiting_jobs(distroseries, package))
>
> + def test_cronscript(self):
> + derived_series = self.makeDerivedDistroSeries()
> + package = self.factory.makeSourcePackageName()
> + self.getJobSource().createForPackagePublication(
> + derived_series, package)
> + transaction.commit() # The cronscript is a different process.
> + script = os.path.join(
> + config.root, 'cronscripts', 'distroseriesdifference_job.py')
> + args = [sys.executable, script, '-v']
> + process = subprocess.Popen(
> + args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
> + stdout, stderr = process.communicate()

So, this whole section could be:
       return_code, stdout...

review: Approve (code)

=== modified file 'configs/development/launchpad-lazr.conf'
--- configs/development/launchpad-lazr.conf	2011-02-25 19:27:05 +0000
+++ configs/development/launchpad-lazr.conf	2011-03-23 01:14:43 +0000
@@ -115,6 +115,10 @@
 timeout: 10
 cdimage_file_list_url: file:lib/canonical/launchpad/doc/ubuntu-releases.testdata
 
+[distroseriesdifferencejob]
+oops_prefix: DSDJ
+error_dir: /var/tmp/soyuz.test
+
 [error_reports]
 oops_prefix: X
 error_dir: /var/tmp/lperr

=== added file 'cronscripts/distroseriesdifference_job.py'
--- cronscripts/distroseriesdifference_job.py	1970-01-01 00:00:00 +0000
+++ cronscripts/distroseriesdifference_job.py	2011-03-23 01:14:43 +0000
@@ -0,0 +1,37 @@
+#!/usr/bin/python -S
+#
+# Copyright 2010 Canonical Ltd.  This software is licensed under the
+# GNU Affero General Public License version 3 (see the file LICENSE).
+
+"""Process DistroSeriesDifferences."""
+
+__metaclass__ = type
+
+import _pythonpath
+
+from lp.services.features import getFeatureFlag
+from lp.services.job.runner import JobCronScript
+from lp.soyuz.model.distroseriesdifferencejob import (
+    FEATURE_FLAG_ENABLE_MODULE,
+    )
+from lp.soyuz.interfaces.distributionjob import (
+    IDistroSeriesDifferenceJobSource,
+    )
+
+
+class RunDistroSeriesDifferenceJob(JobCronScript):
+    """Run DistroSeriesDifferenceJob jobs."""
+
+    config_name = 'distroseriesdifferencejob'
+    source_interface = IDistroSeriesDifferenceJobSource
+
+    def main(self):
+        if not getFeatureFlag(FEATURE_FLAG_ENABLE_MODULE):
+            self.logger.info("Feature flag is not enabled.")
+            return
+        super(RunDistroSeriesDifferenceJob, self).main()
+
+
+if __name__ == '__main__':
+    script = RunDistroSeriesDifferenceJob()
+    script.lock_and_run()

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2011-03-18 03:56:36 +0000
+++ database/schema/security.cfg	2011-03-23 01:14:43 +0000
@@ -1059,6 +1059,21 @@
 public.sourcepackagerelease                     = SELECT
 public.sourcepackagereleasefile                 = SELECT, INSERT, UPDATE
 
+[distroseriesdifferencejob]
+type=user
+groups=script
+public.archive                                  = SELECT
+public.distribution                             = SELECT
+public.distributionjob                          = SELECT
+public.distroseries                             = SELECT
+public.distroseriesdifference                   = SELECT, INSERT, UPDATE
+public.job                                      = SELECT, UPDATE
+public.libraryfilealias                         = SELECT
+public.libraryfilecontent                       = SELECT
+public.sourcepackagename                        = SELECT
+public.sourcepackagepublishinghistory           = SELECT
+public.sourcepackagerelease                     = SELECT
+
 [write]
 type=group
 # Full access except for tables that are exclusively updated by

=== modified file 'lib/canonical/config/schema-lazr.conf'
--- lib/canonical/config/schema-lazr.conf	2011-03-07 20:49:03 +0000
+++ lib/canonical/config/schema-lazr.conf	2011-03-23 01:14:43 +0000
@@ -728,6 +728,19 @@
 cdimage_file_list_url: http://releases.ubuntu.com/.manifest
 
 
+[distroseriesdifferencejob]
+dbuser: distroseriesdifferencejob
+
+# See [error_reports].
+error_dir: none
+
+# See [error_reports].
+oops_prefix: none
+
+# See [error_reports].
+copy_to_zlog: false
+
+
 [error_reports]
 # A prefix for "OOPS" codes for this process instance.
 # This is used to allow storing the reports from different

=== modified file 'lib/canonical/launchpad/icing/style-3-0.css'
=== modified file 'lib/lp/code/model/branchcollection.py'
--- lib/lp/code/model/branchcollection.py	2011-03-22 02:14:15 +0000
+++ lib/lp/code/model/branchcollection.py	2011-03-23 01:14:43 +0000
@@ -203,15 +203,31 @@
                 cache = caches[link.branchID]
                 cache._associatedSuiteSourcePackages.append(
                     link.suite_sourcepackage)
-            load_related(Product, rows, ['productID'])
-            # So far have only needed the persons for their canonical_url - no
-            # need for validity etc in the /branches API call.
-            load_related(Person, rows,
-                ['ownerID', 'registrantID', 'reviewerID'])
-            for code_import in IStore(CodeImport).find(
-                CodeImport, CodeImport.branchID.is_in(branch_ids)):
-                cache = caches[code_import.branchID]
-                cache.code_import = code_import
+<<<<<<< TREE
+            load_related(Product, rows, ['productID'])
+            # So far have only needed the persons for their canonical_url - no
+            # need for validity etc in the /branches API call.
+            load_related(Person, rows,
+                ['ownerID', 'registrantID', 'reviewerID'])
+            for code_import in IStore(CodeImport).find(
+                CodeImport, CodeImport.branchID.is_in(branch_ids)):
+                cache = caches[code_import.branchID]
+                cache.code_import = code_import
+=======
+            load_related(Product, rows, ['productID'])
+            # So far have only needed the persons for their canonical_url - no
+            # need for validity etc in the /branches API call.
+            load_related(Person, rows,
+                ['ownerID', 'registrantID', 'reviewerID'])
+            # Cache all branches as having no code imports to prevent fruitless
+            # lookups on the ones we don't find.
+            for cache in caches.values():
+                cache.code_import = None
+            for code_import in IStore(CodeImport).find(
+                CodeImport, CodeImport.branchID.is_in(branch_ids)):
+                cache = caches[code_import.branchID]
+                cache.code_import = code_import
+>>>>>>> MERGE-SOURCE
         return DecoratedResultSet(resultset, pre_iter_hook=do_eager_load)
 
     def getMergeProposals(self, statuses=None, for_branches=None,

=== modified file 'lib/lp/soyuz/configure.zcml'
--- lib/lp/soyuz/configure.zcml	2011-03-10 14:05:51 +0000
+++ lib/lp/soyuz/configure.zcml	2011-03-23 01:14:43 +0000
@@ -901,14 +901,15 @@
     </class>
 
     <!-- DistroSeriesDifferenceJobSource -->
+    <securedutility
+      component="lp.soyuz.model.distroseriesdifferencejob.DistroSeriesDifferenceJob"
+      provides="lp.soyuz.interfaces.distributionjob.IDistroSeriesDifferenceJobSource">
+        <allow interface="lp.soyuz.interfaces.distributionjob.IDistroSeriesDifferenceJobSource"/>
+    </securedutility>
     <class class="lp.soyuz.model.distroseriesdifferencejob.DistroSeriesDifferenceJob">
+        <allow interface="lp.soyuz.interfaces.distributionjob.IDistroSeriesDifferenceJob" />
         <allow interface="lp.soyuz.interfaces.distributionjob.IDistributionJob" />
     </class>
-    <securedutility
-      component="lp.soyuz.model.distroseriesdifferencejob.DistroSeriesDifferenceJob"
-      provides="lp.soyuz.interfaces.distroseriesdifferencejob.IDistroSeriesDifferenceJobSource">
-        <allow interface="lp.soyuz.interfaces.distroseriesdifferencejob.IDistroSeriesDifferenceJobSource"/>
-    </securedutility>
 
     <!-- SyncPackageJobSource -->
     <securedutility

=== modified file 'lib/lp/soyuz/interfaces/distributionjob.py'
--- lib/lp/soyuz/interfaces/distributionjob.py	2011-03-10 14:05:51 +0000
+++ lib/lp/soyuz/interfaces/distributionjob.py	2011-03-23 01:14:43 +0000
@@ -6,6 +6,8 @@
 __all__ = [
     "DistributionJobType",
     "IDistributionJob",
+    "IDistroSeriesDifferenceJob",
+    "IDistroSeriesDifferenceJobSource",
     "IInitialiseDistroSeriesJob",
     "IInitialiseDistroSeriesJobSource",
     "ISyncPackageJob",
@@ -133,3 +135,20 @@
     include_binaries = Bool(
             title=_("Copy binaries"),
             required=False, readonly=True)
+
+
+class IDistroSeriesDifferenceJob(IRunnableJob):
+        """A Job that performs actions related to DSDs."""
+
+
+class IDistroSeriesDifferenceJobSource(IJobSource):
+    """An `IJob` for creating `DistroSeriesDifference`s."""
+
+    def createForPackagePublication(distroseries, sourcepackagename):
+        """Create jobs as appropriate for a given status publication.
+
+        :param distroseries: A `DistroSeries` that is assumed to be
+            derived from another one.
+        :param sourcepackagename: A `SourcePackageName` that is being
+            published in `distroseries`.
+        """

=== removed file 'lib/lp/soyuz/interfaces/distroseriesdifferencejob.py'
--- lib/lp/soyuz/interfaces/distroseriesdifferencejob.py	2011-03-10 14:05:51 +0000
+++ lib/lp/soyuz/interfaces/distroseriesdifferencejob.py	1970-01-01 00:00:00 +0000
@@ -1,24 +0,0 @@
-# Copyright 2011 Canonical Ltd.  This software is licensed under the
-# GNU Affero General Public License version 3 (see the file LICENSE).
-
-"""`IDistroSeriesDifferenceJob`."""
-
-__metaclass__ = type
-__all__ = [
-    'IDistroSeriesDifferenceJobSource',
-    ]
-
-from lp.services.job.interfaces.job import IJobSource
-
-
-class IDistroSeriesDifferenceJobSource(IJobSource):
-    """An `IJob` for creating `DistroSeriesDifference`s."""
-
-    def createForPackagePublication(distroseries, sourcepackagename):
-        """Create jobs as appropriate for a given status publication.
-
-        :param distroseries: A `DistroSeries` that is assumed to be
-            derived from another one.
-        :param sourcepackagename: A `SourcePackageName` that is being
-            published in `distroseries`.
-        """

=== modified file 'lib/lp/soyuz/model/distroseriesdifferencejob.py'
--- lib/lp/soyuz/model/distroseriesdifferencejob.py	2011-03-11 13:35:48 +0000
+++ lib/lp/soyuz/model/distroseriesdifferencejob.py	2011-03-23 01:14:43 +0000
@@ -8,25 +8,29 @@
     'DistroSeriesDifferenceJob',
     ]
 
+from zope.component import getUtility
 from zope.interface import (
     classProvides,
     implements,
     )
 
 from canonical.launchpad.interfaces.lpstorm import IMasterStore
+from lp.registry.interfaces.distroseriesdifference import (
+    IDistroSeriesDifferenceSource,
+    )
+from lp.registry.model.distroseriesdifference import DistroSeriesDifference
+from lp.registry.model.sourcepackagename import SourcePackageName
 from lp.services.features import getFeatureFlag
 from lp.services.job.model.job import Job
 from lp.soyuz.interfaces.distributionjob import (
     DistributionJobType,
-    IDistributionJob,
+    IDistroSeriesDifferenceJob,
+    IDistroSeriesDifferenceJobSource,
     )
 from lp.soyuz.model.distributionjob import (
     DistributionJob,
     DistributionJobDerived,
     )
-from lp.soyuz.interfaces.distroseriesdifferencejob import (
-    IDistroSeriesDifferenceJobSource,
-    )
 
 
 FEATURE_FLAG_ENABLE_MODULE = u"soyuz.derived_series_jobs.enabled"
@@ -98,7 +102,7 @@
 class DistroSeriesDifferenceJob(DistributionJobDerived):
     """A `Job` type for creating/updating `DistroSeriesDifference`s."""
 
-    implements(IDistributionJob)
+    implements(IDistroSeriesDifferenceJob)
     classProvides(IDistroSeriesDifferenceJobSource)
 
     class_job_type = DistributionJobType.DISTROSERIESDIFFERENCE
@@ -108,12 +112,27 @@
         """See `IDistroSeriesDifferenceJobSource`."""
         if not getFeatureFlag(FEATURE_FLAG_ENABLE_MODULE):
             return
-        children = distroseries.getDerivedSeries()
-        parent = distroseries.parent_series
-        for relative in list(children) + [parent]:
+        jobs = []
+        children = list(distroseries.getDerivedSeries())
+        for relative in children + [distroseries]:
             if may_require_job(relative, sourcepackagename):
-                create_job(relative, sourcepackagename)
+                jobs.append(create_job(relative, sourcepackagename))
+        return jobs
+
+    @property
+    def sourcepackagename(self):
+        return SourcePackageName.get(self.metadata['sourcepackagename'])
 
     def run(self):
         """See `IRunnableJob`."""
-# TODO: Implement the business end.
+        store = IMasterStore(DistroSeriesDifference)
+        ds_diff = store.find(
+            DistroSeriesDifference, 
+            DistroSeriesDifference.derived_series == self.distroseries,
+            DistroSeriesDifference.source_package_name == 
+            self.sourcepackagename).one()
+        if ds_diff is None:
+            ds_diff = getUtility(IDistroSeriesDifferenceSource).new(
+                self.distroseries, self.sourcepackagename)
+        else:
+            ds_diff.update()

=== modified file 'lib/lp/soyuz/model/publishing.py'
--- lib/lp/soyuz/model/publishing.py	2011-03-10 14:05:51 +0000
+++ lib/lp/soyuz/model/publishing.py	2011-03-23 01:14:43 +0000
@@ -85,7 +85,7 @@
     BuildSetStatus,
     IBinaryPackageBuildSet,
     )
-from lp.soyuz.interfaces.distroseriesdifferencejob import (
+from lp.soyuz.interfaces.distributionjob import (
     IDistroSeriesDifferenceJobSource,
     )
 from lp.soyuz.interfaces.publishing import (
@@ -336,6 +336,11 @@
         self.datesuperseded = UTC_NOW
         self.removed_by = removed_by
         self.removal_comment = removal_comment
+        if ISourcePackagePublishingHistory.providedBy(self):
+            dsd_job_source = getUtility(IDistroSeriesDifferenceJobSource)
+            dsd_job_source.createForPackagePublication(
+                self.distroseries,
+                self.sourcepackagerelease.sourcepackagename)
 
     def requestObsolescence(self):
         """See `IArchivePublisher`."""

=== modified file 'lib/lp/soyuz/tests/test_distroseriesdifferencejob.py'
--- lib/lp/soyuz/tests/test_distroseriesdifferencejob.py	2011-03-11 13:35:48 +0000
+++ lib/lp/soyuz/tests/test_distroseriesdifferencejob.py	2011-03-23 01:14:43 +0000
@@ -10,17 +10,26 @@
 from zope.component import getUtility
 from zope.interface.verify import verifyObject
 
+from canonical.launchpad.interfaces.lpstorm import IMasterStore
+from canonical.launchpad.scripts.tests import run_script
 from canonical.testing.layers import (
     LaunchpadZopelessLayer,
     ZopelessDatabaseLayer,
     )
+from lp.registry.enum import (
+    DistroSeriesDifferenceStatus,
+    DistroSeriesDifferenceType,
+    )
+from lp.registry.model.distroseriesdifference import DistroSeriesDifference
 from lp.services.features.testing import FeatureFixture
 from lp.services.job.interfaces.job import JobStatus
-from lp.soyuz.interfaces.distroseriesdifferencejob import (
+from lp.soyuz.enums import PackagePublishingStatus
+from lp.soyuz.interfaces.distributionjob import (
     IDistroSeriesDifferenceJobSource,
     )
 from lp.soyuz.model.distroseriesdifferencejob import (
     create_job,
+    DistroSeriesDifferenceJob,
     FEATURE_FLAG_ENABLE_MODULE,
     find_waiting_jobs,
     make_metadata,
@@ -110,7 +119,7 @@
         sourcepackage = self.factory.makeSourcePackage()
         distroseries, sourcepackagename = (
             sourcepackage.distroseries, sourcepackage.distroseries)
-        job = create_job(distroseries, sourcepackagename)
+        create_job(distroseries, sourcepackagename)
         other_series = self.factory.makeDistroSeries()
         self.assertContentEqual(
             [], find_waiting_jobs(other_series, sourcepackagename))
@@ -119,7 +128,7 @@
         sourcepackage = self.factory.makeSourcePackage()
         distroseries, sourcepackagename = (
             sourcepackage.distroseries, sourcepackage.distroseries)
-        job = create_job(distroseries, sourcepackagename)
+        create_job(distroseries, sourcepackagename)
         other_spn = self.factory.makeSourcePackageName()
         self.assertContentEqual(
             [], find_waiting_jobs(distroseries, other_spn))
@@ -136,22 +145,31 @@
         self.assertContentEqual(
             [], find_waiting_jobs(distroseries, sourcepackagename))
 
-    def test_createForPackagedPublication_creates_job_for_parent_series(self):
+    def test_createForPackagedPublication_creates_jobs_for_its_child(self):
         derived_series = self.factory.makeDistroSeries(
             parent_series=self.makeDerivedDistroSeries())
         package = self.factory.makeSourcePackageName()
+        # Create a job for the derived_series parent, which should create
+        # two jobs. One for derived_series, and the other for its child.
         self.getJobSource().createForPackagePublication(
-            derived_series, package)
-        jobs = list(find_waiting_jobs(derived_series.parent_series, package))
-        self.assertEqual(1, len(jobs))
+            derived_series.parent_series, package)
+        jobs = (list(
+            find_waiting_jobs(derived_series.parent_series, package)) +
+            list(find_waiting_jobs(derived_series, package)))
+        self.assertEqual(2, len(jobs))
         self.assertEqual(package.id, jobs[0].metadata['sourcepackagename'])
+        self.assertEqual(package.id, jobs[1].metadata['sourcepackagename'])
+        # Lastly, a job was not created for the grandparent.
+        jobs = list(
+            find_waiting_jobs(derived_series.parent_series.parent_series,
+                package))
+        self.assertEqual(0, len(jobs))
 
     def test_createForPackagePublication_creates_job_for_derived_series(self):
         derived_series = self.makeDerivedDistroSeries()
-        parent_series = derived_series.parent_series
         package = self.factory.makeSourcePackageName()
         self.getJobSource().createForPackagePublication(
-            parent_series, package)
+            derived_series, package)
         jobs = list(find_waiting_jobs(derived_series, package))
         self.assertEqual(1, len(jobs))
         self.assertEqual(package.id, jobs[0].metadata['sourcepackagename'])
@@ -163,6 +181,242 @@
         self.getJobSource().createForPackagePublication(distroseries, package)
         self.assertContentEqual([], find_waiting_jobs(distroseries, package))
 
+    def test_cronscript(self):
+        derived_series = self.makeDerivedDistroSeries()
+        package = self.factory.makeSourcePackageName()
+        self.getJobSource().createForPackagePublication(
+            derived_series, package)
+        transaction.commit() # The cronscript is a different process.
+        return_code, stdout, stderr = run_script(
+            'cronscripts/distroseriesdifference_job.py', ['-v'])
+        # The cronscript ran how we expected it to.
+        self.assertEqual(return_code, 0)
+        self.assertIn(
+            'INFO    Ran 1 DistroSeriesDifferenceJob jobs.', stderr)
+        # And it did what we expected.
+        jobs = list(find_waiting_jobs(derived_series, package))
+        self.assertEqual(0, len(jobs))
+        store = IMasterStore(DistroSeriesDifference)
+        ds_diff = store.find(
+            DistroSeriesDifference, 
+            DistroSeriesDifference.derived_series == derived_series,
+            DistroSeriesDifference.source_package_name == package)
+        self.assertEqual(1, ds_diff.count())
+
+    def test_job_runner_does_not_create_multiple_dsds(self):
+        derived_series = self.makeDerivedDistroSeries()
+        package = self.factory.makeSourcePackageName()
+        job = self.getJobSource().createForPackagePublication(
+            derived_series, package)
+        job[0].start()
+        job[0].run()
+        job[0].job.complete() # So we can create another job.
+        # The first job would have created a DSD for us.
+        store = IMasterStore(DistroSeriesDifference)
+        ds_diff = store.find(
+            DistroSeriesDifference, 
+            DistroSeriesDifference.derived_series == derived_series,
+            DistroSeriesDifference.source_package_name == package)
+        self.assertEqual(1, ds_diff.count())
+        # If we run the job again, it will not create another DSD.
+        job = self.getJobSource().createForPackagePublication(
+            derived_series, package)
+        job[0].start()
+        job[0].run()
+        ds_diff = store.find(
+            DistroSeriesDifference, 
+            DistroSeriesDifference.derived_series == derived_series,
+            DistroSeriesDifference.source_package_name == package)
+        self.assertEqual(1, ds_diff.count())
+
+class TestDistroSeriesDifferenceJobEndToEnd(TestCaseWithFactory):
+
+    layer = LaunchpadZopelessLayer
+
+    def setUp(self):
+        super(TestDistroSeriesDifferenceJobEndToEnd, self).setUp()
+        self.useFixture(FeatureFixture({FEATURE_FLAG_ENABLE_MODULE: u'on'}))
+        self.store = IMasterStore(DistroSeriesDifference)
+
+    def getJobSource(self):
+        return getUtility(IDistroSeriesDifferenceJobSource)
+
+    def makeDerivedDistroSeries(self):
+        return self.factory.makeDistroSeries(
+            parent_series=self.factory.makeDistroSeries())
+
+    def createPublication(self, source_package_name, versions, distroseries):
+        changelog_lfa = self.factory.makeChangelog(
+            source_package_name.name, versions)
+        transaction.commit() # Yay, librarian.
+        spr = self.factory.makeSourcePackageRelease(
+            sourcepackagename=source_package_name, version=versions[0],
+            changelog=changelog_lfa)
+        return self.factory.makeSourcePackagePublishingHistory(
+            sourcepackagerelease=spr, archive=distroseries.main_archive,
+            distroseries=distroseries,
+            status=PackagePublishingStatus.PUBLISHED)
+
+    def findDSD(self, derived_series, source_package_name):
+        return self.store.find(
+            DistroSeriesDifference, 
+            DistroSeriesDifference.derived_series == derived_series,
+            DistroSeriesDifference.source_package_name ==
+            source_package_name)
+
+    def runJob(self, job):
+        transaction.commit() # Switching DB user performs an abort.
+        self.layer.switchDbUser('distroseriesdifferencejob')
+        dsdjob = DistroSeriesDifferenceJob(job)
+        dsdjob.start()
+        dsdjob.run()
+        dsdjob.complete()
+        transaction.commit() # Switching DB user performs an abort.
+        self.layer.switchDbUser('launchpad')
+
+    def test_parent_gets_newer(self):
+        # When a new source package is uploaded to the parent distroseries,
+        # a job is created that updates the relevant DSD.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        self.createPublication(
+            source_package_name, ['1.0-1derived1', '1.0-1'], derived_series)
+        self.createPublication(
+            source_package_name, ['1.0-1'], derived_series.parent_series)
+        # Creating the SPPHs has created jobs for us, so grab it off the
+        # queue.
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual(1, ds_diff.count())
+        self.assertEqual('1.0-1', ds_diff[0].parent_source_version)
+        self.assertEqual('1.0-1derived1', ds_diff[0].source_version)
+        self.assertEqual('1.0-1', ds_diff[0].base_version)
+        # Now create a 1.0-2 upload to the parent.
+        self.createPublication(
+            source_package_name, ['1.0-2', '1.0-1'],
+            derived_series.parent_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        # And the DSD we have a hold of will have updated.
+        self.assertEqual('1.0-2', ds_diff[0].parent_source_version)
+        self.assertEqual('1.0-1derived1', ds_diff[0].source_version)
+        self.assertEqual('1.0-1', ds_diff[0].base_version)
+
+    def test_child_gets_newer(self):
+        # When a new source is uploaded to the child distroseries, the DSD is
+        # updated.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        self.createPublication(
+            source_package_name, ['1.0-1'], derived_series)
+        self.createPublication(
+            source_package_name, ['1.0-1'], derived_series.parent_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual(
+            DistroSeriesDifferenceStatus.RESOLVED, ds_diff[0].status)
+        self.createPublication(
+            source_package_name, ['2.0-0derived1', '1.0-1'], derived_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        self.assertEqual(
+            DistroSeriesDifferenceStatus.NEEDS_ATTENTION, ds_diff[0].status)
+        self.assertEqual('1.0-1', ds_diff[0].base_version)
+
+    def test_child_is_synced(self):
+        # If the source package gets 'synced' to the child from the parent,
+        # the job correctly updates the DSD.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        self.createPublication(
+            source_package_name, ['1.0-1derived1', '1.0-1'], derived_series)
+        self.createPublication(
+            source_package_name, ['1.0-2', '1.0-1'],
+            derived_series.parent_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual('1.0-1', ds_diff[0].base_version)
+        self.createPublication(
+            source_package_name, ['1.0-2', '1.0-1'], derived_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        self.assertEqual(
+            DistroSeriesDifferenceStatus.RESOLVED, ds_diff[0].status)
+        
+    def test_only_in_child(self):
+        # If a source package only exists in the child distroseries, the DSD
+        # is created with the right type.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        self.createPublication(
+            source_package_name, ['1.0-0derived1'], derived_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual(
+            DistroSeriesDifferenceType.UNIQUE_TO_DERIVED_SERIES,
+            ds_diff[0].difference_type)
+
+    def test_only_in_parent(self):
+        # If a source package only exists in the parent distroseries, the DSD
+        # is created with the right type.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        self.createPublication(
+            source_package_name, ['1.0-1'],
+            derived_series.parent_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual(
+            DistroSeriesDifferenceType.MISSING_FROM_DERIVED_SERIES,
+            ds_diff[0].difference_type)
+
+    def test_deleted_in_parent(self):
+        # If a source package is deleted in the parent, a job is created, and
+        # the DSD is updated correctly.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        self.createPublication(
+            source_package_name, ['1.0-1'], derived_series)
+        spph = self.createPublication(
+            source_package_name, ['1.0-1'], derived_series.parent_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual(
+            DistroSeriesDifferenceStatus.RESOLVED, ds_diff[0].status)
+        spph.requestDeletion(self.factory.makePerson())
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        self.assertEqual(
+            DistroSeriesDifferenceType.UNIQUE_TO_DERIVED_SERIES,
+            ds_diff[0].difference_type)
+
+    def test_deleted_in_child(self):
+        # If a source package is deleted in the child, a job is created, and
+        # the DSD is updated correctly.
+        derived_series = self.makeDerivedDistroSeries()
+        source_package_name = self.factory.makeSourcePackageName()
+        spph = self.createPublication(
+            source_package_name, ['1.0-1'], derived_series)
+        self.createPublication(
+            source_package_name, ['1.0-1'], derived_series.parent_series)
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        ds_diff = self.findDSD(derived_series, source_package_name)
+        self.assertEqual(
+            DistroSeriesDifferenceStatus.RESOLVED, ds_diff[0].status)
+        spph.requestDeletion(self.factory.makePerson())
+        jobs = find_waiting_jobs(derived_series, source_package_name)
+        self.runJob(jobs[0])
+        self.assertEqual(
+            DistroSeriesDifferenceType.MISSING_FROM_DERIVED_SERIES,
+            ds_diff[0].difference_type)
+
 
 class TestDistroSeriesDifferenceJobPermissions(TestCaseWithFactory):
     """Database permissions test for `DistroSeriesDifferenceJob`."""