Merge lp:~julian-edwards/launchpad/async-copying-part-2 into lp:launchpad

[Julian Edwards]

Expose a simple webservice API to copy packages asynchronously

This is the second part of my copyPackage(s) change for asynchronous package copying. This part adds copyPackages() which does multiple packages in one go similar to syncSources().

See the pre-requisite at https://code.launchpad.net/~julian-edwards/launchpad/async-copying-bug-809805/+merge/67941

[j.c.sackett (jcsackett)]

Looks good to me.

review: Approve

=== modified file 'database/schema/security.cfg'
--- database/schema/security.cfg	2011-07-14 10:05:11 +0000
+++ database/schema/security.cfg	2011-07-18 08:59:28 +0000
@@ -1058,6 +1058,7 @@
 public.sourcepackagepublishinghistory         = SELECT, INSERT
 public.sourcepackagerelease                   = SELECT
 public.sourcepackagereleasefile               = SELECT, INSERT, UPDATE
+public.teamparticipation                      = SELECT
 type=user
 
 [distroseriesdifferencejob]

=== modified file 'lib/canonical/launchpad/interfaces/_schema_circular_imports.py'
--- lib/canonical/launchpad/interfaces/_schema_circular_imports.py	2011-07-18 08:59:27 +0000
+++ lib/canonical/launchpad/interfaces/_schema_circular_imports.py	2011-07-18 08:59:28 +0000
@@ -405,6 +405,8 @@
 patch_plain_parameter_type(IArchive, 'syncSources', 'from_archive', IArchive)
 patch_plain_parameter_type(IArchive, 'syncSource', 'from_archive', IArchive)
 patch_plain_parameter_type(IArchive, 'copyPackage', 'from_archive', IArchive)
+patch_plain_parameter_type(
+    IArchive, 'copyPackages', 'from_archive', IArchive)
 patch_entry_return_type(IArchive, 'newSubscription', IArchiveSubscriber)
 patch_plain_parameter_type(
     IArchive, 'getArchiveDependency', 'dependency', IArchive)

=== modified file 'lib/lp/registry/browser/distroseries.py'
--- lib/lp/registry/browser/distroseries.py	2011-07-08 09:06:24 +0000
+++ lib/lp/registry/browser/distroseries.py	2011-07-18 08:59:28 +0000
@@ -1161,7 +1161,7 @@
             )
             for dsd in self.getUpgrades()]
         getUtility(IPlainPackageCopyJobSource).createMultiple(
-            target_distroseries, copies,
+            target_distroseries, copies, self.user,
             copy_policy=PackageCopyPolicy.MASS_SYNC)
 
         self.request.response.addInfoNotification(

=== modified file 'lib/lp/services/job/interfaces/job.py'
--- lib/lp/services/job/interfaces/job.py	2011-07-06 14:02:54 +0000
+++ lib/lp/services/job/interfaces/job.py	2011-07-18 08:59:28 +0000
@@ -22,6 +22,7 @@
     DBEnumeratedType,
     DBItem,
     )
+from lazr.restful.fields import Reference
 from zope.interface import (
     Attribute,
     Interface,
@@ -35,6 +36,7 @@
     )
 
 from canonical.launchpad import _
+from lp.registry.interfaces.person import IPerson
 
 
 class SuspendJobException(Exception):
@@ -109,6 +111,11 @@
     max_retries = Int(title=_(
         'The number of retries permitted before this job permanently fails.'))
 
+    requester = Reference(
+        IPerson, title=_("The person who requested the job"),
+        required=False, readonly=True
+        )
+
     is_pending = Bool(
         title=_("Whether or not this job's status is such that it "
                 "could eventually complete."))

=== modified file 'lib/lp/services/job/model/job.py'
--- lib/lp/services/job/model/job.py	2011-07-06 14:02:54 +0000
+++ lib/lp/services/job/model/job.py	2011-07-18 08:59:28 +0000
@@ -21,6 +21,10 @@
     Or,
     Select,
     )
+from storm.locals import (
+    Int,
+    Reference,
+    )
 from zope.interface import implements
 
 from canonical.database.constants import UTC_NOW
@@ -73,6 +77,9 @@
 
     max_retries = IntCol(default=0)
 
+    requester_id = Int(name='requester', allow_none=True)
+    requester = Reference(requester_id, 'Person.id')
+
     # Mapping of valid target states from a given state.
     _valid_transitions = {
         JobStatus.WAITING:
@@ -108,16 +115,19 @@
         return self.status in self.PENDING_STATUSES
 
     @classmethod
-    def createMultiple(self, store, num_jobs):
+    def createMultiple(self, store, num_jobs, requester=None):
         """Create multiple `Job`s at once.
 
         :param store: `Store` to ceate the jobs in.
         :param num_jobs: Number of `Job`s to create.
+        :param request: The `IPerson` requesting the jobs.
         :return: An iterable of `Job.id` values for the new jobs.
         """
-        job_contents = ["(%s)" % quote(JobStatus.WAITING)] * num_jobs
+        job_contents = [
+            "(%s, %s)" % (
+                quote(JobStatus.WAITING), quote(requester))] * num_jobs
         result = store.execute("""
-            INSERT INTO Job (status)
+            INSERT INTO Job (status, requester)
             VALUES %s
             RETURNING id
             """ % ", ".join(job_contents))

=== modified file 'lib/lp/services/job/tests/test_job.py'
--- lib/lp/services/job/tests/test_job.py	2011-07-06 14:02:54 +0000
+++ lib/lp/services/job/tests/test_job.py	2011-07-18 08:59:28 +0000
@@ -22,10 +22,13 @@
     Job,
     LeaseHeld,
     )
