/brz/remove-bazaar

# Copyright (C) 2011 Canonical Ltd

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA

"""Tools for dealing with the Launchpad API without using launchpadlib.

The api itself is a RESTful interface, so we can make HTTP queries directly.

loading launchpadlib itself has a fairly high overhead (just calling

Launchpad.login_anonymously() takes a 500ms once the WADL is cached, and 5+s to

get the WADL.

"""

try:

    # Use simplejson if available, much faster, and can be easily installed in

    # older versions of python

    import simplejson as json

except ImportError:

    # Is present since python 2.6

    try:

        import json

    except ImportError:

        json = None

import urllib

import urllib2

from bzrlib import (

    revision,

    trace,

    )

class LatestPublication(object):

    """Encapsulate how to find the latest publication for a given project."""

    LP_API_ROOT = 'https://api.launchpad.net/1.0'

    def __init__(self, archive, series, project):

        self._archive = archive

        self._project = project

        self._setup_series_and_pocket(series)

    def _setup_series_and_pocket(self, series):

        """Parse the 'series' info into a series and a pocket.

        eg::

            _setup_series_and_pocket('natty-proposed')

            => _series == 'natty'

               _pocket == 'Proposed'

        """

        self._series = series

        self._pocket = None

        if self._series is not None and '-' in self._series:

            self._series, self._pocket = self._series.split('-', 1)

            self._pocket = self._pocket.title()

        else:

            self._pocket = 'Release'

    def _archive_URL(self):

        """Return the Launchpad 'Archive' URL that we will query.

        This is everything in the URL except the query parameters.

        """

        return '%s/%s/+archive/primary' % (self.LP_API_ROOT, self._archive)

    def _publication_status(self):

        """Handle the 'status' field.

        It seems that Launchpad tracks all 'debian' packages as 'Pending', while

        for 'ubuntu' we care about the 'Published' packages.

        """

        if self._archive == 'debian':

            # Launchpad only tracks debian packages as "Pending", it doesn't mark

            # them Published

            return 'Pending'

        return 'Published'

    def _query_params(self):

        """Get the parameters defining our query.

        This defines the actions we are making against the archive.

        :return: A dict of query parameters.

        """

        params = {'ws.op': 'getPublishedSources',

                  'exact_match': 'true',

                  # If we need to use "" shouldn't we quote the project somehow?

                  'source_name': '"%s"' % (self._project,),

                  'status': self._publication_status(),

                  # We only need the latest one, the results seem to be properly

                  # most-recent-debian-version sorted

                  'ws.size': '1',

        }

        if self._series is not None:

            params['distro_series'] = '/%s/%s' % (self._archive, self._series)

        if self._pocket is not None:

            params['pocket'] = self._pocket

        return params

    def _query_URL(self):

        """Create the full URL that we need to query, including parameters."""

        params = self._query_params()

        # We sort to give deterministic results for testing

        encoded = urllib.urlencode(sorted(params.items()))

        return '%s?%s' % (self._archive_URL(), encoded)

    def _get_lp_info(self):

        """Place an actual HTTP query against the Launchpad service."""

        if json is None:

            return None

        query_URL = self._query_URL()

        try:

            req = urllib2.Request(query_URL)

            response = urllib2.urlopen(req)

            json_info = response.read()

        # TODO: We haven't tested the HTTPError

        except (urllib2.URLError, urllib2.HTTPError), e:

            trace.mutter('failed to place query to %r' % (query_URL,))

            trace.log_exception_quietly()

            return None

        return json_info

    def _parse_json_info(self, json_info):

        """Parse the json response from Launchpad into objects."""

        if json is None:

            return None

        try:

            return json.loads(json_info)

        except Exception:

            trace.mutter('Failed to parse json info: %r' % (json_info,))

            trace.log_exception_quietly()

            return None

    def get_latest_version(self):

        """Get the latest published version for the given package."""

        json_info = self._get_lp_info()

        if json_info is None:

            return None

        info = self._parse_json_info(json_info)

        if info is None:

            return None

        try:

            entries = info['entries']

            if len(entries) == 0:

                return None

            return entries[0]['source_package_version']

        except KeyError:

            trace.log_exception_quietly()

            return None

def get_latest_publication(archive, series, project):

    """Get the most recent publication for a given project.

    :param archive: Either 'ubuntu' or 'debian'

    :param series: Something like 'natty', 'sid', etc. Can be set as None. Can

        also include a pocket such as 'natty-proposed'.

    :param project: Something like 'bzr'

    :return: A version string indicating the most-recent version published in

        Launchpad. Might return None if there is an error.

    """

    lp = LatestPublication(archive, series, project)

    return lp.get_latest_version()

def get_most_recent_tag(tag_dict, the_branch):

    """Get the most recent revision that has been tagged."""

    # Note: this assumes that a given rev won't get tagged multiple times. But

    #       it should be valid for the package importer branches that we care

    #       about

    reverse_dict = dict((rev, tag) for tag, rev in tag_dict.iteritems())

    the_branch.lock_read()

    try:

        last_rev = the_branch.last_revision()

        graph = the_branch.repository.get_graph()

        stop_revisions = (None, revision.NULL_REVISION)

        for rev_id in graph.iter_lefthand_ancestry(last_rev, stop_revisions):

            if rev_id in reverse_dict:

                return reverse_dict[rev_id]

    finally:

        the_branch.unlock()