~launchpad-pqm/launchpad/devel

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

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

"""Branch views."""

__metaclass__ = type

__all__ = [

    'BranchAddView',

    'BranchContextMenu',

    'BranchDeletionView',

    'BranchEditStatusView',

    'BranchEditView',

    'BranchEditWhiteboardView',

    'BranchRequestImportView',

    'BranchReviewerEditView',

    'BranchMergeQueueView',

    'BranchMirrorStatusView',

    'BranchMirrorMixin',

    'BranchNameValidationMixin',

    'BranchNavigation',

    'BranchEditMenu',

    'BranchInProductView',

    'BranchUpgradeView',

    'BranchURL',

    'BranchView',

    'BranchSubscriptionsView',

    'RegisterBranchMergeProposalView',

    'TryImportAgainView',

    ]

import cgi

from datetime import (

    datetime,

    timedelta,

    )

from lazr.enum import (

    EnumeratedType,

    Item,

    )

from lazr.lifecycle.event import ObjectModifiedEvent

from lazr.lifecycle.snapshot import Snapshot

from lazr.restful.interface import (

    copy_field,

    use_template,

    )

from lazr.restful.utils import smartquote

from lazr.uri import URI

import pytz

from zope.app.form import CustomWidgetFactory

from zope.app.form.browser import TextAreaWidget

from zope.app.form.browser.boolwidgets import CheckBoxWidget

from zope.component import (

    getUtility,

    queryAdapter,

    )

from zope.event import notify

from zope.formlib import form

from zope.interface import (

    implements,

    Interface,

    providedBy,

    )

from zope.publisher.interfaces import NotFound

from zope.schema import (

    Bool,

    Choice,

    Text,

    )

from zope.schema.vocabulary import (

    SimpleTerm,

    SimpleVocabulary,

    )

from zope.traversing.interfaces import IPathAdapter

from lp.services.config import config

from canonical.database.constants import UTC_NOW

from lp import _

from lp.services import searchbuilder

from lp.services.feeds.browser import (

    BranchFeedLink,

    FeedsMixin,

    )

from lp.services.helpers import truncate_text

from lp.services.webapp import (

    canonical_url,

    ContextMenu,

    enabled_with_permission,

    LaunchpadView,

    Link,

    Navigation,

    NavigationMenu,

    stepthrough,

    stepto,

    )

from lp.services.webapp.authorization import (

    check_permission,

    precache_permission_for_objects,

    )

from lp.services.webapp.interfaces import ICanonicalUrlData

from lp.services.webapp.menu import structured

from lp.app.browser.launchpad import Hierarchy

from lp.app.browser.launchpadform import (

    action,

    custom_widget,

    LaunchpadEditFormView,

    LaunchpadFormView,

    )

from lp.app.browser.lazrjs import EnumChoiceWidget

from lp.app.errors import NotFoundError

from lp.app.interfaces.launchpad import ILaunchpadCelebrities

from lp.app.widgets.itemswidgets import LaunchpadRadioWidgetWithDescription

from lp.app.widgets.suggestion import TargetBranchWidget

from lp.blueprints.interfaces.specificationbranch import ISpecificationBranch

from lp.bugs.interfaces.bug import IBugSet

from lp.bugs.interfaces.bugbranch import IBugBranch

from lp.bugs.interfaces.bugtask import UNRESOLVED_BUGTASK_STATUSES

from lp.code.browser.branchmergeproposal import (

    latest_proposals_for_each_branch,

    )

from lp.code.browser.branchref import BranchRef

from lp.code.browser.decorations import DecoratedBranch

from lp.code.browser.sourcepackagerecipelisting import HasRecipesMenuMixin

from lp.code.enums import (

    BranchType,

    CodeImportResultStatus,

    CodeImportReviewStatus,

    RevisionControlSystems,

    UICreatableBranchType,

    )

