~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).

"""People Merge related wiew classes."""

__metaclass__ = type

__all__ = [
    'AdminPeopleMergeView',
    'AdminTeamMergeView',
    'DeleteTeamView',
    'FinishedPeopleMergeRequestView',
    'RequestPeopleMergeMultipleEmailsView',
    'RequestPeopleMergeView',
    ]


from zope.component import getUtility
from zope.security.proxy import removeSecurityProxy

from canonical.launchpad import _
from canonical.launchpad.interfaces.authtoken import LoginTokenType
from canonical.launchpad.interfaces.emailaddress import (
    EmailAddressStatus,
    IEmailAddressSet,
    )
from canonical.launchpad.interfaces.logintoken import ILoginTokenSet
from canonical.launchpad.interfaces.lpstorm import IMasterObject
from canonical.launchpad.webapp import (
    canonical_url,
    LaunchpadView,
    )
from canonical.launchpad.webapp.interfaces import ILaunchBag
from lp.app.browser.launchpadform import (
    action,
    LaunchpadFormView,
    )
from lp.app.interfaces.launchpad import ILaunchpadCelebrities
from lp.registry.interfaces.mailinglist import (
    MailingListStatus,
    PURGE_STATES,
    )
from lp.registry.interfaces.person import (
    IAdminPeopleMergeSchema,
    IAdminTeamMergeSchema,
    IPersonSet,
    IRequestPeopleMerge,
    )
from lp.services.propertycache import cachedproperty
from lp.soyuz.enums import ArchiveStatus
from lp.soyuz.interfaces.archive import IArchiveSet


class ValidatingMergeView(LaunchpadFormView):

    def validate(self, data):
        """Check that user is not attempting to merge a person into itself."""
        dupe_person = data.get('dupe_person')
        target_person = data.get('target_person') or self.user
        if dupe_person is None:
            self.setFieldError(
                'dupe_person', 'The duplicate is not a valid person or team.')
        else:
            if dupe_person == target_person:
                self.addError(_("You can't merge ${name} into itself.",
                      mapping=dict(name=dupe_person.name)))
            dupe_person_ppas = getUtility(IArchiveSet).getPPAOwnedByPerson(
                dupe_person, statuses=[ArchiveStatus.ACTIVE,
                                       ArchiveStatus.DELETING])
            if dupe_person_ppas is not None:
                self.addError(_(
                    "${name} has a PPA that must be deleted before it "
                    "can be merged. It may take ten minutes to remove the "
                    "deleted PPA's files.",
                    mapping=dict(name=dupe_person.name)))
            if dupe_person.is_merge_pending:
                self.addError(_("${name} is already queued for merging.",
                      mapping=dict(name=dupe_person.name)))
        if target_person is not None and target_person.is_merge_pending:
            self.addError(_("${name} is already queued for merging.",
                  mapping=dict(name=target_person.name)))


class AdminMergeBaseView(ValidatingMergeView):
    """Base view for the pages where admins can merge people/teams."""

    page_title = 'Merge Launchpad accounts'
    # Both subclasses share the same template so we need to define these
    # variables (which are used in the template) here rather than on
    # subclasses.
    should_confirm_email_reassignment = False
    should_confirm_member_deactivation = False
    merge_message = _(
        'A merge is queued and is expected to complete in a few minutes.')

    dupe_person_emails = ()
    dupe_person = None
    target_person = None
    delete = False

    @property
    def cancel_url(self):
        return canonical_url(getUtility(IPersonSet))

    @property
    def success_url(self):
        return canonical_url(self.target_person)

    def render(self):
        # Subclasses may define other actions that they will render manually
        # only in certain circunstances, so don't include them in the list of
        # actions to be rendered.
        self.actions = [self.merge_action]
        return super(AdminMergeBaseView, self).render()

    def setUpPeople(self, data):
        """Store the people to be merged in instance variables.

        Also store all emails associated with the dupe account in an
        instance variable.
        """
        emailset = getUtility(IEmailAddressSet)
        self.dupe_person = data['dupe_person']
        self.target_person = data.get('target_person', None)
        self.dupe_person_emails = emailset.getByPerson(self.dupe_person)

    def doMerge(self, data):
        """Merge the two person/team entries specified in the form.

        Before merging this moves each email address of the duplicate person
        to the target person, and resets them to `NEW`.
        """
        if not self.dupe_person.is_team:
            # Transfer user email addresses. Team addresses will be deleted.
            for email in self.dupe_person_emails:
                email = IMasterObject(email)
                # EmailAddress.person and EmailAddress.account are readonly
                # fields, so we need to remove the security proxy here.
                naked_email = removeSecurityProxy(email)
                naked_email.personID = self.target_person.id
                naked_email.accountID = self.target_person.accountID
                naked_email.status = EmailAddressStatus.NEW
        getUtility(IPersonSet).mergeAsync(
            self.dupe_person, self.target_person, reviewer=self.user,
            delete=self.delete)
        self.request.response.addInfoNotification(self.merge_message)
        self.next_url = self.success_url


