~launchpad-pqm/launchpad/devel

"""Land an approved merge proposal."""

from bzrlib.errors import BzrCommandError
from launchpadlib.launchpad import Launchpad
from launchpadlib.uris import (
    DEV_SERVICE_ROOT,
    EDGE_SERVICE_ROOT,
    LPNET_SERVICE_ROOT,
    STAGING_SERVICE_ROOT,
    )
from lazr.uri import URI


class MissingReviewError(Exception):
    """Raised when we try to get a review message without enough reviewers."""


class MissingBugsError(Exception):
    """Merge proposal has no linked bugs and no [no-qa] tag."""


class MissingBugsIncrementalError(Exception):
    """Merge proposal has the [incr] tag but no linked bugs."""


class LaunchpadBranchLander:

    name = 'launchpad-branch-lander'

    def __init__(self, launchpad):
        self._launchpad = launchpad

    @classmethod
    def load(cls, service_root='production'):
        # XXX: JonathanLange 2009-09-24: No unit tests.
        # XXX: JonathanLange 2009-09-24 bug=435813: If cached data invalid,
        # there's no easy way to delete it and try again.
        launchpad = Launchpad.login_with(cls.name, service_root)
        return cls(launchpad)

    def load_merge_proposal(self, mp_url):
        """Get the merge proposal object for the 'mp_url'."""
        # XXX: JonathanLange 2009-09-24: No unit tests.
        web_mp_uri = URI(mp_url)
        api_mp_uri = self._launchpad._root_uri.append(
            web_mp_uri.path.lstrip('/'))
        return MergeProposal(self._launchpad.load(str(api_mp_uri)))

    def get_lp_branch(self, branch):
        """Get the launchpadlib branch based on a bzr branch."""
        # First try the public branch.
        branch_url = branch.get_public_branch()
        if branch_url:
            lp_branch = self._launchpad.branches.getByUrl(
                url=branch_url)
            if lp_branch is not None:
                return lp_branch
        # If that didn't work try the push location.
        branch_url = branch.get_push_location()
        if branch_url:
            lp_branch = self._launchpad.branches.getByUrl(
                url=branch_url)
            if lp_branch is not None:
                return lp_branch
        raise BzrCommandError(
            "No public branch could be found.  Please re-run and specify "
            "the URL for the merge proposal.")

    def get_merge_proposal_from_branch(self, branch):
        """Get the merge proposal from the branch."""

        lp_branch = self.get_lp_branch(branch)
        proposals = [
            mp for mp in lp_branch.landing_targets
            if mp.queue_status in ('Needs review', 'Approved')]
        if len(proposals) == 0:
            raise BzrCommandError(
                "The public branch has no open source merge proposals.  "
                "You must have a merge proposal before attempting to "
                "land the branch.")
        elif len(proposals) > 1:
            raise BzrCommandError(
                "The public branch has multiple open source merge "
                "proposals.  You must provide the URL to the one you wish "
                "to use.")
        return MergeProposal(proposals[0])


class MergeProposal:
    """Wrapper around launchpadlib `IBranchMergeProposal` for landing."""

    def __init__(self, mp):
        """Construct a merge proposal.

        :param mp: A launchpadlib `IBranchMergeProposal`.
        """
        self._mp = mp
        self._launchpad = mp._root

    @property
    def source_branch(self):
        """The push URL of the source branch."""
        return str(self._get_push_url(self._mp.source_branch))

    @property
    def target_branch(self):
        """The push URL of the target branch."""
        return str(self._get_push_url(self._mp.target_branch))

    @property
    def commit_message(self):
        """The commit message specified on the merge proposal."""
        return self._mp.commit_message

    @property
    def is_approved(self):
        """Is this merge proposal approved for landing."""
        return self._mp.queue_status == 'Approved'

    def get_stakeholder_emails(self):
        """Return a collection of people who should know about branch landing.

        Used to determine who to email with the ec2 test results.

        :return: A set of `IPerson`s.
        """
        # XXX: JonathanLange 2009-09-24: No unit tests.
        return set(
            map(get_email,
                [self._mp.source_branch.owner, self._launchpad.me]))

    def get_reviews(self):
        """Return a dictionary of all Approved reviews.

        Used to determine who has actually approved a branch for landing. The
        key of the dictionary is the type of review, and the value is the list
        of people who have voted Approve with that type.

        Common types include 'code', 'db', 'ui' and of course `None`.
        """
        reviews = {}
        for vote in self._mp.votes:
            comment = vote.comment
            if comment is None or comment.vote != "Approve":
                continue
            reviewers = reviews.setdefault(vote.review_type, [])
            reviewers.append(vote.reviewer)
        if self.is_approved and not reviews:
            reviews[None] = [self._mp.reviewer]
        return reviews

    def get_bugs(self):
        """Return a collection of bugs linked to the source branch."""
        return self._mp.source_branch.linked_bugs

    def _get_push_url(self, branch):
        """Return the push URL for 'branch'.

        This function is a work-around for Launchpad's lack of exposing the
        branch's push URL.

        :param branch: A launchpadlib `IBranch`.
        """
        # XXX: JonathanLange 2009-09-24: No unit tests.
        host = get_bazaar_host(str(self._launchpad._root_uri))
        # XXX: JonathanLange 2009-09-24 bug=435790: lazr.uri allows a path
        # without a leading '/' and then doesn't insert a '/' in the final
        # URL. Do it ourselves.
        return URI(scheme='bzr+ssh', host=host, path='/' + branch.unique_name)

    def build_commit_message(self, commit_text, testfix=False, no_qa=False,
                           incremental=False, rollback=None):
        """Get the Launchpad-style commit message for a merge proposal."""
        reviews = self.get_reviews()
        bugs = self.get_bugs()

        tags = [
            get_testfix_clause(testfix),
            get_reviewer_clause(reviews),
            get_bugs_clause(bugs),
            get_qa_clause(bugs, no_qa,
                incremental, rollback=rollback),
            ]

        # Make sure we don't add duplicated tags to commit_text.
        commit_tags = tags[:]
        for tag in tags:
            if tag in commit_text:
                commit_tags.remove(tag)

        if commit_tags:
            return '%s %s' % (''.join(commit_tags), commit_text)
        else:
            return commit_text

    def set_commit_message(self, commit_message):
        """Set the Launchpad-style commit message for a merge proposal."""
        self._mp.commit_message = commit_message
        self._mp.lp_save()