-from lp.testing import TestCase
-
-
-class TestJob(TestCase):
+from lp.testing import (
+    TestCase,
+    TestCaseWithFactory,
+    )
+
+
+class TestJob(TestCaseWithFactory):
     """Ensure Job behaves as intended."""
 
     layer = ZopelessDatabaseLayer
@@ -39,6 +42,12 @@
         job = Job()
         self.assertEqual(job.status, JobStatus.WAITING)
 
+    def test_stores_requester(self):
+        job = Job()
+        random_joe = self.factory.makePerson()
+        job.requester = random_joe
+        self.assertEqual(random_joe, job.requester)
+
     def test_createMultiple_creates_requested_number_of_jobs(self):
         job_ids = list(Job.createMultiple(IStore(Job), 3))
         self.assertEqual(3, len(job_ids))
@@ -55,6 +64,17 @@
         job = store.get(Job, Job.createMultiple(store, 1)[0])
         self.assertEqual(JobStatus.WAITING, job.status)
 
+    def test_createMultiple_sets_requester(self):
+        store = IStore(Job)
+        requester = self.factory.makePerson()
+        job = store.get(Job, Job.createMultiple(store, 1, requester)[0])
+        self.assertEqual(requester, job.requester)
+
+    def test_createMultiple_defaults_requester_to_None(self):
+        store = IStore(Job)
+        job = store.get(Job, Job.createMultiple(store, 1)[0])
+        self.assertEqual(None, job.requester)
+
     def test_start(self):
         """Job.start should update the object appropriately.
 

=== modified file 'lib/lp/soyuz/browser/archive.py'
--- lib/lp/soyuz/browser/archive.py	2011-06-29 21:49:36 +0000
+++ lib/lp/soyuz/browser/archive.py	2011-07-18 08:59:28 +0000
@@ -1284,7 +1284,8 @@
             spph.source_package_name, spph.archive, dest_archive, dest_series,
             dest_pocket, include_binaries=include_binaries,
             package_version=spph.sourcepackagerelease.version,
-            copy_policy=PackageCopyPolicy.INSECURE)
+            copy_policy=PackageCopyPolicy.INSECURE,
+            requester=person)
 
     return structured("""
         <p>Requested sync of %s packages.</p>

=== modified file 'lib/lp/soyuz/browser/tests/test_archive_webservice.py'
--- lib/lp/soyuz/browser/tests/test_archive_webservice.py	2011-07-18 08:59:27 +0000
+++ lib/lp/soyuz/browser/tests/test_archive_webservice.py	2011-07-18 08:59:28 +0000
@@ -19,7 +19,10 @@
 from canonical.testing.layers import DatabaseFunctionalLayer
 from lp.app.interfaces.launchpad import ILaunchpadCelebrities
 from lp.registry.interfaces.pocket import PackagePublishingPocket
-from lp.soyuz.enums import ArchivePurpose
+from lp.soyuz.enums import (
+    ArchivePurpose,
+    PackagePublishingStatus,
+    )
 from lp.soyuz.interfaces.packagecopyjob import IPlainPackageCopyJobSource
 from lp.soyuz.interfaces.processor import IProcessorFamilySet
 from lp.testing import (
@@ -311,17 +314,16 @@
 
 
 class TestCopyPackage(WebServiceTestCase):
-    """Webservice test cases for the copyPackage method"""
+    """Webservice test cases for the copyPackage/copyPackages methods"""
 
-    def test_copyPackage(self):
-        """Basic smoke test"""
+    def setup_data(self):
         self.ws_version = "devel"
         uploader_dude = self.factory.makePerson()
         source_archive = self.factory.makeArchive()
         target_archive = self.factory.makeArchive(
             purpose=ArchivePurpose.PRIMARY)
         source = self.factory.makeSourcePackagePublishingHistory(
-            archive=source_archive)
+            archive=source_archive, status=PackagePublishingStatus.PUBLISHED)
         source_name = source.source_package_name
         version = source.source_package_version
         to_pocket = PackagePublishingPocket.RELEASE
@@ -330,6 +332,13 @@
         with person_logged_in(target_archive.owner):
             target_archive.newComponentUploader(uploader_dude, "universe")
         transaction.commit()