from lp.code.errors import (

    BranchCreationForbidden,

    BranchExists,

    CannotUpgradeBranch,

    CodeImportAlreadyRequested,

    CodeImportAlreadyRunning,

    CodeImportNotInReviewedState,

    InvalidBranchMergeProposal,

    )

from lp.code.interfaces.branch import (

    IBranch,

    user_has_special_branch_access,

    )

from lp.code.interfaces.branchcollection import IAllBranches

from lp.code.interfaces.branchmergeproposal import IBranchMergeProposal

from lp.code.interfaces.branchnamespace import IBranchNamespacePolicy

from lp.code.interfaces.branchtarget import IBranchTarget

from lp.code.interfaces.codereviewvote import ICodeReviewVoteReference

from lp.registry.interfaces.person import (

    IPerson,

    IPersonSet,

    )

from lp.registry.interfaces.productseries import IProductSeries

from lp.registry.vocabularies import UserTeamsParticipationPlusSelfVocabulary

from lp.services.propertycache import cachedproperty

from lp.translations.interfaces.translationtemplatesbuild import (

    ITranslationTemplatesBuildSource,

    )

def quote(text):

    return cgi.escape(text, quote=True)

class BranchURL:

    """Branch URL creation rules."""

    implements(ICanonicalUrlData)

    rootsite = 'code'

    inside = None

    def __init__(self, branch):

        self.branch = branch

    @property

    def path(self):

        return self.branch.unique_name

def branch_root_context(branch):

    """Return the IRootContext for the branch."""

    return branch.target.components[0]

class BranchHierarchy(Hierarchy):

    """The hierarchy for a branch should be the product if there is one."""

    @property

    def objects(self):

        """See `Hierarchy`."""

        traversed = list(self.request.traversed_objects)

        # Pass back the root object.

        yield traversed.pop(0)

        # Now pop until we find the branch.

        branch = traversed.pop(0)

        while not IBranch.providedBy(branch):

            branch = traversed.pop(0)

        # Now pass back the branch components.

        for component in branch.target.components:

            yield component

        # Now the branch.

        yield branch

        # Now whatever is left.

        for obj in traversed:

            yield obj

class BranchNavigation(Navigation):

    usedfor = IBranch

    @stepthrough("+bug")

    def traverse_bug_branch(self, bugid):

        """Traverses to an `IBugBranch`."""

        bug = getUtility(IBugSet).get(bugid)

        for bug_branch in bug.linked_branches:

            if bug_branch.branch == self.context:

                return bug_branch

    @stepto(".bzr")

    def dotbzr(self):

        return BranchRef(self.context)

    @stepthrough("+subscription")

    def traverse_subscription(self, name):

        """Traverses to an `IBranchSubcription`."""

        person = getUtility(IPersonSet).getByName(name)

        if person is not None:

            return self.context.getSubscription(person)

    @stepthrough("+merge")

    def traverse_merge_proposal(self, id):

        """Traverse to an `IBranchMergeProposal`."""

        try:

            id = int(id)

        except ValueError:

            # Not a number.

            return None

        for proposal in self.context.landing_targets:

            if proposal.id == id:

                return proposal

    @stepto("+code-import")

    def traverse_code_import(self):

        """Traverses to the `ICodeImport` for the branch."""

        return self.context.code_import

    @stepthrough("+translation-templates-build")

    def traverse_translation_templates_build(self, id_string):

        """Traverses to a `TranslationTemplatesBuild`."""

        try:

            ttbj_id = int(id_string)

        except ValueError:

            raise NotFoundError(id_string)

        source = getUtility(ITranslationTemplatesBuildSource)

        return source.getByID(ttbj_id)