class AdminPeopleMergeView(AdminMergeBaseView):
    """A view for merging two Persons.

    If the duplicate person has any email addresses associated with we'll
    ask the user to confirm that it's okay to reassign these emails to the
    other account.  We do it because the fact that the dupe person still has
    email addresses is a possible indication that the admin may be merging
    the wrong person.
    """

    label = "Merge Launchpad people"
    schema = IAdminPeopleMergeSchema

    @action('Merge', name='merge')
    def merge_action(self, action, data):
        """Merge the two person entries specified in the form.

        If we're merging a person which has email addresses associated with
        we'll ask for confirmation before actually performing the merge.
        """
        self.setUpPeople(data)
        if self.dupe_person_emails.count() > 0:
            # We're merging a person which has one or more email addresses,
            # so we better warn the admin doing the operation and have him
            # check the emails that will be reassigned to ensure he's not
            # doing anything stupid.
            self.should_confirm_email_reassignment = True
            return
        self.doMerge(data)

    @action('Reassign E-mails and Merge', name='reassign_emails_and_merge')
    def reassign_emails_and_merge_action(self, action, data):
        """Reassign emails of the person to be merged and merge them."""
        self.setUpPeople(data)
        self.doMerge(data)


class AdminTeamMergeView(AdminMergeBaseView):
    """A view for merging two Teams.

    The duplicate team cannot be associated with a mailing list and if it
    has any active members we'll ask for confirmation from the user as we'll
    need to deactivate all members before we can do the merge.
    """

    label = "Merge Launchpad teams"
    schema = IAdminTeamMergeSchema

    def hasMailingList(self, team):
        unused_states = [state for state in PURGE_STATES]
        unused_states.append(MailingListStatus.PURGED)
        return (
            team.mailing_list is not None
            and team.mailing_list.status not in unused_states)

    @cachedproperty
    def registry_experts(self):
        return getUtility(ILaunchpadCelebrities).registry_experts

    def validate(self, data):
        """Check there are no mailing lists associated with the dupe team."""
        # If errors have already been discovered there is no need to continue,
        # especially since some of our expected data may be missing in the
        # case of user-entered invalid data.
        if len(self.errors) > 0:
            return

        super(AdminTeamMergeView, self).validate(data)
        dupe_team = data['dupe_person']
        # We cannot merge the teams if there is a mailing list on the
        # duplicate person, unless that mailing list is purged.
        if self.hasMailingList(dupe_team):
            self.addError(_(
                "${name} is associated with a Launchpad mailing list; we "
                "can't merge it.", mapping=dict(name=dupe_team.name)))

    @action('Merge', name='merge')
    def merge_action(self, action, data):
        """Merge the two team entries specified in the form.

        A confirmation will be asked if the team we're merging from still
        has active members, as in that case we'll have to deactivate all
        members first.
        """
        self.setUpPeople(data)
        if self.dupe_person.activemembers.count() > 0:
            # Merging teams with active members is not possible, so we'll
            # ask the admin if he wants to deactivate all members and then
            # merge.
            self.should_confirm_member_deactivation = True
            return
        super(AdminTeamMergeView, self).doMerge(data)

    @action('Deactivate Members and Merge',
            name='deactivate_members_and_merge')
    def deactivate_members_and_merge_action(self, action, data):
        """Deactivate all members of the team to be merged and merge them."""
        self.setUpPeople(data)
        super(AdminTeamMergeView, self).doMerge(data)


class DeleteTeamView(AdminTeamMergeView):
    """A view that deletes a team by merging it with Registry experts."""

    page_title = 'Delete'
    field_names = ['dupe_person']
    merge_message = _('The team is queued to be deleted.')

    @property
    def label(self):
        return 'Delete %s' % self.context.displayname

    def __init__(self, context, request):
        super(DeleteTeamView, self).__init__(context, request)
        if ('field.dupe_person' in self.request.form):
            # These fields have fixed values and are managed by this method.
            # The user has crafted a request to gain ownership of the dupe
            # team's assets.
            self.addError('Unable to process submitted data.')
        elif 'field.actions.delete' in self.request.form:
            # In the case of deleting a team, the form values are always
            # the context team, and the registry experts team. These values
            # are injected during __init__ because the base classes assume the
            # values are submitted. The validations performed by the base
            # classes are still required to ensure the team can be deleted.
            self.request.form.update(self.default_values)
        else:
            # Show the page explaining the action.
            pass

    @property
    def default_values(self):
        return {
            'field.dupe_person': self.context.name,
            'field.delete': True,
            }

    @property
    def cancel_url(self):
        return canonical_url(self.context)

    @property
    def success_url(self):
        return canonical_url(getUtility(IPersonSet))

    @property
    def has_mailing_list(self):
        return self.hasMailingList(self.context)

    def canDelete(self, data):
        return not self.has_mailing_list

    @action('Delete', name='delete', condition=canDelete)
    def merge_action(self, action, data):
        base = super(DeleteTeamView, self)
        self.delete = True
        base.deactivate_members_and_merge_action.success(data)