+        return (source_archive, source_name, target_archive, to_pocket,
+                to_series, uploader_dude, version)
+
+    def test_copyPackage(self):
+        """Basic smoke test"""
+        (source_archive, source_name, target_archive, to_pocket, to_series,
+         uploader_dude, version) = self.setup_data()
 
         ws_target_archive = self.wsObject(target_archive, user=uploader_dude)
         ws_source_archive = self.wsObject(source_archive)
@@ -344,6 +353,23 @@
         copy_job = job_source.getActiveJobs(target_archive).one()
         self.assertEqual(target_archive, copy_job.target_archive)
 
+    def test_copyPackages(self):
+        """Basic smoke test"""
+        (source_archive, source_name, target_archive, to_pocket, to_series,
+         uploader_dude, version) = self.setup_data()
+
+        ws_target_archive = self.wsObject(target_archive, user=uploader_dude)
+        ws_source_archive = self.wsObject(source_archive)
+
+        ws_target_archive.copyPackages(
+            source_names=[source_name], from_archive=ws_source_archive,
+            to_pocket=to_pocket.name, to_series=to_series.name,
+            include_binaries=False)
+        transaction.commit()
+
+        job_source = getUtility(IPlainPackageCopyJobSource)
+        copy_job = job_source.getActiveJobs(target_archive).one()
+        self.assertEqual(target_archive, copy_job.target_archive)
 
 def test_suite():
     return unittest.TestLoader().loadTestsFromName(__name__)

=== modified file 'lib/lp/soyuz/browser/tests/test_package_copying_mixin.py'
--- lib/lp/soyuz/browser/tests/test_package_copying_mixin.py	2011-06-02 08:16:03 +0000
+++ lib/lp/soyuz/browser/tests/test_package_copying_mixin.py	2011-07-18 08:59:28 +0000
@@ -219,7 +219,7 @@
         pocket = self.factory.getAnyPocket()
         copy_asynchronously(
             [spph], archive, dest_series, pocket, include_binaries=False,
-            check_permissions=False)
+            check_permissions=False, person=self.factory.makePerson())
         self.assertEqual(None, find_spph_copy(archive, spph))
 
     def test_copy_synchronously_lists_packages(self):
@@ -243,7 +243,7 @@
         archive = dest_series.distribution.main_archive
         copy_asynchronously(
             [spph], archive, dest_series, pocket, include_binaries=False,
-            check_permissions=False)
+            check_permissions=False, person=self.factory.makePerson())
         jobs = list(getUtility(IPlainPackageCopyJobSource).getActiveJobs(
             archive))
         self.assertEqual(1, len(jobs))
@@ -263,7 +263,7 @@
         view.canCopySynchronously = FakeMethod(result=False)
         view.do_copy(
             'selected_differences', [spph], archive, dest_series, pocket,
-            False, check_permissions=False)
+            False, check_permissions=False, person=self.factory.makePerson())
         jobs = list(getUtility(IPlainPackageCopyJobSource).getActiveJobs(
             archive))
         self.assertNotEqual([], jobs)

=== modified file 'lib/lp/soyuz/interfaces/archive.py'
--- lib/lp/soyuz/interfaces/archive.py	2011-07-18 08:59:27 +0000
+++ lib/lp/soyuz/interfaces/archive.py	2011-07-18 08:59:28 +0000
@@ -1258,6 +1258,51 @@
         :raises CannotCopy: if there is a problem copying.
         """
 
+    @call_with(person=REQUEST_USER)
+    @operation_parameters(
+        source_names=List(
+            title=_("Source package names"),
+            value_type=TextLine()),
+        from_archive=Reference(schema=Interface),
+        #Really IArchive, see below
+        to_pocket=TextLine(title=_("Pocket name")),
+        to_series=TextLine(title=_("Distroseries name"), required=False),
+        include_binaries=Bool(
+            title=_("Include Binaries"),
+            description=_("Whether or not to copy binaries already built for"
+                          " this source"),
+            required=False))
+    @export_write_operation()
+    @operation_for_version('devel')
+    def copyPackages(source_names, from_archive, to_pocket, person,
+                     to_series=None, include_binaries=False):
+        """Atomically copy multiple named sources into this archive from another.
+
+        Asynchronously copy the most recent PUBLISHED versions of the named
+        sources to the destination archive if necessary.  Calls to this
+        method will return immediately if the copy passes basic security
+        checks and the copy will happen sometime later with full checking.
+
+        This operation will only succeed when all requested packages
+        are synchronised between the archives. If any of the requested
+        copies cannot be performed, the whole operation will fail. There
+        will be no partial changes of the destination archive.
+
+        :param source_names: a list of string names of packages to copy.
+        :param from_archive: the source archive from which to copy.
+        :param to_pocket: the target pocket (as a string).
+        :param to_series: the target distroseries (as a string).
+        :param include_binaries: optional boolean, controls whether or not
+            the published binaries for each given source should also be
+            copied along with the source.
+        :param person: the `IPerson` who requests the sync.
+
+        :raises NoSuchSourcePackageName: if the source name is invalid
+        :raises PocketNotFound: if the pocket name is invalid
+        :raises NoSuchDistroSeries: if the distro series name is invalid
+        :raises CannotCopy: if there is a problem copying.
+        """
+
 
 class IArchiveAppend(Interface):
     """Archive interface for operations restricted by append privilege."""

=== modified file 'lib/lp/soyuz/interfaces/packagecopyjob.py'
--- lib/lp/soyuz/interfaces/packagecopyjob.py	2011-07-12 14:23:40 +0000
+++ lib/lp/soyuz/interfaces/packagecopyjob.py	2011-07-18 08:59:28 +0000
@@ -126,7 +126,7 @@
     def create(package_name, source_archive,
                target_archive, target_distroseries, target_pocket,
                include_binaries=False, package_version=None,
-               copy_policy=PackageCopyPolicy.INSECURE):
+               copy_policy=PackageCopyPolicy.INSECURE, requester=None):
         """Create a new `IPlainPackageCopyJob`.
 
         :param package_name: The name of the source package to copy.