def get_testfix_clause(testfix=False):
    """Get the testfix clause."""
    if testfix:
        testfix_clause = '[testfix]'
    else:
        testfix_clause = ''
    return testfix_clause


def get_qa_clause(bugs, no_qa=False, incremental=False, rollback=None):
    """Check the no-qa and incremental options, getting the qa clause.

    The qa clause will always be or no-qa, or incremental, or no-qa and
    incremental, or a revno for the rollback clause, or no tags.

    See https://dev.launchpad.net/QAProcessContinuousRollouts for detailed
    explanation of each clause.
    """
    qa_clause = ""

    if not bugs and not no_qa and not incremental and not rollback:
        raise MissingBugsError

    if incremental and not bugs:
        raise MissingBugsIncrementalError

    if no_qa and incremental:
        qa_clause = '[no-qa][incr]'
    elif incremental:
        qa_clause = '[incr]'
    elif no_qa:
        qa_clause = '[no-qa]'
    elif rollback:
        qa_clause = '[rollback=%d]' % rollback
    else:
        qa_clause = ''

    return qa_clause


def get_email(person):
    """Get the preferred email address for 'person'."""
    email_object = person.preferred_email_address
    return email_object.email


def get_bugs_clause(bugs):
    """Return the bugs clause of a commit message.

    :param bugs: A collection of `IBug` objects.
    :return: A string of the form "[bug=A,B,C]".
    """
    if not bugs:
        return ''
    bug_ids = []
    for bug in bugs:
        for task in bug.bug_tasks:
            if (task.bug_target_name == 'launchpad'
                and task.status not in ['Fix Committed', 'Fix Released']):
                bug_ids.append(str(bug.id))
                break
    if not bug_ids:
        return ''
    return '[bug=%s]' % ','.join(bug_ids)


def get_reviewer_handle(reviewer):
    """Get the handle for 'reviewer'.

    The handles of reviewers are included in the commit message for Launchpad
    changes. Historically, these handles have been the IRC nicks. Thus, if
    'reviewer' has an IRC nickname for Freenode, we use that. Otherwise we use
    their Launchpad username.

    :param reviewer: A launchpadlib `IPerson` object.
    :return: unicode text.
    """
    irc_handles = reviewer.irc_nicknames
    for handle in irc_handles:
        if handle.network == 'irc.freenode.net':
            return handle.nickname
    return reviewer.name


def _comma_separated_names(things):
    """Return a string of comma-separated names of 'things'.

    The list is sorted before being joined.
    """
    return ','.join(sorted(thing.name for thing in things))


def get_reviewer_clause(reviewers):
    """Get the reviewer section of a commit message, given the reviewers.

    :param reviewers: A dict mapping review types to lists of reviewers, as
        returned by 'get_reviews'.
    :return: A string like u'[r=foo,bar][ui=plop]'.
    """
    # If no review type is specified it is assumed to be a code review.
    code_reviewers = reviewers.get(None, [])
    ui_reviewers = []
    rc_reviewers = []
    for review_type, reviewer in reviewers.items():
        if review_type is None:
            continue
        if review_type == '':
            code_reviewers.extend(reviewer)
        if 'code' in review_type or 'db' in review_type:
            code_reviewers.extend(reviewer)
        if 'ui' in review_type:
            ui_reviewers.extend(reviewer)
        if 'release-critical' in review_type:
            rc_reviewers.extend(reviewer)
    if not code_reviewers:
        raise MissingReviewError("Need approved votes in order to land.")
    if ui_reviewers:
        ui_clause = '[ui=%s]' % _comma_separated_names(ui_reviewers)
    else:
        ui_clause = ''
    if rc_reviewers:
        rc_clause = (
            '[release-critical=%s]' % _comma_separated_names(rc_reviewers))
    else:
        rc_clause = ''
    return '%s[r=%s]%s' % (
        rc_clause, _comma_separated_names(code_reviewers), ui_clause)


def get_bazaar_host(api_root):
    """Get the Bazaar service for the given API root."""
    # XXX: JonathanLange 2009-09-24 bug=435803: This is only needed because
    # Launchpad doesn't expose the push URL for branches.
    if api_root.startswith(EDGE_SERVICE_ROOT):
        return 'bazaar.launchpad.net'
    elif api_root.startswith(DEV_SERVICE_ROOT):
        return 'bazaar.launchpad.dev'
    elif api_root.startswith(STAGING_SERVICE_ROOT):
        return 'bazaar.staging.launchpad.net'
    elif api_root.startswith(LPNET_SERVICE_ROOT):
        return 'bazaar.launchpad.net'
    else:
        raise ValueError(
            'Cannot determine Bazaar host. "%s" not a recognized Launchpad '
            'API root.' % (api_root,))