~launchpad-pqm/launchpad/devel

# Copyright 2009 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

"""Branch targets."""

__metaclass__ = type
__all__ = [
    'branch_to_target',
    'PackageBranchTarget',
    'PersonBranchTarget',
    'ProductBranchTarget',
    ]

from operator import attrgetter

from zope.component import getUtility
from zope.interface import implements
from zope.security.proxy import isinstance as zope_isinstance

from lp.code.errors import NoLinkedBranch
from lp.code.interfaces.branchcollection import IAllBranches
from lp.code.interfaces.branchtarget import (
    check_default_stacked_on,
    IBranchTarget,
    )
from lp.code.interfaces.codeimport import ICodeImportSet
from lp.code.interfaces.linkedbranch import get_linked_to_branch
from lp.registry.interfaces.pocket import PackagePublishingPocket
from lp.registry.interfaces.series import SeriesStatus
from lp.services.webapp.authorization import check_permission
from lp.services.webapp.interfaces import ICanonicalUrlData
from lp.services.webapp.sorting import sorted_version_numbers


def branch_to_target(branch):
    """Adapt an IBranch to an IBranchTarget."""
    return branch.target


class _BaseBranchTarget:

    def __eq__(self, other):
        return self.context == other.context

    def __ne__(self, other):
        return self.context != other.context

    def newCodeImport(self, registrant, branch_name, rcs_type, url=None,
                      cvs_root=None, cvs_module=None, owner=None):
        """See `IBranchTarget`."""
        return getUtility(ICodeImportSet).new(
            registrant, self, branch_name, rcs_type, url=url,
            cvs_root=cvs_root, cvs_module=cvs_module, owner=owner)

    def getRelatedSeriesBranchInfo(self, parent_branch, limit_results=None):
        """See `IBranchTarget`."""
        return []

    def getRelatedPackageBranchInfo(self, parent_branch, limit_results=None):
        """See `IBranchTarget`."""
        return []


class PackageBranchTarget(_BaseBranchTarget):
    implements(IBranchTarget)

    def __init__(self, sourcepackage):
        self.sourcepackage = sourcepackage

    @property
    def name(self):
        """See `IBranchTarget`."""
        return self.sourcepackage.path

    @property
    def components(self):
        """See `IBranchTarget`."""
        return [
            self.sourcepackage.distribution,
            self.sourcepackage.distroseries,
            self.sourcepackage,
            ]

    @property
    def context(self):
        """See `IBranchTarget`."""
        return self.sourcepackage

    def getNamespace(self, owner):
        """See `IBranchTarget`."""
        from lp.code.model.branchnamespace import (
            PackageNamespace)
        return PackageNamespace(owner, self.sourcepackage)

    @property
    def collection(self):
        """See `IBranchTarget`."""
        return getUtility(IAllBranches).inSourcePackage(self.sourcepackage)

    @property
    def default_stacked_on_branch(self):
        """See `IBranchTarget`."""
        return check_default_stacked_on(
            self.sourcepackage.development_version.getBranch(
                PackagePublishingPocket.RELEASE))

    @property
    def default_merge_target(self):
        """See `IBranchTarget`."""
        return self.sourcepackage.getBranch(PackagePublishingPocket.RELEASE)

    @property
    def displayname(self):
        """See `IBranchTarget`."""
        return self.sourcepackage.displayname

    @property
    def supports_merge_proposals(self):
        """See `IBranchTarget`."""
        return True

    @property
    def supports_short_identites(self):
        """See `IBranchTarget`."""
        return True

    @property
    def supports_code_imports(self):
        """See `IBranchTarget`."""
        return True

    def areBranchesMergeable(self, other_target):
        """See `IBranchTarget`."""
        # Branches are mergable into a PackageTarget if the source package
        # name is the same, or the branch is associated with the linked
        # product.
        if zope_isinstance(other_target, PackageBranchTarget):
            my_sourcepackagename = self.context.sourcepackagename
            other_sourcepackagename = other_target.context.sourcepackagename
            return my_sourcepackagename == other_sourcepackagename
        elif zope_isinstance(other_target, ProductBranchTarget):
            # If the sourcepackage has a related product, then branches of
            # that product are mergeable.
            product_series = self.sourcepackage.productseries
            if product_series is None:
                return False
            else:
                return other_target.context == product_series.product
        else:
            return False

    def assignKarma(self, person, action_name, date_created=None):
        """See `IBranchTarget`."""
        return person.assignKarma(
            action_name,
            distribution=self.context.distribution,
            sourcepackagename=self.context.sourcepackagename,
            datecreated=date_created)

    def getBugTask(self, bug):
        """See `IBranchTarget`."""
        # XXX: rockstar - See bug 397251.  Basically, source packages may have
        # specific bug tasks.  This should return those specific bugtasks in
        # those cases.
        return bug.default_bugtask

    def _retargetBranch(self, branch):
        """Set the branch target to refer to this target.

        This only updates the target related attributes of the branch, and
        expects a branch without a security proxy as a parameter.
        """
        branch.product = None
        branch.distroseries = self.sourcepackage.distroseries
        branch.sourcepackagename = self.sourcepackage.sourcepackagename

    def getRelatedPackageBranchInfo(self, parent_branch, limit_results=None):
        """See `IBranchTarget`."""
        result = []
        linked_branches = self.sourcepackage.linkedBranches()
        distroseries = self.sourcepackage.distroseries
        result.extend(
                [(branch, distroseries)
                 for branch in linked_branches.values()
                 if (check_permission('launchpad.View', branch) and
                    branch != parent_branch and
                    distroseries.status != SeriesStatus.OBSOLETE)])

        result = sorted_version_numbers(result, key=lambda branch_info: (
                    getattr(branch_info[1], 'name')))

        if limit_results is not None:
            # We only want the most recent branches
            result = result[:limit_results]
        return result