@@ -141,9 +141,10 @@
         :param package_version: The version string for the package version
             that is to be copied.
         :param copy_policy: Applicable `PackageCopyPolicy`.
+        :param requester: The user requesting the copy.
         """
 
-    def createMultiple(target_distroseries, copy_tasks,
+    def createMultiple(target_distroseries, copy_tasks, requester,
                        copy_policy=PackageCopyPolicy.INSECURE,
                        include_binaries=False):
         """Create multiple new `IPlainPackageCopyJob`s at once.
@@ -153,6 +154,7 @@
         :param copy_tasks: A list of tuples describing the copies to be
             performed: (package name, package version, source archive,
             target archive, target pocket).
+        :param requester: The user requesting the copy.
         :param copy_policy: Applicable `PackageCopyPolicy`.
         :param include_binaries: As in `do_copy`.
         :return: An iterable of `PackageCopyJob` ids.

=== modified file 'lib/lp/soyuz/model/archive.py'
--- lib/lp/soyuz/model/archive.py	2011-07-18 08:59:27 +0000
+++ lib/lp/soyuz/model/archive.py	2011-07-18 08:59:28 +0000
@@ -102,6 +102,7 @@
 from lp.registry.interfaces.sourcepackagename import ISourcePackageNameSet
 from lp.registry.model.sourcepackagename import SourcePackageName
 from lp.registry.model.teammembership import TeamParticipation
+from lp.services.database.bulk import load_related
 from lp.services.job.interfaces.job import JobStatus
 from lp.services.propertycache import (
     cachedproperty,
@@ -1550,6 +1551,47 @@
             package_version=version, include_binaries=include_binaries,
             copy_policy=PackageCopyPolicy.INSECURE)
 
+    def copyPackages(self, source_names, from_archive, to_pocket,
+                     person, to_series=None, include_binaries=None):
+        """See `IArchive`."""
+        sources = self._collectLatestPublishedSources(
+            from_archive, source_names)
+        if not sources:
+            raise CannotCopy(
+                "None of the supplied package names are published")
+
+        # Bulk-load the sourcepackagereleases so that the list
+        # comprehension doesn't generate additional queries. The
+        # sourcepackagenames themselves will already have been loaded when
+        # generating the list of source publications in "sources".
+        load_related(
+            SourcePackageRelease, sources, ["sourcepackagereleaseID"])
+        sourcepackagenames = [source.sourcepackagerelease.sourcepackagename
+                              for source in sources]
+
+        # Now do a mass check of permissions.
+        pocket = self._text_to_pocket(to_pocket)
+        series = self._text_to_series(to_series)
+        check_copy_permissions(
+            person, self, series, pocket, sourcepackagenames)
+
+        # If we get this far then we can create the PackageCopyJob.
+        copy_tasks = []
+        for source in sources:
+            task = (
+                source.sourcepackagerelease.sourcepackagename,
+                source.sourcepackagerelease.version,
+                from_archive,
+                self,
+                PackagePublishingPocket.RELEASE
+                )
+            copy_tasks.append(task)
+
+        job_source = getUtility(IPlainPackageCopyJobSource)
+        job_source.createMultiple(
+            series, copy_tasks, copy_policy=PackageCopyPolicy.INSECURE,
+            include_binaries=include_binaries)
+
     def _collectLatestPublishedSources(self, from_archive, source_names):
         """Private helper to collect the latest published sources for an
         archive.
@@ -1557,6 +1599,9 @@
         :raises NoSuchSourcePackageName: If any of the source_names do not
             exist.
         """
+        # XXX bigjools bug=810421
+        # This code is inefficient.  It should try to bulk load all the
+        # sourcepackagenames and publications instead of iterating.
         sources = []
         for name in source_names:
             # Check to see if the source package exists. This will raise

