~launchpad-pqm/launchpad/devel

# Copyright 2010-2011 Canonical Ltd.  This software is licensed under the

# GNU Affero General Public License version 3 (see the file LICENSE).

__metaclass__ = type

__all__ = [

    "PackageCopyJob",

    "PlainPackageCopyJob",

    ]

import logging

from lazr.delegates import delegates

import simplejson

from storm.locals import (

    And,

    Int,

    JSON,

    Reference,

    Unicode,

    )

import transaction

from zope.component import getUtility

from zope.interface import (

    classProvides,

    implements,

    )

from zope.security.proxy import removeSecurityProxy

from canonical.database.enumcol import EnumCol

from canonical.database.sqlbase import sqlvalues

from lp.services.database.lpstorm import (

    IMasterStore,

    IStore,

    )

from lp.app.errors import NotFoundError

from lp.app.interfaces.launchpad import ILaunchpadCelebrities

from lp.registry.enum import DistroSeriesDifferenceStatus

from lp.registry.interfaces.distroseriesdifference import (

    IDistroSeriesDifferenceSource,

    )

from lp.registry.interfaces.distroseriesdifferencecomment import (

    IDistroSeriesDifferenceCommentSource,

    )

from lp.registry.interfaces.person import IPersonSet

from lp.registry.interfaces.pocket import PackagePublishingPocket

from lp.registry.interfaces.sourcepackagename import ISourcePackageNameSet

from lp.registry.model.distroseries import DistroSeries

from lp.services.database.decoratedresultset import DecoratedResultSet

from lp.services.database.stormbase import StormBase

from lp.services.job.interfaces.job import (

    JobStatus,

    SuspendJobException,

    )

from lp.services.job.model.job import Job

from lp.services.job.runner import BaseRunnableJob

from lp.soyuz.adapters.overrides import (

    FromExistingOverridePolicy,

    SourceOverride,

    UnknownOverridePolicy,

    )

from lp.soyuz.enums import PackageCopyPolicy

from lp.soyuz.interfaces.archive import CannotCopy

from lp.soyuz.interfaces.component import IComponentSet

from lp.soyuz.interfaces.copypolicy import ICopyPolicy

from lp.soyuz.interfaces.packagecopyjob import (

    IPackageCopyJob,

    IPackageCopyJobSource,

    IPlainPackageCopyJob,

    IPlainPackageCopyJobSource,

    PackageCopyJobType,

    )

from lp.soyuz.interfaces.packagediff import PackageDiffAlreadyRequested

from lp.soyuz.interfaces.queue import IPackageUploadSet

from lp.soyuz.interfaces.section import ISectionSet

from lp.soyuz.model.archive import Archive

from lp.soyuz.scripts.packagecopier import do_copy

