~launchpad-pqm/launchpad/devel

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

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

# pylint: disable-msg=E0611,W0212,W0141,F0401

__metaclass__ = type

__all__ = [

    'Branch',

    'BranchSet',

    ]

from datetime import datetime

import operator

from bzrlib import urlutils

from bzrlib.revision import NULL_REVISION

import pytz

import simplejson

from sqlobject import (

    BoolCol,

    ForeignKey,

    IntCol,

    SQLMultipleJoin,

    SQLRelatedJoin,

    StringCol,

    )

from storm.expr import (

    And,

    Count,

    Desc,

    NamedFunc,

    Not,

    Or,

    Select,

    )

from storm.locals import (

    AutoReload,

    Int,

    Reference,

    )

from storm.store import Store

from zope.component import getUtility

from zope.event import notify

from zope.interface import implements

from zope.security.proxy import (

    ProxyFactory,

    removeSecurityProxy,

    )

from canonical.config import config

from canonical.database.constants import (

    DEFAULT,

    UTC_NOW,

    )

from canonical.database.datetimecol import UtcDateTimeCol

from canonical.database.enumcol import EnumCol

from canonical.database.sqlbase import (

    SQLBase,

    sqlvalues,

    )

from canonical.launchpad import _

from canonical.launchpad.components.decoratedresultset import (

    DecoratedResultSet,

    )

from canonical.launchpad.helpers import shortlist

from canonical.launchpad.interfaces.launchpad import (

    ILaunchpadCelebrities,

    IPrivacy,

    )

from canonical.launchpad.interfaces.lpstorm import IMasterStore

from canonical.launchpad.webapp import urlappend

from lp.app.errors import UserCannotUnsubscribePerson

from lp.bugs.interfaces.bugtask import (

    BugTaskSearchParams,

    IBugTaskSet,

    )

from lp.bugs.interfaces.bugtaskfilter import filter_bugtasks_by_context

from lp.buildmaster.model.buildqueue import BuildQueue

from lp.code.bzr import (

    BranchFormat,

    ControlFormat,

    CURRENT_BRANCH_FORMATS,

    CURRENT_REPOSITORY_FORMATS,

    RepositoryFormat,

    )

from lp.code.enums import (

    BranchLifecycleStatus,

    BranchMergeProposalStatus,

    BranchType,

    )

from lp.code.errors import (

    BranchCannotBePrivate,

    BranchCannotBePublic,

    BranchMergeProposalExists,

    BranchTargetError,

    BranchTypeError,

    CannotDeleteBranch,

    InvalidBranchMergeProposal,

    InvalidMergeQueueConfig,

    )

from lp.code.event.branchmergeproposal import (

    BranchMergeProposalNeedsReviewEvent,

    NewBranchMergeProposalEvent,

    )

from lp.code.interfaces.branch import (

    BzrIdentityMixin,

    DEFAULT_BRANCH_STATUS_IN_LISTING,

    IBranch,

    IBranchNavigationMenu,

    IBranchSet,

    user_has_special_branch_access,

    )

from lp.code.interfaces.branchcollection import IAllBranches

from lp.code.interfaces.branchlookup import IBranchLookup

from lp.code.interfaces.branchmergeproposal import (

    BRANCH_MERGE_PROPOSAL_FINAL_STATES,

    )

from lp.code.interfaces.branchnamespace import IBranchNamespacePolicy

from lp.code.interfaces.branchpuller import IBranchPuller

from lp.code.interfaces.branchtarget import IBranchTarget

from lp.code.interfaces.codehosting import (

    BRANCH_ID_ALIAS_PREFIX,

    compose_public_url,

    )

from lp.code.interfaces.seriessourcepackagebranch import (

    IFindOfficialBranchLinks,

    )

from lp.code.mail.branch import send_branch_modified_notifications

from lp.code.model.branchmergeproposal import (

    BranchMergeProposal,

    BranchMergeProposalGetter,

    )

from lp.code.model.branchrevision import BranchRevision

from lp.code.model.branchsubscription import BranchSubscription

from lp.code.model.revision import (

    Revision,

    RevisionAuthor,

    )

from lp.code.model.seriessourcepackagebranch import SeriesSourcePackageBranch

from lp.codehosting.bzrutils import safe_open