=== modified file 'lib/lp/soyuz/model/packagecopyjob.py'
--- lib/lp/soyuz/model/packagecopyjob.py	2011-07-08 09:06:24 +0000
+++ lib/lp/soyuz/model/packagecopyjob.py	2011-07-18 08:59:28 +0000
@@ -131,9 +131,11 @@
         return cls.wrap(IStore(PackageCopyJob).get(PackageCopyJob, pcj_id))
 
     def __init__(self, source_archive, target_archive, target_distroseries,
-                 job_type, metadata, package_name=None, copy_policy=None):
+                 job_type, metadata, requester, package_name=None,
+                 copy_policy=None):
         super(PackageCopyJob, self).__init__()
         self.job = Job()
+        self.job.requester = requester
         self.job_type = job_type
         self.source_archive = source_archive
         self.target_archive = target_archive
@@ -250,9 +252,10 @@
     def create(cls, package_name, source_archive,
                target_archive, target_distroseries, target_pocket,
                include_binaries=False, package_version=None,
-               copy_policy=PackageCopyPolicy.INSECURE):
+               copy_policy=PackageCopyPolicy.INSECURE, requester=None):
         """See `IPlainPackageCopyJobSource`."""
         assert package_version is not None, "No package version specified."
+        assert requester is not None, "No requester specified."
         metadata = cls._makeMetadata(
             target_pocket, package_version, include_binaries)
         job = PackageCopyJob(
@@ -262,7 +265,8 @@
             target_distroseries=target_distroseries,
             package_name=package_name,
             copy_policy=copy_policy,
-            metadata=metadata)
+            metadata=metadata,
+            requester=requester)
         IMasterStore(PackageCopyJob).add(job)
         return cls(job)
 
@@ -292,12 +296,12 @@
         return format_string % sqlvalues(*data)
 
     @classmethod
-    def createMultiple(cls, target_distroseries, copy_tasks,
+    def createMultiple(cls, target_distroseries, copy_tasks, requester,
                        copy_policy=PackageCopyPolicy.INSECURE,
                        include_binaries=False):
         """See `IPlainPackageCopyJobSource`."""
         store = IMasterStore(Job)
-        job_ids = Job.createMultiple(store, len(copy_tasks))
+        job_ids = Job.createMultiple(store, len(copy_tasks), requester)
         job_contents = [
             cls._composeJobInsertionTuple(
                 target_distroseries, copy_policy, include_binaries, job_id,
@@ -478,13 +482,14 @@
         do_copy(
             sources=[source_package], archive=self.target_archive,
             series=self.target_distroseries, pocket=self.target_pocket,
-            include_binaries=self.include_binaries, check_permissions=False,
-            overrides=[override], send_email=send_email)
+            include_binaries=self.include_binaries, check_permissions=True,
+            person=self.requester, overrides=[override],
+            send_email=send_email)
 
         if pu is not None:
             # A PackageUpload will only exist if the copy job had to be
             # held in the queue because of policy/ancestry checks.  If one
-            # does exist we need to make sure 
+            # does exist we need to make sure it gets moved to DONE.
             pu.setDone()
 
     def abort(self):

=== modified file 'lib/lp/soyuz/tests/test_archive.py'
--- lib/lp/soyuz/tests/test_archive.py	2011-07-18 08:59:27 +0000
+++ lib/lp/soyuz/tests/test_archive.py	2011-07-18 08:59:28 +0000
@@ -1975,7 +1975,7 @@
         source_archive = self.factory.makeArchive()
         target_archive = self.factory.makeArchive(purpose=target_purpose)
         source = self.factory.makeSourcePackagePublishingHistory(
-            archive=source_archive)
+            archive=source_archive, status=PackagePublishingStatus.PUBLISHED)
         source_name = source.source_package_name
         version = source.source_package_version
         to_pocket = PackagePublishingPocket.RELEASE
@@ -2068,3 +2068,90 @@
             to_pocket.name, to_series=to_series.name, include_binaries=False,
             person=target_archive.owner)
 