class FinishedPeopleMergeRequestView(LaunchpadView):
    """A simple view for a page where we only tell the user that we sent the
    email with further instructions to complete the merge.

    This view is used only when the dupe account has a single email address.
    """

    page_title = 'Merge request sent'

    def initialize(self):
        user = getUtility(ILaunchBag).user
        try:
            dupe_id = int(self.request.get('dupe'))
        except (ValueError, TypeError):
            self.request.response.redirect(canonical_url(user))
            return

        dupe_account = getUtility(IPersonSet).get(dupe_id)
        results = getUtility(IEmailAddressSet).getByPerson(dupe_account)

        result_count = results.count()
        if not result_count:
            # The user came back to visit this page with nothing to
            # merge, so we redirect him away to somewhere useful.
            self.request.response.redirect(canonical_url(user))
            return
        assert result_count == 1
        # Need to remove the security proxy because the dupe account may have
        # hidden email addresses.
        self.dupe_email = removeSecurityProxy(results[0]).email

    def render(self):
        if self.dupe_email:
            return LaunchpadView.render(self)
        else:
            return ''


class RequestPeopleMergeMultipleEmailsView(LaunchpadView):
    """Merge request view when dupe account has multiple email addresses."""

    label = 'Merge Launchpad accounts'
    page_title = label

    def __init__(self, context, request):
        super(RequestPeopleMergeMultipleEmailsView, self).__init__(
            context, request)
        self.form_processed = False
        self.dupe = None
        self.notified_addresses = []

    def processForm(self):
        dupe = self.request.form.get('dupe')
        if dupe is None:
            # We just got redirected to this page and we don't have the dupe
            # hidden field in request.form.
            dupe = self.request.get('dupe')
            if dupe is None:
                return

        self.dupe = getUtility(IPersonSet).get(int(dupe))
        emailaddrset = getUtility(IEmailAddressSet)
        self.dupeemails = emailaddrset.getByPerson(self.dupe)

        if self.request.method != "POST":
            return

        login = getUtility(ILaunchBag).login
        logintokenset = getUtility(ILoginTokenSet)

        email_addresses = []
        if self.email_hidden:
            # If the email addresses are hidden we must send a merge request
            # to each of them.  But first we've got to remove the security
            # proxy so we can get to them.
            email_addresses = [removeSecurityProxy(email).email
                               for email in self.dupeemails]
        else:
            # Otherwise we send a merge request only to the ones the user
            # selected.
            emails = self.request.form.get("selected")
            if emails is not None:
                # We can have multiple email addresses selected, and in this
                # case emails will be a list. Otherwise it will be a string
                # and we need to make a list with that value to use in the for
                # loop.
                if not isinstance(emails, list):
                    emails = [emails]

                for emailaddress in emails:
                    email = emailaddrset.getByEmail(emailaddress)
                    if email is None or email not in self.dupeemails:
                        # The dupe person has changes his email addresses.
                        # See bug 239838.
                        self.request.response.addNotification(
                            "An address was removed from the duplicate "
                            "account while you were making this merge "
                            "request. Select again.")
                        return
                    email_addresses.append(emailaddress)

        for emailaddress in email_addresses:
            token = logintokenset.new(
                self.user, login, emailaddress, LoginTokenType.ACCOUNTMERGE)
            token.sendMergeRequestEmail()
            self.notified_addresses.append(emailaddress)
        self.form_processed = True

    @property
    def cancel_url(self):
        """Cancel URL."""
        return canonical_url(self.user)

    @property
    def email_hidden(self):
        """Does the duplicate account hide email addresses?"""
        return self.dupe.hide_email_addresses


class RequestPeopleMergeView(ValidatingMergeView):
    """The view for the page where the user asks a merge of two accounts.

    If the dupe account have only one email address we send a message to that
    address and then redirect the user to other page saying that everything
    went fine. Otherwise we redirect the user to another page where we list
    all email addresses owned by the dupe account and the user selects which
    of those (s)he wants to claim.
    """

    label = 'Merge Launchpad accounts'
    page_title = label
    schema = IRequestPeopleMerge

    @property
    def cancel_url(self):
        return canonical_url(getUtility(IPersonSet))

    @action('Continue', name='continue')
    def continue_action(self, action, data):
        dupeaccount = data['dupe_person']
        if dupeaccount == self.user:
            # Please, don't try to merge you into yourself.
            return

        emails = getUtility(IEmailAddressSet).getByPerson(dupeaccount)
        emails_count = emails.count()
        if emails_count > 1:
            # The dupe account have more than one email address. Must redirect
            # the user to another page to ask which of those emails (s)he
            # wants to claim.
            self.next_url = '+requestmerge-multiple?dupe=%d' % dupeaccount.id
            return

        assert emails_count == 1
        email = emails[0]
        login = getUtility(ILaunchBag).login
        logintokenset = getUtility(ILoginTokenSet)
        # Need to remove the security proxy because the dupe account may have
        # hidden email addresses.
        token = logintokenset.new(
            self.user, login, removeSecurityProxy(email).email,
            LoginTokenType.ACCOUNTMERGE)
        token.sendMergeRequestEmail()
        self.next_url = './+mergerequest-sent?dupe=%d' % dupeaccount.id