class PackageCopyJob(StormBase):

    """Base class for package copying jobs."""

    implements(IPackageCopyJob)

    classProvides(IPackageCopyJobSource)

    __storm_table__ = 'PackageCopyJob'

    id = Int(primary=True)

    job_id = Int(name='job')

    job = Reference(job_id, Job.id)

    source_archive_id = Int(name='source_archive')

    source_archive = Reference(source_archive_id, Archive.id)

    target_archive_id = Int(name='target_archive')

    target_archive = Reference(target_archive_id, Archive.id)

    target_distroseries_id = Int(name='target_distroseries')

    target_distroseries = Reference(target_distroseries_id, DistroSeries.id)

    package_name = Unicode('package_name')

    copy_policy = EnumCol(enum=PackageCopyPolicy)

    job_type = EnumCol(enum=PackageCopyJobType, notNull=True)

    metadata = JSON('json_data')

    # Derived concrete classes.  The entire class gets one dict for

    # this; it's not meant to be on an instance.

    concrete_classes = {}

    @classmethod

    def registerConcreteClass(cls, new_class):

        """Register a concrete `IPackageCopyJob`-implementing class."""

        assert new_class.class_job_type not in cls.concrete_classes, (

            "Class %s is already registered." % new_class)

        cls.concrete_classes[new_class.class_job_type] = new_class

    @classmethod

    def wrap(cls, package_copy_job):

        """See `IPackageCopyJobSource`."""

        if package_copy_job is None:

            return None

        # Read job_type so You Don't Have To.  If any other code reads

        # job_type, that's probably a sign that the interfaces need more

        # work.

        job_type = removeSecurityProxy(package_copy_job).job_type

        concrete_class = cls.concrete_classes[job_type]

        return concrete_class(package_copy_job)

    @classmethod

    def getByID(cls, pcj_id):

        """See `IPackageCopyJobSource`."""

        return cls.wrap(IStore(PackageCopyJob).get(PackageCopyJob, pcj_id))

    def __init__(self, source_archive, target_archive, target_distroseries,

                 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

        self.target_distroseries = target_distroseries

        self.package_name = unicode(package_name)

        self.copy_policy = copy_policy

        self.metadata = metadata

    @property

    def package_version(self):

        return self.metadata["package_version"]

    def extendMetadata(self, metadata_dict):

        """Add metadata_dict to the existing metadata."""

        existing = self.metadata

        existing.update(metadata_dict)

        self.metadata = existing

    @property

    def component_name(self):

        """See `IPackageCopyJob`."""

        return self.metadata.get("component_override")

    @property

    def section_name(self):

        """See `IPackageCopyJob`."""

        return self.metadata.get("section_override")

class PackageCopyJobDerived(BaseRunnableJob):

    """Abstract class for deriving from PackageCopyJob."""

    delegates(IPackageCopyJob)

    def __init__(self, job):

        self.context = job

    @classmethod

    def get(cls, job_id):

        """Get a job by id.

        :return: the PackageCopyJob with the specified id, as the current

            PackageCopyJobDerived subclass.

        :raises: NotFoundError if there is no job with the specified id, or

            its job_type does not match the desired subclass.

        """

        job = IStore(PackageCopyJob).get(PackageCopyJob, job_id)

        if job.job_type != cls.class_job_type:

            raise NotFoundError(

                'No object found with id %d and type %s' % (job_id,

                cls.class_job_type.title))

        return cls(job)

    @classmethod

    def iterReady(cls):

        """Iterate through all ready PackageCopyJobs."""

        jobs = IStore(PackageCopyJob).find(

            PackageCopyJob,

            And(PackageCopyJob.job_type == cls.class_job_type,

                PackageCopyJob.job == Job.id,

                Job.id.is_in(Job.ready_jobs)))

        return (cls(job) for job in jobs)

    def getOopsVars(self):

        """See `IRunnableJob`."""

        vars = super(PackageCopyJobDerived, self).getOopsVars()

        vars.extend([

            ('source_archive_id', self.context.source_archive_id),

            ('target_archive_id', self.context.target_archive_id),

            ('target_distroseries_id', self.context.target_distroseries_id),

            ('package_copy_job_id', self.context.id),

            ('package_copy_job_type', self.context.job_type.title),

            ])

        return vars

    @property

    def copy_policy(self):

        """See `PlainPackageCopyJob`."""

        return self.context.copy_policy

class PlainPackageCopyJob(PackageCopyJobDerived):

    """Job that copies a package from one archive to another."""

    # This job type serves in different places: it supports copying

    # packages between archives, but also the syncing of packages from

    # parents into a derived distroseries.  We may split these into

    # separate types at some point, but for now we (allenap, bigjools,

    # jtv) chose to keep it as one.

    implements(IPlainPackageCopyJob)

    class_job_type = PackageCopyJobType.PLAIN

    classProvides(IPlainPackageCopyJobSource)

    @classmethod

    def _makeMetadata(cls, target_pocket, package_version,

                      include_binaries, sponsored=None):

        """Produce a metadata dict for this job."""

        if sponsored:

            sponsored_name = sponsored.name

        else:

            sponsored_name = None

        return {

            'target_pocket': target_pocket.value,

            'package_version': package_version,

            'include_binaries': bool(include_binaries),

            'sponsored': sponsored_name,

        }

    @classmethod

    def create(cls, package_name, source_archive,

               target_archive, target_distroseries, target_pocket,

               include_binaries=False, package_version=None,

               copy_policy=PackageCopyPolicy.INSECURE, requester=None,

               sponsored=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, sponsored)

        job = PackageCopyJob(

            job_type=cls.class_job_type,

            source_archive=source_archive,

            target_archive=target_archive,

            target_distroseries=target_distroseries,

            package_name=package_name,

            copy_policy=copy_policy,

            metadata=metadata,

            requester=requester)

        IMasterStore(PackageCopyJob).add(job)

        return cls(job)

    @classmethod

    def _composeJobInsertionTuple(cls, target_distroseries, copy_policy,

                                  include_binaries, job_id, copy_task,

                                  sponsored):

        """Create an SQL fragment for inserting a job into the database.

        :return: A string representing an SQL tuple containing initializers

            for a `PackageCopyJob` in the database (minus `id`, which is

            assigned automatically).  Contents are escaped for use in SQL.

        """

        (

            package_name,

            package_version,

            source_archive,

            target_archive,

            target_pocket,

        ) = copy_task

        metadata = cls._makeMetadata(

            target_pocket, package_version, include_binaries, sponsored)

        data = (

            cls.class_job_type, target_distroseries, copy_policy,

            source_archive, target_archive, package_name, job_id,

            simplejson.dumps(metadata, ensure_ascii=False))

        format_string = "(%s)" % ", ".join(["%s"] * len(data))

        return format_string % sqlvalues(*data)

    @classmethod

    def createMultiple(cls, target_distroseries, copy_tasks, requester,

                       copy_policy=PackageCopyPolicy.INSECURE,

                       include_binaries=False, sponsored=None):

        """See `IPlainPackageCopyJobSource`."""

        store = IMasterStore(Job)

        job_ids = Job.createMultiple(store, len(copy_tasks), requester)

        job_contents = [

            cls._composeJobInsertionTuple(

                target_distroseries, copy_policy, include_binaries, job_id,

                task, sponsored)

            for job_id, task in zip(job_ids, copy_tasks)]

        result = store.execute("""

            INSERT INTO PackageCopyJob (

                job_type,

                target_distroseries,

                copy_policy,

                source_archive,

                target_archive,

                package_name,

                job,

                json_data)

            VALUES %s

            RETURNING id

            """ % ", ".join(job_contents))

        return [job_id for job_id, in result]

    @classmethod

    def getActiveJobs(cls, target_archive):

        """See `IPlainPackageCopyJobSource`."""

        jobs = IStore(PackageCopyJob).find(

            PackageCopyJob,

            PackageCopyJob.job_type == cls.class_job_type,

            PackageCopyJob.target_archive == target_archive,

            Job.id == PackageCopyJob.job_id,

            Job._status == JobStatus.WAITING)

        jobs = jobs.order_by(PackageCopyJob.id)

        return DecoratedResultSet(jobs, cls)

    @classmethod

    def getPendingJobsForTargetSeries(cls, target_series):

        """Get upcoming jobs for `target_series`, ordered by age."""

        raw_jobs = IStore(PackageCopyJob).find(

            PackageCopyJob,

            Job.id == PackageCopyJob.job_id,

            PackageCopyJob.job_type == cls.class_job_type,

            PackageCopyJob.target_distroseries == target_series,

            Job._status.is_in(Job.PENDING_STATUSES))

        raw_jobs = raw_jobs.order_by(PackageCopyJob.id)

        return DecoratedResultSet(raw_jobs, cls)

    @classmethod

    def getPendingJobsPerPackage(cls, target_series):

        """See `IPlainPackageCopyJobSource`."""

        result = {}

        # Go through jobs in-order, picking the first matching job for

        # any (package, version) tuple.  Because of how

        # getPendingJobsForTargetSeries orders its results, the first

        # will be the oldest and thus presumably the first to finish.

        for job in cls.getPendingJobsForTargetSeries(target_series):

            result.setdefault(job.package_name, job)

        return result

    @classmethod

    def getIncompleteJobsForArchive(cls, archive):

        """See `IPlainPackageCopyJobSource`."""

        jobs = IStore(PackageCopyJob).find(

            PackageCopyJob,

            PackageCopyJob.target_archive == archive,

            PackageCopyJob.job_type == cls.class_job_type,

            Job.id == PackageCopyJob.job_id,

            Job._status.is_in(

                [JobStatus.WAITING, JobStatus.RUNNING, JobStatus.FAILED])

            )

        return DecoratedResultSet(jobs, cls)

    @property

    def target_pocket(self):

        return PackagePublishingPocket.items[self.metadata['target_pocket']]

    @property

    def include_binaries(self):

        return self.metadata['include_binaries']

    @property

    def sponsored(self):

        name = self.metadata['sponsored']

        if name is None:

            return None

        return getUtility(IPersonSet).getByName(name)

    def _createPackageUpload(self, unapproved=False):

        pu = self.target_distroseries.createQueueEntry(

            pocket=self.target_pocket, archive=self.target_archive,

            package_copy_job=self.context)

        if unapproved:

            pu.setUnapproved()

    def addSourceOverride(self, override):

        """Add an `ISourceOverride` to the metadata."""

        metadata_changes = {}

        if override.component is not None:

            metadata_changes['component_override'] = override.component.name

        if override.section is not None:

            metadata_changes['section_override'] = override.section.name

        self.context.extendMetadata(metadata_changes)

    def getSourceOverride(self):

        """Fetch an `ISourceOverride` from the metadata."""

        name = self.package_name

        component_name = self.component_name

        section_name = self.section_name

        source_package_name = getUtility(ISourcePackageNameSet)[name]

        try:

            component = getUtility(IComponentSet)[component_name]

        except NotFoundError:

            component = None

        try:

            section = getUtility(ISectionSet)[section_name]

        except NotFoundError:

            section = None

        return SourceOverride(source_package_name, component, section)

    def _checkPolicies(self, source_name, source_component=None):

        # This helper will only return if it's safe to carry on with the

        # copy, otherwise it raises SuspendJobException to tell the job

        # runner to suspend the job.

        override_policy = FromExistingOverridePolicy()

        ancestry = override_policy.calculateSourceOverrides(

            self.target_archive, self.target_distroseries,

            self.target_pocket, [source_name])

        copy_policy = self.getPolicyImplementation()

        if len(ancestry) == 0:

            # We need to get the default overrides and put them in the

            # metadata.

            defaults = UnknownOverridePolicy().calculateSourceOverrides(

                self.target_archive, self.target_distroseries,

                self.target_pocket, [source_name], source_component)

            self.addSourceOverride(defaults[0])

            approve_new = copy_policy.autoApproveNew(

                self.target_archive, self.target_distroseries,

                self.target_pocket)

            if not approve_new:

                # There's no existing package with the same name and the

                # policy says unapproved, so we poke it in the NEW queue.

                self._createPackageUpload()

                raise SuspendJobException

        else:

            # Put the existing override in the metadata.

            self.addSourceOverride(ancestry[0])

        # The package is not new (it has ancestry) so check the copy

        # policy for existing packages.

        approve_existing = copy_policy.autoApprove(

            self.target_archive, self.target_distroseries, self.target_pocket)

        if not approve_existing:

            self._createPackageUpload(unapproved=True)

            raise SuspendJobException

    def _rejectPackageUpload(self):

        # Helper to find and reject any associated PackageUpload.

        pu = getUtility(IPackageUploadSet).getByPackageCopyJobIDs(

            [self.context.id]).any()

        if pu is not None:

            pu.setRejected()

    def run(self):

        """See `IRunnableJob`."""

        try:

            self.attemptCopy()

        except CannotCopy, e:

            logger = logging.getLogger()

            logger.info("Job:\n%s\nraised CannotCopy:\n%s" % (self, e))

            self.abort()  # Abort the txn.

            self.reportFailure(e)

            # If there is an associated PackageUpload we need to reject it,

            # else it will sit in ACCEPTED forever.

            self._rejectPackageUpload()

            # Rely on the job runner to do the final commit.  Note that

            # we're not raising any exceptions here, failure of a copy is

            # not a failure of the job.

        except SuspendJobException:

            raise

        except:

            # Abort work done so far, but make sure that we commit the

            # rejection to the PackageUpload.

            transaction.abort()

            self._rejectPackageUpload()

            transaction.commit()

            raise

    def attemptCopy(self):

        """Attempt to perform the copy.

        :raise CannotCopy: If the copy fails for a reason that the user

            can deal with.

        """

        if self.target_archive.is_ppa:

            if self.target_pocket != PackagePublishingPocket.RELEASE:

                raise CannotCopy(

                    "Destination pocket must be 'release' for a PPA.")

        name = self.package_name

        version = self.package_version

        source_package = self.source_archive.getPublishedSources(

            name=name, version=version, exact_match=True).first()

        if source_package is None:

            raise CannotCopy("Package %r %r not found." % (name, version))

        source_name = getUtility(ISourcePackageNameSet)[name]

        # If there's a PackageUpload associated with this job then this

        # job has just been released by an archive admin from the queue.

        # We don't need to check any policies, but the admin may have

        # set overrides which we will get from the job's metadata.

        pu = getUtility(IPackageUploadSet).getByPackageCopyJobIDs(

            [self.context.id]).any()

        if pu is None:

            self._checkPolicies(

                source_name, source_package.sourcepackagerelease.component)

        # The package is free to go right in, so just copy it now.

        ancestry = self.target_archive.getPublishedSources(

            name=name, distroseries=self.target_distroseries,

            pocket=self.target_pocket, exact_match=True).first()

        override = self.getSourceOverride()

        copy_policy = self.getPolicyImplementation()

        send_email = copy_policy.send_email(self.target_archive)

        copied_sources = do_copy(

            sources=[source_package], archive=self.target_archive,

            series=self.target_distroseries, pocket=self.target_pocket,

            include_binaries=self.include_binaries, check_permissions=True,

            person=self.requester, overrides=[override],

            send_email=send_email, announce_from_person=self.requester,

            sponsored=self.sponsored)

        # Add a PackageDiff for this new upload if it has ancestry.

        if ancestry is not None:

            to_sourcepackagerelease = ancestry.sourcepackagerelease

            copied_source = copied_sources[0]

            try:

                to_sourcepackagerelease.requestDiffTo(

                    self.requester, copied_source.sourcepackagerelease)

            except PackageDiffAlreadyRequested:

                pass

        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 it gets moved to DONE.

            pu.setDone()

    def abort(self):

        """Abort work."""

        transaction.abort()

    def findMatchingDSDs(self):

        """Find any `DistroSeriesDifference`s that this job might resolve."""

        dsd_source = getUtility(IDistroSeriesDifferenceSource)

        target_series = self.target_distroseries

        candidates = dsd_source.getForDistroSeries(

            distro_series=target_series, name_filter=self.package_name,

            status=DistroSeriesDifferenceStatus.NEEDS_ATTENTION)

        # The job doesn't know what distroseries a given package is

        # coming from, and the version number in the DSD may have

        # changed.  We can however filter out DSDs that are from

        # different distributions, based on the job's target archive.

        source_distro_id = self.source_archive.distributionID

        return [

            dsd

            for dsd in candidates

                if dsd.parent_series.distributionID == source_distro_id]

    def reportFailure(self, cannotcopy_exception):

        """Attempt to report failure to the user."""

        message = unicode(cannotcopy_exception)

        dsds = self.findMatchingDSDs()

        comment_source = getUtility(IDistroSeriesDifferenceCommentSource)

        # Register the error comment in the name of the Janitor.  Not a

        # great choice, but we have no user identity to represent

        # Launchpad; it's far too costly to create one; and

        # impersonating the requester can be misleading and would also

        # involve extra bookkeeping.

        reporting_persona = getUtility(ILaunchpadCelebrities).janitor

        for dsd in dsds:

            comment_source.new(dsd, reporting_persona, message)

    def __repr__(self):

        """Returns an informative representation of the job."""

        parts = ["%s to copy" % self.__class__.__name__]

        if self.package_name is None:

            parts.append(" no package (!)")

        else:

            parts.append(" package %s" % self.package_name)

        parts.append(

            " from %s/%s" % (

                self.source_archive.distribution.name,

                self.source_archive.name))

        parts.append(

            " to %s/%s" % (

                self.target_archive.distribution.name,

                self.target_archive.name))

        parts.append(

            ", %s pocket," % self.target_pocket.name)

        if self.target_distroseries is not None:

            parts.append(" in %s" % self.target_distroseries)

        if self.include_binaries:

            parts.append(", including binaries")

        return "<%s>" % "".join(parts)

    def getPolicyImplementation(self):

        """Return the `ICopyPolicy` applicable to this job."""

        return ICopyPolicy(self.copy_policy)

PackageCopyJob.registerConcreteClass(PlainPackageCopyJob)