+    def test_copyPackages_with_single_package(self):
+        (source, source_archive, source_name, target_archive, to_pocket,
+         to_series, version) = self._setup_copy_data()
+
+        with person_logged_in(target_archive.owner):
+            target_archive.copyPackages(
+                [source_name], source_archive, to_pocket.name,
+                to_series=to_series.name, include_binaries=False,
+                person=target_archive.owner)
+
+        # The source should not be published yet in the target_archive.
+        published = target_archive.getPublishedSources(
+            name=source.source_package_name).any()
+        self.assertIs(None, published)
+
+        # There should be one copy job.
+        job_source = getUtility(IPlainPackageCopyJobSource)
+        copy_job = job_source.getActiveJobs(target_archive).one()
+        self.assertEqual(target_archive, copy_job.target_archive)
+
+    def test_copyPackages_with_multiple_packages(self):
+        (source, source_archive, source_name, target_archive, to_pocket,
+         to_series, version) = self._setup_copy_data()
+        sources = [source]
+        sources.append(self.factory.makeSourcePackagePublishingHistory(
+            archive=source_archive,
+            status=PackagePublishingStatus.PUBLISHED))
+        sources.append(self.factory.makeSourcePackagePublishingHistory(
+            archive=source_archive,
+            status=PackagePublishingStatus.PUBLISHED))
+        names = [source.sourcepackagerelease.sourcepackagename.name
+                 for source in sources]
+
+        with person_logged_in(target_archive.owner):
+            target_archive.copyPackages(
+                names, source_archive, to_pocket.name,
+                to_series=to_series.name, include_binaries=False,
+                person=target_archive.owner)
+
+        # Make sure three copy jobs exist.
+        job_source = getUtility(IPlainPackageCopyJobSource)
+        copy_jobs = job_source.getActiveJobs(target_archive)
+        self.assertEqual(3, copy_jobs.count())
+
+    def test_copyPackages_disallows_non_primary_archive_uploaders(self):
+        # If copying to a primary archive and you're not an uploader for
+        # the package then you can't copy.
+        (source, source_archive, source_name, target_archive, to_pocket,
+         to_series, version) = self._setup_copy_data(
+            target_purpose=ArchivePurpose.PRIMARY)
+        person = self.factory.makePerson()
+        self.assertRaises(
+            CannotCopy,
+            target_archive.copyPackages, [source_name], source_archive,
+            to_pocket.name, to_series=to_series.name, include_binaries=False,
+            person=person)
+
+    def test_copyPackages_allows_primary_archive_uploaders(self):
+        # Copying to a primary archive if you're already an uploader is OK.
+        (source, source_archive, source_name, target_archive, to_pocket,
+         to_series, version) = self._setup_copy_data(
+            target_purpose=ArchivePurpose.PRIMARY)
+        person = self.factory.makePerson()
+        with person_logged_in(target_archive.owner):
+            target_archive.newComponentUploader(person, "universe")
+        target_archive.copyPackages(
+            [source_name], source_archive, to_pocket.name,
+            to_series=to_series.name, include_binaries=False,
+            person=person)
+
+        # There should be one copy job.
+        job_source = getUtility(IPlainPackageCopyJobSource)
+        copy_job = job_source.getActiveJobs(target_archive).one()
+        self.assertEqual(target_archive, copy_job.target_archive)
+
+    def test_copyPackages_disallows_non_PPA_owners(self):
+        # Only people with launchpad.Append are allowed to call copyPackage.
+        (source, source_archive, source_name, target_archive, to_pocket,
+         to_series, version) = self._setup_copy_data()
+        person = self.factory.makePerson()
+        self.assertTrue(target_archive.is_ppa)
+        self.assertRaises(
+            CannotCopy,
+            target_archive.copyPackages, [source_name], source_archive,
+            to_pocket.name, to_series=to_series.name, include_binaries=False,
+            person=person)
+

=== modified file 'lib/lp/soyuz/tests/test_packagecopyjob.py'
--- lib/lp/soyuz/tests/test_packagecopyjob.py	2011-07-12 14:52:52 +0000
+++ lib/lp/soyuz/tests/test_packagecopyjob.py	2011-07-18 08:59:28 +0000
@@ -3,6 +3,7 @@
 
 """Tests for sync package jobs."""
 
+import operator
 from testtools.content import text_content
 from testtools.matchers import (
     Equals,
@@ -92,9 +93,10 @@
         target_archive = dsd.derived_series.main_archive
         target_distroseries = dsd.derived_series
         target_pocket = self.factory.getAnyPocket()
+        requester = self.factory.makePerson()
         return getUtility(IPlainPackageCopyJobSource).create(
             dsd.source_package_name.name, source_archive, target_archive,
-            target_distroseries, target_pocket,
+            target_distroseries, target_pocket, requester=requester,
             package_version=dsd.parent_source_version, **kwargs)
 
     def runJob(self, job):
@@ -122,13 +124,15 @@
         distroseries = self.factory.makeDistroSeries()
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
+        requester = self.factory.makePerson()
         source = getUtility(IPlainPackageCopyJobSource)
         job = source.create(
             package_name="foo", source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
             package_version="1.0-1", include_binaries=False,
-            copy_policy=PackageCopyPolicy.MASS_SYNC)
+            copy_policy=PackageCopyPolicy.MASS_SYNC,
+            requester=requester)
         self.assertProvides(job, IPackageCopyJob)
         self.assertEquals(archive1.id, job.source_archive_id)
         self.assertEquals(archive1, job.source_archive)
@@ -140,6 +144,7 @@
         self.assertEqual("1.0-1", job.package_version)
         self.assertEquals(False, job.include_binaries)
         self.assertEquals(PackageCopyPolicy.MASS_SYNC, job.copy_policy)
+        self.assertEqual(requester, job.requester)
 
     def test_createMultiple_creates_one_job_per_copy(self):
         mother = self.factory.makeDistroSeriesParent()
@@ -148,6 +153,7 @@
             derived_series=derived_series)
         mother_package = self.factory.makeSourcePackageName()
         father_package = self.factory.makeSourcePackageName()