class BranchEditMenu(NavigationMenu):

    """Edit menu for IBranch."""

    usedfor = IBranch

    facet = 'branches'

    title = 'Edit branch'

    links = (

        'edit', 'reviewer', 'edit_whiteboard', 'delete')

    def branch_is_import(self):

        return self.context.branch_type == BranchType.IMPORTED

    @enabled_with_permission('launchpad.Edit')

    def edit(self):

        text = 'Change branch details'

        return Link('+edit', text, icon='edit')

    @enabled_with_permission('launchpad.Edit')

    def delete(self):

        text = 'Delete branch'

        return Link('+delete', text, icon='trash-icon')

    @enabled_with_permission('launchpad.AnyPerson')

    def edit_whiteboard(self):

        text = 'Edit whiteboard'

        enabled = self.branch_is_import()

        return Link(

            '+whiteboard', text, icon='edit', enabled=enabled)

    @enabled_with_permission('launchpad.Edit')

    def reviewer(self):

        text = 'Set branch reviewer'

        return Link('+reviewer', text, icon='edit')

class BranchContextMenu(ContextMenu, HasRecipesMenuMixin):

    """Context menu for branches."""

    usedfor = IBranch

    facet = 'branches'

    links = [

        'add_subscriber', 'browse_revisions', 'create_recipe', 'link_bug',

        'link_blueprint', 'register_merge', 'source', 'subscription',

        'edit_status', 'edit_import', 'upgrade_branch', 'view_recipes',

        'create_queue']

    @enabled_with_permission('launchpad.Edit')

    def edit_status(self):

        text = 'Change branch status'

        return Link('+edit-status', text, icon='edit')

    def browse_revisions(self):

        """Return a link to the branch's revisions on codebrowse."""

        text = 'All revisions'

        enabled = self.context.code_is_browseable

        url = self.context.codebrowse_url('changes')

        return Link(url, text, enabled=enabled)

    @enabled_with_permission('launchpad.AnyPerson')

    def subscription(self):

        if self.context.hasSubscription(self.user):

            url = '+edit-subscription'

            text = 'Edit your subscription'

            icon = 'edit'

        else:

            url = '+subscribe'

            text = 'Subscribe yourself'

            icon = 'add'

        return Link(url, text, icon=icon)

    @enabled_with_permission('launchpad.AnyPerson')

    def add_subscriber(self):

        text = 'Subscribe someone else'

        return Link('+addsubscriber', text, icon='add')

    def register_merge(self):

        text = 'Propose for merging'

        enabled = (

            self.context.target.supports_merge_proposals and

            not self.context.branch_type == BranchType.IMPORTED)

        return Link('+register-merge', text, icon='add', enabled=enabled)

    def link_bug(self):

        text = 'Link a bug report'

        return Link('+linkbug', text, icon='add')

    def link_blueprint(self):

        if self.context.spec_links:

            text = 'Link to another blueprint'

        else:

            text = 'Link to a blueprint'

        # XXX: JonathanLange 2009-05-13 spec=package-branches: Actually,

        # distroseries can also have blueprints, so it makes sense to

        # associate package-branches with them.

        #

        # Since the blueprints are only related to products, there is no

        # point showing this link if the branch is junk.

        enabled = self.context.product is not None

        return Link('+linkblueprint', text, icon='add', enabled=enabled)

    def source(self):

        """Return a link to the branch's file listing on codebrowse."""

        text = 'Browse the code'

        enabled = self.context.code_is_browseable

        url = self.context.codebrowse_url('files')

        return Link(url, text, icon='info', enabled=enabled)

    def edit_import(self):

        text = 'Edit import source or review import'

        enabled = True

        enabled = (

            self.context.branch_type == BranchType.IMPORTED and

            check_permission('launchpad.Edit', self.context.code_import))

        return Link(

            '+edit-import', text, icon='edit', enabled=enabled)

    @enabled_with_permission('launchpad.Edit')

    def upgrade_branch(self):

        enabled = False

        if self.context.needs_upgrading:

            enabled = True

        return Link(

            '+upgrade', 'Upgrade this branch', icon='edit', enabled=enabled)

    def create_recipe(self):

        # You can't create a recipe for a private branch.

        enabled = not self.context.private

        text = 'Create packaging recipe'

        return Link('+new-recipe', text, enabled=enabled, icon='add')

    @enabled_with_permission('launchpad.Edit')

    def create_queue(self):

        return Link('+create-queue', 'Create a new queue', icon='add')