class PersonBranchTarget(_BaseBranchTarget):
    implements(IBranchTarget)

    name = u'+junk'
    default_stacked_on_branch = None
    default_merge_target = None

    def __init__(self, person):
        self.person = person

    @property
    def components(self):
        """See `IBranchTarget`."""
        return [self.person]

    @property
    def context(self):
        """See `IBranchTarget`."""
        return self.person

    @property
    def displayname(self):
        """See `IBranchTarget`."""
        return "~%s/+junk" % self.person.name

    def getNamespace(self, owner):
        """See `IBranchTarget`."""
        from lp.code.model.branchnamespace import (
            PersonalNamespace)
        return PersonalNamespace(owner)

    @property
    def collection(self):
        """See `IBranchTarget`."""
        return getUtility(IAllBranches).ownedBy(self.person).isJunk()

    @property
    def supports_merge_proposals(self):
        """See `IBranchTarget`."""
        return False

    @property
    def supports_short_identites(self):
        """See `IBranchTarget`."""
        return False

    @property
    def supports_code_imports(self):
        """See `IBranchTarget`."""
        return False

    def areBranchesMergeable(self, other_target):
        """See `IBranchTarget`."""
        return False

    def assignKarma(self, person, action_name, date_created=None):
        """See `IBranchTarget`."""
        # Does nothing. No karma for +junk.
        return None

    def getBugTask(self, bug):
        """See `IBranchTarget`."""
        return bug.default_bugtask

    def _retargetBranch(self, branch):
        """Set the branch target to refer to this target.

        This only updates the target related attributes of the branch, and
        expects a branch without a security proxy as a parameter.
        """
        branch.product = None
        branch.distroseries = None
        branch.sourcepackagename = None