+        requester = self.factory.makePerson()
         job_source = getUtility(IPlainPackageCopyJobSource)
         copy_tasks = [
             (
@@ -166,7 +172,8 @@
                 ),
             ]
         job_ids = list(
-            job_source.createMultiple(mother.derived_series, copy_tasks))
+            job_source.createMultiple(mother.derived_series, copy_tasks,
+                                      requester))
         jobs = list(job_source.getActiveJobs(derived_series.main_archive))
         self.assertContentEqual(job_ids, [job.id for job in jobs])
         self.assertEqual(len(copy_tasks), len(set([job.job for job in jobs])))
@@ -185,17 +192,24 @@
             for job in jobs]
         self.assertEqual(copy_tasks, requested_copies)
 
+        # The passed requester should be the same on all jobs.
+        actual_requester = set(job.requester for job in jobs)
+        self.assertEqual(1, len(actual_requester))
+        self.assertEqual(requester, jobs[0].requester)
+
     def test_getActiveJobs(self):
         # getActiveJobs() can retrieve all active jobs for an archive.
         distroseries = self.factory.makeDistroSeries()
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson()
         job = source.create(
             package_name="foo", source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            package_version="1.0-1", include_binaries=False)
+            package_version="1.0-1", include_binaries=False,
+            requester=requester)
         self.assertContentEqual([job], source.getActiveJobs(archive2))
 
     def test_getActiveJobs_gets_oldest_first(self):
@@ -249,12 +263,14 @@
         distroseries = self.factory.makeDistroSeries()
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
+        requester = self.factory.makePerson()
         job_source = getUtility(IPlainPackageCopyJobSource)
         job = job_source.create(
             package_name="foo", source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            package_version="1.0-1", include_binaries=False)
+            package_version="1.0-1", include_binaries=False,
+            requester=requester)
         naked_job = removeSecurityProxy(job)
         naked_job.reportFailure = FakeMethod()
 
@@ -268,12 +284,14 @@
         package = self.factory.makeSourcePackageName()
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
+        requester = self.factory.makePerson()
         source = getUtility(IPlainPackageCopyJobSource)
         job = source.create(
             package_name=package.name, source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.UPDATES,
-            include_binaries=False, package_version='1.0')
+            include_binaries=False, package_version='1.0',
+            requester=requester)
 
         naked_job = removeSecurityProxy(job)
         naked_job.reportFailure = FakeMethod()
@@ -315,12 +333,16 @@
             archive=target_archive)
 
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson()
+        with person_logged_in(target_archive.owner):
+            target_archive.newComponentUploader(requester, "main")
         job = source.create(
             package_name="libc",
             source_archive=breezy_archive, target_archive=target_archive,
             target_distroseries=target_series,
             target_pocket=PackagePublishingPocket.RELEASE,
-            package_version="2.8-1", include_binaries=False)
+            package_version="2.8-1", include_binaries=False,
+            requester=requester)
         self.assertEqual("libc", job.package_name)
         self.assertEqual("2.8-1", job.package_version)
 
@@ -348,12 +370,14 @@
         distroseries = self.factory.makeDistroSeries()
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
+        requester = self.factory.makePerson()
         source = getUtility(IPlainPackageCopyJobSource)
         job = source.create(
             package_name="foo", source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            package_version="1.0-1", include_binaries=False)
+            package_version="1.0-1", include_binaries=False,
+            requester=requester)
         oops_vars = job.getOopsVars()
         naked_job = removeSecurityProxy(job)
         self.assertIn(
@@ -374,6 +398,7 @@
         distroseries = publisher.breezy_autotest
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
+        requester = self.factory.makePerson()
         publisher.getPubSource(
             distroseries=distroseries, sourcename="libc",
             version="2.8-1", status=PackagePublishingStatus.PUBLISHED,
@@ -382,7 +407,10 @@
             package_name="libc", source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            package_version="2.8-1", include_binaries=False)
+            package_version="2.8-1", include_binaries=False,
+            requester=requester)
+        with person_logged_in(archive2.owner):
+            archive2.newComponentUploader(requester, "main")
         transaction.commit()
 
         out, err, exit_code = run_script(
@@ -401,12 +429,14 @@
         distroseries = self.factory.makeDistroSeries()
         archive1 = self.factory.makeArchive(distroseries.distribution)
         archive2 = self.factory.makeArchive(distroseries.distribution)
+        requester = self.factory.makePerson()
         source = getUtility(IPlainPackageCopyJobSource)
         job = source.create(
             package_name="foo", source_archive=archive1,
             target_archive=archive2, target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            package_version="1.0-1", include_binaries=True)
+            package_version="1.0-1", include_binaries=True,
+            requester=requester)
         self.assertEqual(
             ("<PlainPackageCopyJob to copy package foo from "
              "{distroseries.distribution.name}/{archive1.name} to "
@@ -519,6 +549,9 @@
         # target_archive.
 
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson()
+        with person_logged_in(target_archive.owner):
+            target_archive.newComponentUploader(requester, "restricted")
         job = source.create(
             package_name="libc",
             package_version="2.8-1",
@@ -526,7 +559,8 @@
             target_archive=target_archive,
             target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            include_binaries=False)
+            include_binaries=False,
+            requester=requester)
 
         self.runJob(job)
 
@@ -556,6 +590,9 @@
         # Now, run the copy job, which should raise an error because
         # there's no ancestry.
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson()
+        with person_logged_in(target_archive.owner):
+            target_archive.newComponentUploader(requester, "main")
         job = source.create(
             package_name="copyme",
             package_version="2.8-1",
@@ -563,7 +600,8 @@
             target_archive=target_archive,
             target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            include_binaries=False)
+            include_binaries=False,
+            requester=requester)
 
         self.assertRaises(SuspendJobException, self.runJob, job)
         # Simulate the job runner suspending after getting a