class BranchMirrorMixin:

    """Provide mirror_location property.

    Requires self.branch to be set by the class using this mixin.

    """

    @property

    def mirror_location(self):

        """Check the mirror location to see if it is a private one."""

        branch = self.branch

        # If the user has edit permissions, then show the actual location.

        if branch.url is None or check_permission('launchpad.Edit', branch):

            return branch.url

        # XXX: Tim Penhey, 2008-05-30, bug 235916

        # Instead of a configuration hack we should support the users

        # specifying whether or not they want the mirror location

        # hidden or not.  Given that this is a database patch,

        # it isn't going to happen today.

        hosts = config.codehosting.private_mirror_hosts.split(',')

        private_mirror_hosts = [name.strip() for name in hosts]

        uri = URI(branch.url)

        for private_host in private_mirror_hosts:

            if uri.underDomain(private_host):

                return '<private server>'

        return branch.url

class BranchView(LaunchpadView, FeedsMixin, BranchMirrorMixin):

    feed_types = (

        BranchFeedLink,

        )

    @property

    def page_title(self):

        return self.context.bzr_identity

    label = page_title

    def initialize(self):

        self.branch = self.context

        self.notices = []

        # Cache permission so private team owner can be rendered.

        # The security adaptor will do the job also but we don't want or need

        # the expense of running several complex SQL queries.

        authorised_people = [self.branch.owner]

        if self.user is not None:

            precache_permission_for_objects(

                self.request, "launchpad.LimitedView", authorised_people)

        # Replace our context with a decorated branch, if it is not already

        # decorated.

        if not isinstance(self.context, DecoratedBranch):

            self.context = DecoratedBranch(self.context)

    def user_is_subscribed(self):

        """Is the current user subscribed to this branch?"""

        if self.user is None:

            return False

        return self.context.hasSubscription(self.user)

    def recent_revision_count(self, days=30):

        """Number of revisions committed during the last N days."""

        timestamp = datetime.now(pytz.UTC) - timedelta(days=days)

        return self.context.getRevisionsSince(timestamp).count()

    def owner_is_registrant(self):

        """Is the branch owner the registrant?"""

        return self.context.owner == self.context.registrant

    def owner_is_reviewer(self):

        """Is the branch owner the default reviewer?"""

        if self.context.reviewer == None:

            return True

        return self.context.owner == self.context.reviewer

    def show_whiteboard(self):

        """Return whether or not the whiteboard should be shown.

        The whiteboard is only shown for import branches.

        """

        if (self.context.branch_type == BranchType.IMPORTED and

            self.context.whiteboard):

            return True

        else:

            return False

    def has_metadata(self):

        """Return whether there is branch metadata to display."""

        return (

            self.context.branch_format or

            self.context.repository_format or

            self.context.control_format or

            self.context.stacked_on)

    @property

    def is_empty_directory(self):

        """True if the branch is an empty directory without even a '.bzr'."""

        return self.context.control_format is None

    @property

    def codebrowse_url(self):

        """Return the link to codebrowse for this branch."""

        return self.context.codebrowse_url()

    @property

    def pending_writes(self):

        """Whether or not there are pending writes for this branch."""

        return self.context.pending_writes

    def bzr_download_url(self):

        """Return the generic URL for downloading the branch."""

        if self.user_can_download():

            return self.context.bzr_identity

        else:

            return None

    def bzr_upload_url(self):

        """Return the generic URL for uploading the branch."""

        if self.user_can_upload():

            return self.context.bzr_identity

        else:

            return None

    @property

    def user_can_upload(self):

        """Whether the user can upload to this branch."""

        branch = self.context

        if branch.branch_type != BranchType.HOSTED:

            return False

        return check_permission('launchpad.Edit', branch)

    def user_can_download(self):

        """Whether the user can download this branch."""

        return (self.context.branch_type != BranchType.REMOTE and

                self.context.revision_count > 0)

    @cachedproperty

    def landing_targets(self):

        """Return a filtered list of landing targets."""

        return latest_proposals_for_each_branch(self.context.landing_targets)

    @property

    def latest_landing_candidates(self):

        """Return a decorated filtered list of landing candidates."""

        # Only show the most recent 5 landing_candidates

        return self.landing_candidates[:5]

    @cachedproperty

    def landing_candidates(self):

        """Return a decorated list of landing candidates."""

        candidates = self.context.landing_candidates

        return [proposal for proposal in candidates

                if check_permission('launchpad.View', proposal)]

    @property

    def recipe_count_text(self):

        count = self.context.recipes.count()

        if count == 0:

            return 'No recipes'

        elif count == 1:

            return '1 recipe'

        else:

            return '%s recipes' % count

    @property

    def is_import_branch_with_no_landing_candidates(self):

        """Is the branch an import branch with no landing candidates?"""

        if self.landing_candidates:

            return False

        if not self.context.branch_type == BranchType.IMPORTED:

            return False

        return True

    def _getBranchCountText(self, count):

        """Help to show user friendly text."""

        if count == 0:

            return 'No branches'

        elif count == 1:

            return '1 branch'

        else:

            return '%s branches' % count

    @cachedproperty

    def dependent_branch_count_text(self):

        branch_count = len(self.dependent_branches)

        return self._getBranchCountText(branch_count)

    @cachedproperty

    def landing_candidate_count_text(self):

        branch_count = len(self.landing_candidates)

        return self._getBranchCountText(branch_count)

    @cachedproperty

    def dependent_branches(self):

        return [branch for branch in self.context.dependent_branches

                if check_permission('launchpad.View', branch)]

    @cachedproperty

    def no_merges(self):

        """Return true if there are no pending merges"""

        return (len(self.landing_targets) +

                len(self.landing_candidates) +

                len(self.dependent_branches) == 0)

    @property

    def show_candidate_more_link(self):

        """Only show the link if there are more than five."""

        return len(self.landing_candidates) > 5

    @cachedproperty

    def linked_bugtasks(self):

        """Return a list of bugtasks linked to the branch."""

        if self.context.is_series_branch:

            status_filter = searchbuilder.any(*UNRESOLVED_BUGTASK_STATUSES)

        else:

            status_filter = None

        return list(self.context.getLinkedBugTasks(

            self.user, status_filter))

    @cachedproperty

    def revision_info(self):

        collection = getUtility(IAllBranches).visibleByUser(self.user)

        return collection.getExtendedRevisionDetails(

            self.user, self.context.latest_revisions)

    @cachedproperty

    def latest_code_import_results(self):

        """Return the last 10 CodeImportResults."""

        return list(self.context.code_import.results[:10])

    def iconForCodeImportResultStatus(self, status):

        """The icon to represent the `CodeImportResultStatus` `status`."""

        if status == CodeImportResultStatus.SUCCESS_PARTIAL:

            return "/@@/yes-gray"

        elif status in CodeImportResultStatus.successes:

            return "/@@/yes"

        else:

            return "/@@/no"

    @property

    def is_svn_import(self):

        """True if an imported branch is a SVN import."""

        # You should only be calling this if it's a code import

        assert self.context.code_import

        return self.context.code_import.rcs_type in \

               (RevisionControlSystems.SVN, RevisionControlSystems.BZR_SVN)

    @property

    def url_is_web(self):

        """True if an imported branch's URL is HTTP or HTTPS."""

        # You should only be calling this if it's an SVN, BZR, GIT or HG code

        # import

        assert self.context.code_import

        url = self.context.code_import.url

        assert url

        # https starts with http too!

        return url.startswith("http")

    @property

    def show_merge_links(self):

        """Return whether or not merge proposal links should be shown.

        Merge proposal links should not be shown if there is only one branch

        in a non-final state.

        """

        if not self.context.target.supports_merge_proposals:

            return False

        return self.context.target.collection.getBranches().count() > 1

    def translations_sources(self):