class ProductBranchTarget(_BaseBranchTarget):
    implements(IBranchTarget)

    def __init__(self, product):
        self.product = product

    @property
    def components(self):
        """See `IBranchTarget`."""
        return [self.product]

    @property
    def context(self):
        """See `IBranchTarget`."""
        return self.product

    @property
    def displayname(self):
        """See `IBranchTarget`."""
        return self.product.displayname

    @property
    def name(self):
        """See `IBranchTarget`."""
        return self.product.name

    @property
    def default_stacked_on_branch(self):
        """See `IBranchTarget`."""
        return check_default_stacked_on(self.product.development_focus.branch)

    @property
    def default_merge_target(self):
        """See `IBranchTarget`."""
        return self.product.development_focus.branch

    def getNamespace(self, owner):
        """See `IBranchTarget`."""
        from lp.code.model.branchnamespace import (
            ProductNamespace)
        return ProductNamespace(owner, self.product)

    @property
    def collection(self):
        """See `IBranchTarget`."""
        return getUtility(IAllBranches).inProduct(self.product)

    @property
    def supports_merge_proposals(self):
        """See `IBranchTarget`."""
        return True

    @property
    def supports_short_identites(self):
        """See `IBranchTarget`."""
        return True

    @property
    def supports_code_imports(self):
        """See `IBranchTarget`."""
        return True

    def areBranchesMergeable(self, other_target):
        """See `IBranchTarget`."""
        # Branches are mergable into a PackageTarget if the source package
        # name is the same, or the branch is associated with the linked
        # product.
        if zope_isinstance(other_target, ProductBranchTarget):
            return self.product == other_target.context
        elif zope_isinstance(other_target, PackageBranchTarget):
            # If the sourcepackage has a related product, and that product is
            # the same as ours, then the branches are mergeable.
            product_series = other_target.context.productseries
            if product_series is None:
                return False
            else:
                return self.product == product_series.product
        else:
            return False

    def assignKarma(self, person, action_name, date_created=None):
        """See `IBranchTarget`."""
        return person.assignKarma(
            action_name, product=self.product, datecreated=date_created)

    def getBugTask(self, bug):
        """See `IBranchTarget`."""
        task = bug.getBugTask(self.product)
        if task is None:
            # Just choose the first task for the bug.
            task = bug.bugtasks[0]
        return task

    def _retargetBranch(self, branch):
        """Set the branch target to refer to this target.

        This only updates the target related attributes of the branch, and
        expects a branch without a security proxy as a parameter.
        """
        branch.product = self.product
        branch.distroseries = None
        branch.sourcepackagename = None

    def getRelatedSeriesBranchInfo(self, parent_branch, limit_results=None):
        """See `IBranchTarget`."""
        sorted_series = []
        for series in self.product.series:
            if (series.status != SeriesStatus.OBSOLETE
                and series != self.product.development_focus):
                sorted_series.append(series)
        # Now sort the list by name with newer versions before older.
        sorted_series = sorted_version_numbers(sorted_series,
                                             key=attrgetter('name'))
        # Add the development focus first.
        sorted_series.insert(0, self.product.development_focus)

        result = []
        for series in sorted_series:
            try:
                branch = get_linked_to_branch(series).branch
                if (branch not in result and branch != parent_branch and
                    check_permission('launchpad.View', branch)):
                    result.append((branch, series))
            except NoLinkedBranch:
                # If there's no branch for a particular series, we don't care.
                pass

        if limit_results is not None:
            # We only want the most recent branches
            result = result[:limit_results]
        return result

    def getRelatedPackageBranchInfo(self, parent_branch, limit_results=None):
        """See `IBranchTarget`."""
        result = []
        for distrosourcepackage in self.product.distrosourcepackages:
            try:
                branch = get_linked_to_branch(distrosourcepackage).branch
                series = distrosourcepackage.distribution.currentseries
                if (branch != parent_branch and series is not None and
                    check_permission('launchpad.View', branch)):
                        result.append((branch, series))
            except NoLinkedBranch:
                # If there's no branch for a particular source package,
                # we don't care.
                pass

        result = sorted_version_numbers(result, key=lambda branch_info: (
                    getattr(branch_info[1], 'name')))

        if limit_results is not None:
            # We only want the most recent branches
            result = result[:limit_results]
        return result


def get_canonical_url_data_for_target(branch_target):
    """Return the `ICanonicalUrlData` for an `IBranchTarget`."""
    return ICanonicalUrlData(branch_target.context)


def product_series_to_branch_target(product_series):
    """The Product itself is the branch target given a ProductSeries."""
    return ProductBranchTarget(product_series.product)


def distribution_sourcepackage_to_branch_target(distro_sourcepackage):
    """The development version of the distro sourcepackage is the target."""
    dev_version = distro_sourcepackage.development_version
    # It is possible for distributions to not have any series, and if that is
    # the case, the dev_version is None.
    if dev_version is None:
        return None
    return PackageBranchTarget(dev_version)