from lp.registry.interfaces.person import (

    validate_person,

    validate_public_person,

    )

from lp.services.database.bulk import load_related

from lp.services.job.interfaces.job import JobStatus

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

from lp.services.mail.notificationrecipientset import NotificationRecipientSet

from lp.services.propertycache import cachedproperty

class Branch(SQLBase, BzrIdentityMixin):

    """A sequence of ordered revisions in Bazaar."""

    implements(IBranch, IBranchNavigationMenu, IPrivacy)

    _table = 'Branch'

    branch_type = EnumCol(enum=BranchType, notNull=True)

    name = StringCol(notNull=False)

    url = StringCol(dbName='url')

    description = StringCol(dbName='summary')

    branch_format = EnumCol(enum=BranchFormat)

    repository_format = EnumCol(enum=RepositoryFormat)

    # XXX: Aaron Bentley 2008-06-13

    # Rename the metadir_format in the database, see bug 239746

    control_format = EnumCol(enum=ControlFormat, dbName='metadir_format')

    whiteboard = StringCol(default=None)

    mirror_status_message = StringCol(default=None)

    private = BoolCol(default=False, notNull=True)

    def setPrivate(self, private, user):

        """See `IBranch`."""

        if private == self.private:

            return

        # Only check the privacy policy if the user is not special.

        if (not user_has_special_branch_access(user)):

            policy = IBranchNamespacePolicy(self.namespace)

            if private and not policy.canBranchesBePrivate():

                raise BranchCannotBePrivate()

            if not private and not policy.canBranchesBePublic():

                raise BranchCannotBePublic()

        self.private = private

    registrant = ForeignKey(

        dbName='registrant', foreignKey='Person',

        storm_validator=validate_public_person, notNull=True)

    owner = ForeignKey(

        dbName='owner', foreignKey='Person',

        storm_validator=validate_person, notNull=True)

    def setOwner(self, new_owner, user):

        """See `IBranch`."""

        new_namespace = self.target.getNamespace(new_owner)

        new_namespace.moveBranch(self, user, rename_if_necessary=True)

    reviewer = ForeignKey(

        dbName='reviewer', foreignKey='Person',

        storm_validator=validate_public_person, default=None)

    product = ForeignKey(dbName='product', foreignKey='Product', default=None)

    distroseries = ForeignKey(

        dbName='distroseries', foreignKey='DistroSeries', default=None)

    sourcepackagename = ForeignKey(

        dbName='sourcepackagename', foreignKey='SourcePackageName',

        default=None)

    lifecycle_status = EnumCol(

        enum=BranchLifecycleStatus, notNull=True,

        default=BranchLifecycleStatus.DEVELOPMENT)

    last_mirrored = UtcDateTimeCol(default=None)

    last_mirrored_id = StringCol(default=None)

    last_mirror_attempt = UtcDateTimeCol(default=None)

    mirror_failures = IntCol(default=0, notNull=True)

    next_mirror_time = UtcDateTimeCol(default=None)

    last_scanned = UtcDateTimeCol(default=None)

    last_scanned_id = StringCol(default=None)

    revision_count = IntCol(default=DEFAULT, notNull=True)

    stacked_on = ForeignKey(

        dbName='stacked_on', foreignKey='Branch', default=None)

    # The unique_name is maintined by a SQL trigger.

    unique_name = StringCol()

    # Denormalised colums used primarily for sorting.

    owner_name = StringCol()

    target_suffix = StringCol()

    def __repr__(self):

        return '<Branch %r (%d)>' % (self.unique_name, self.id)

    @property

    def target(self):

        """See `IBranch`."""

        if self.product is None:

            if self.distroseries is None:

                target = self.owner

            else:

                target = self.sourcepackage

        else:

            target = self.product

        return IBranchTarget(target)

    def setTarget(self, user, project=None, source_package=None):

        """See `IBranch`."""

        if project is not None:

            if source_package is not None:

                raise BranchTargetError(

                    'Cannot specify both a project and a source package.')

            else:

                target = IBranchTarget(project)

                if target is None:

                    raise BranchTargetError(

                        '%r is not a valid project target' % project)

        elif source_package is not None:

            target = IBranchTarget(source_package)

            if target is None:

                raise BranchTargetError(

                    '%r is not a valid source package target' %

                    source_package)

        else:

            target = IBranchTarget(self.owner)

            # Person targets are always valid.

        namespace = target.getNamespace(self.owner)

        namespace.moveBranch(self, user, rename_if_necessary=True)

    @property

    def namespace(self):

        """See `IBranch`."""

        return self.target.getNamespace(self.owner)

    @property

    def distribution(self):

        """See `IBranch`."""

        if self.distroseries is None:

            return None

        return self.distroseries.distribution

    @property

    def sourcepackage(self):

        """See `IBranch`."""

        # Avoid circular imports.

        from lp.registry.model.sourcepackage import SourcePackage

        if self.distroseries is None:

            return None

        return SourcePackage(self.sourcepackagename, self.distroseries)

    @property

    def revision_history(self):

        result = Store.of(self).find(

            (BranchRevision, Revision),

            BranchRevision.branch_id == self.id,

            Revision.id == BranchRevision.revision_id,

            BranchRevision.sequence != None)

        result = result.order_by(Desc(BranchRevision.sequence))

        return DecoratedResultSet(result, operator.itemgetter(0))

    subscriptions = SQLMultipleJoin(

        'BranchSubscription', joinColumn='branch', orderBy='id')

    subscribers = SQLRelatedJoin(

        'Person', joinColumn='branch', otherColumn='person',

        intermediateTable='BranchSubscription', orderBy='name')

    bug_branches = SQLMultipleJoin(

        'BugBranch', joinColumn='branch', orderBy='id')

    linked_bugs = SQLRelatedJoin(

        'Bug', joinColumn='branch', otherColumn='bug',

        intermediateTable='BugBranch', orderBy='id')

    def getLinkedBugTasks(self, user, status_filter=None):

        """See `IBranch`."""

        params = BugTaskSearchParams(user=user, linked_branches=self.id,

            status=status_filter)

        tasks = shortlist(getUtility(IBugTaskSet).search(params), 1000)

        # Post process to discard irrelevant tasks: we only return one task per

        # bug, and cannot easily express this in sql (yet).

        return filter_bugtasks_by_context(self.target.context, tasks)

    def linkBug(self, bug, registrant):

        """See `IBranch`."""

        return bug.linkBranch(self, registrant)

    def unlinkBug(self, bug, user):

        """See `IBranch`."""

        return bug.unlinkBranch(self, user)

    spec_links = SQLMultipleJoin('SpecificationBranch',

        joinColumn='branch',

        orderBy='id')

    def linkSpecification(self, spec, registrant):

        """See `IBranch`."""

        return spec.linkBranch(self, registrant)

    def unlinkSpecification(self, spec, user):

        """See `IBranch`."""

        return spec.unlinkBranch(self, user)

    date_created = UtcDateTimeCol(notNull=True, default=DEFAULT)

    date_last_modified = UtcDateTimeCol(notNull=True, default=DEFAULT)

    landing_targets = SQLMultipleJoin(

        'BranchMergeProposal', joinColumn='source_branch')

    @property

    def active_landing_targets(self):

        """Merge proposals not in final states where this branch is source."""

        store = Store.of(self)

        return store.find(

            BranchMergeProposal, BranchMergeProposal.source_branch == self,

            Not(BranchMergeProposal.queue_status.is_in(

                BRANCH_MERGE_PROPOSAL_FINAL_STATES)))

    @property

    def landing_candidates(self):

        """See `IBranch`."""

        return BranchMergeProposal.select("""

            BranchMergeProposal.target_branch = %s AND

            BranchMergeProposal.queue_status NOT IN %s

            """ % sqlvalues(self, BRANCH_MERGE_PROPOSAL_FINAL_STATES))

    @property

    def dependent_branches(self):

        """See `IBranch`."""

        return BranchMergeProposal.select("""

            BranchMergeProposal.dependent_branch = %s AND

            BranchMergeProposal.queue_status NOT IN %s

            """ % sqlvalues(self, BRANCH_MERGE_PROPOSAL_FINAL_STATES))

    def getMergeProposals(self, status=None, visible_by_user=None,

                          merged_revnos=None):

        """See `IBranch`."""

        if not status:

            status = (

                BranchMergeProposalStatus.CODE_APPROVED,

                BranchMergeProposalStatus.NEEDS_REVIEW,

                BranchMergeProposalStatus.WORK_IN_PROGRESS)

        collection = getUtility(IAllBranches).visibleByUser(visible_by_user)

        return collection.getMergeProposals(

            status, target_branch=self, merged_revnos=merged_revnos)

    def isBranchMergeable(self, target_branch):

        """See `IBranch`."""

        # In some imaginary time we may actually check to see if this branch

        # and the target branch have common ancestry.

        return self.target.areBranchesMergeable(target_branch.target)

    def addLandingTarget(self, registrant, target_branch,

                         prerequisite_branch=None, whiteboard=None,

                         date_created=None, needs_review=False,

                         description=None, review_requests=None,

                         review_diff=None, commit_message=None):

        """See `IBranch`."""

        if not self.target.supports_merge_proposals:

            raise InvalidBranchMergeProposal(

                '%s branches do not support merge proposals.'

                % self.target.displayname)

        if self == target_branch:

            raise InvalidBranchMergeProposal(

                'Source and target branches must be different.')

        if not target_branch.isBranchMergeable(self):

            raise InvalidBranchMergeProposal(

                '%s is not mergeable into %s' % (

                    self.displayname, target_branch.displayname))

        if prerequisite_branch is not None:

            if not self.isBranchMergeable(prerequisite_branch):

                raise InvalidBranchMergeProposal(

                    '%s is not mergeable into %s' % (

                        prerequisite_branch.displayname, self.displayname))

            if self == prerequisite_branch:

                raise InvalidBranchMergeProposal(

                    'Source and prerequisite branches must be different.')

            if target_branch == prerequisite_branch:

                raise InvalidBranchMergeProposal(

                    'Target and prerequisite branches must be different.')

        target = BranchMergeProposalGetter.activeProposalsForBranches(

            self, target_branch)

        if target.count() > 0:

            raise BranchMergeProposalExists(

                'There is already a branch merge proposal registered for '

                'branch %s to land on %s that is still active.'

                % (self.displayname, target_branch.displayname))

        if date_created is None:

            date_created = UTC_NOW

        if needs_review:

            queue_status = BranchMergeProposalStatus.NEEDS_REVIEW

            date_review_requested = date_created

        else:

            queue_status = BranchMergeProposalStatus.WORK_IN_PROGRESS

            date_review_requested = None

        if review_requests is None:

            review_requests = []

        # If no reviewer is specified, use the default for the branch.

        if len(review_requests) == 0:

            review_requests.append((target_branch.code_reviewer, None))

        bmp = BranchMergeProposal(

            registrant=registrant, source_branch=self,

            target_branch=target_branch,

            prerequisite_branch=prerequisite_branch, whiteboard=whiteboard,

            date_created=date_created,

            date_review_requested=date_review_requested,

            queue_status=queue_status, review_diff=review_diff,

            commit_message=commit_message,

            description=description)

        for reviewer, review_type in review_requests:

            bmp.nominateReviewer(

                reviewer, registrant, review_type, _notify_listeners=False)

        notify(NewBranchMergeProposalEvent(bmp))

        if needs_review:

            notify(BranchMergeProposalNeedsReviewEvent(bmp))

        return bmp

    def _createMergeProposal(

        self, registrant, target_branch, prerequisite_branch=None,

        needs_review=True, initial_comment=None, commit_message=None,

        reviewers=None, review_types=None):

        """See `IBranch`."""

        if reviewers is None:

            reviewers = []

        if review_types is None:

            review_types = []

        if len(reviewers) != len(review_types):

            raise ValueError(

                'reviewers and review_types must be equal length.')

        review_requests = zip(reviewers, review_types)

        return self.addLandingTarget(

            registrant, target_branch, prerequisite_branch,

            needs_review=needs_review, description=initial_comment,

            commit_message=commit_message, review_requests=review_requests)

    def scheduleDiffUpdates(self):

        """See `IBranch`."""

        from lp.code.model.branchmergeproposaljob import (

                GenerateIncrementalDiffJob,

                UpdatePreviewDiffJob,

            )

        jobs = []

        for merge_proposal in self.active_landing_targets:

            if merge_proposal.target_branch.last_scanned_id is None:

                continue

            jobs.append(UpdatePreviewDiffJob.create(merge_proposal))

            for old, new in merge_proposal.getMissingIncrementalDiffs():

                GenerateIncrementalDiffJob.create(

                    merge_proposal, old.revision_id, new.revision_id)

        return jobs

    def addToLaunchBag(self, launchbag):

        """See `IBranch`."""

        launchbag.add(self.product)

        if self.distroseries is not None:

            launchbag.add(self.distroseries)

            launchbag.add(self.distribution)

            launchbag.add(self.sourcepackage)

    def getStackedBranches(self):

        """See `IBranch`."""

        store = Store.of(self)

        return store.find(Branch, Branch.stacked_on == self)

    @property

    def code_is_browseable(self):

        """See `IBranch`."""

        return (self.revision_count > 0 or self.last_mirrored != None)

    def codebrowse_url(self, *extras):

        """See `IBranch`."""

        if self.private:

            root = config.codehosting.secure_codebrowse_root

        else:

            root = config.codehosting.codebrowse_root

        return urlutils.join(root, self.unique_name, *extras)

    @property

    def browse_source_url(self):

        return self.codebrowse_url('files')

    def composePublicURL(self, scheme='http'):

        """See `IBranch`."""

        # Not all protocols work for private branches.

        public_schemes = ['http']

        assert not (self.private and scheme in public_schemes), (

            "Private branch %s has no public URL." % self.unique_name)

        return compose_public_url(scheme, self.unique_name)

    def getInternalBzrUrl(self):

        """See `IBranch`."""

        return 'lp-internal:///' + self.unique_name

    def getBzrBranch(self):

        """See `IBranch`."""

        return safe_open('lp-internal', self.getInternalBzrUrl())

    @property

    def displayname(self):

        """See `IBranch`."""

        return self.bzr_identity

    @property

    def code_reviewer(self):

        """See `IBranch`."""

        if self.reviewer:

            return self.reviewer

        else:

            return self.owner

    def isPersonTrustedReviewer(self, reviewer):

        """See `IBranch`."""

        if reviewer is None:

            return False

        # We trust Launchpad admins.

        lp_admins = getUtility(ILaunchpadCelebrities).admin

        # Both the branch owner and the review team are checked.

        owner = self.owner

        review_team = self.code_reviewer

        return (

            reviewer.inTeam(owner) or

            reviewer.inTeam(review_team) or

            reviewer.inTeam(lp_admins))

    def latest_revisions(self, quantity=10):

        """See `IBranch`."""

        return self.revision_history.config(limit=quantity)

    def getMainlineBranchRevisions(self, start_date, end_date=None,

                                   oldest_first=False):

        """See `IBranch`."""

        date_clause = Revision.revision_date >= start_date

        if end_date is not None:

            date_clause = And(date_clause, Revision.revision_date <= end_date)

        result = Store.of(self).find(

            (BranchRevision, Revision),

            BranchRevision.branch == self,

            BranchRevision.sequence != None,

            BranchRevision.revision == Revision.id,

            date_clause)

        if oldest_first:

            result = result.order_by(BranchRevision.sequence)

        else:

            result = result.order_by(Desc(BranchRevision.sequence))

        def eager_load(rows):

            revisions = map(operator.itemgetter(1), rows)

            load_related(RevisionAuthor, revisions, ['revision_author_id'])

        return DecoratedResultSet(result, pre_iter_hook=eager_load)

    def getRevisionsSince(self, timestamp):

        """See `IBranch`."""

        result = Store.of(self).find(

            (BranchRevision, Revision),

            Revision.id == BranchRevision.revision_id,

            BranchRevision.branch == self,

            BranchRevision.sequence != None,

            Revision.revision_date > timestamp)

        result = result.order_by(Desc(BranchRevision.sequence))

        # Return BranchRevision but prejoin Revision as well.

        return DecoratedResultSet(result, operator.itemgetter(0))

    def canBeDeleted(self):

        """See `IBranch`."""

        if ((len(self.deletionRequirements()) != 0) or

            self.getStackedBranches().count() > 0):

            # Can't delete if the branch is associated with anything.

            return False

        else:

            return True

    @cachedproperty

    def code_import(self):

        from lp.code.model.codeimport import CodeImportSet

        return CodeImportSet().getByBranch(self)

    def _deletionRequirements(self):

        """Determine what operations must be performed to delete this branch.