@@ -619,6 +657,7 @@
         # Now, run the copy job.
 
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson()
         job = source.create(
             package_name="copyme",
             package_version="2.8-1",
@@ -626,7 +665,8 @@
             target_archive=target_archive,
             target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            include_binaries=False)
+            include_binaries=False,
+            requester=requester)
 
         # The job should be suspended and there's a PackageUpload with
         # its package_copy_job set.
@@ -672,6 +712,7 @@
 
         # Now, run the copy job.
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson()
         job = source.create(
             package_name="copyme",
             package_version="2.8-1",
@@ -679,7 +720,8 @@
             target_archive=target_archive,
             target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            include_binaries=False)
+            include_binaries=False,
+            requester=requester)
 
         # The job should be suspended and there's a PackageUpload with
         # its package_copy_job set in the UNAPPROVED queue.
@@ -703,6 +745,7 @@
         publisher = SoyuzTestPublisher()
         publisher.prepareBreezyAutotest()
         distroseries = publisher.breezy_autotest
+        distroseries.changeslist = "[e-mail]"
 
         target_archive = self.factory.makeArchive(
             distroseries.distribution, purpose=ArchivePurpose.PRIMARY)
@@ -715,6 +758,9 @@
             archive=source_archive)
 
         source = getUtility(IPlainPackageCopyJobSource)
+        requester = self.factory.makePerson(email="[e-mail]")
+        with person_logged_in(target_archive.owner):
+            target_archive.newComponentUploader(requester, "main")
         job = source.create(
             package_name="copyme",
             package_version="2.8-1",
@@ -722,7 +768,8 @@
             target_archive=target_archive,
             target_distroseries=distroseries,
             target_pocket=PackagePublishingPocket.RELEASE,
-            include_binaries=False)
+            include_binaries=False,
+            requester=requester)
 
         # Run the job so it gains a PackageUpload.
         self.assertRaises(SuspendJobException, self.runJob, job)
@@ -736,6 +783,9 @@
         pu = getUtility(IPackageUploadSet).getByPackageCopyJobIDs(
             [removeSecurityProxy(job).context.id]).one()
         pu.acceptFromQueue()
+        # Clear existing emails so we can see only the ones the job
+        # generates later.
+        pop_notifications()
         self.runJob(job)
 
         # The job should have set the PU status to DONE:
@@ -745,6 +795,16 @@
         existing_sources = target_archive.getPublishedSources(name='copyme')
         self.assertIsNot(None, existing_sources.any())
 
+        # It would be nice to test emails in a separate test but it would
+        # require all of the same setup as above again so we might as well
+        # do it here.
+        emails = pop_notifications(sort_key=operator.itemgetter('To'))
+
+        # We expect an uploader email and an announcement to the changeslist.
+        self.assertEquals(2, len(emails))
+        self.assertIn("[e-mail]", emails[0]['To'])
+        self.assertIn("[e-mail]", emails[1]['To'])
+
     def test_findMatchingDSDs_matches_all_DSDs_for_job(self):
         # findMatchingDSDs finds matching DSDs for any of the packages
         # in the job.

=== modified file 'lib/lp/testing/factory.py'
--- lib/lp/testing/factory.py	2011-07-13 20:55:34 +0000
+++ lib/lp/testing/factory.py	2011-07-18 08:59:28 +0000
@@ -4165,7 +4165,8 @@
 
     def makePlainPackageCopyJob(
         self, package_name=None, package_version=None, source_archive=None,
-        target_archive=None, target_distroseries=None, target_pocket=None):
+        target_archive=None, target_distroseries=None, target_pocket=None,
+        requester=None):
         """Create a new `PlainPackageCopyJob`."""
         if package_name is None and package_version is None:
             package_name = self.makeSourcePackageName().name
@@ -4178,10 +4179,12 @@
             target_distroseries = self.makeDistroSeries()
         if target_pocket is None:
             target_pocket = self.getAnyPocket()
+        if requester is None:
+            requester = self.makePerson()
         return getUtility(IPlainPackageCopyJobSource).create(
             package_name, source_archive, target_archive,
             target_distroseries, target_pocket,
-            package_version=package_version)
+            package_version=package_version, requester=requester)
 
 
 # Some factory methods return simple Python types. We don't add