ppa-britney/ubuntu-archive-tools/lputils.py

# Copyright 2012 Canonical Ltd.
# Author: Colin Watson <cjwatson@ubuntu.com>

# 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; version 3 of the License.
#
# 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, see <http://www.gnu.org/licenses/>.

"""Launchpad API utility functions."""

from operator import attrgetter

from debian import debian_support
from lazr.restfulclient.resource import Entry


class PackageMissing(Exception):
    "Generic exception generated by `lputils`."

    def __init__(self, message=None):
        Exception.__init__(self, message)
        self.message = message


known_pockets = (
    "Security",
    "Updates",
    "Proposed",
    "Backports",
    )

ARCHIVE_REFERENCE_DESCRIPTION = (
    'ARCHIVE can take one of four forms: "ubuntu" for a primary archive, '
    '"~canonical-kernel-team/ubuntu/ppa" or '
    '"ppa:canonical-kernel-team/ubuntu/ppa" for a PPA, or '
    '"ubuntu/partner" for a partner or copy archive.')


def setup_location(args, default_pocket="Release"):
    archive = None
    if getattr(args, "archive", False):
        # Try parsing an archive reference first.
        archive = args.launchpad.archives.getByReference(
            reference=args.archive)
        if archive is None:
            raise AssertionError("No such archive: %s" % args.archive)
    else:
        # Otherwise derive the archive from the deprecated
        # -d/--ppa/--ppa-name/--partner options.
        if isinstance(args.distribution, Entry):
            distro = args.distribution
        else:
            distro = args.launchpad.distributions[args.distribution]
        if getattr(args, "partner", False):
            archive = [
                archive for archive in distro.archives
                if archive.name == "partner"][0]
        elif getattr(args, "ppa", None):
            archive = args.launchpad.people[args.ppa].getPPAByName(
                distribution=distro, name=args.ppa_name)
        else:
            archive = distro.main_archive

    args.archive = archive
    args.distribution = archive.distribution
    if args.suite:
        if "-" in args.suite:
            args.series, args.pocket = args.suite.rsplit("-", 1)
            args.pocket = args.pocket.title()
            if args.pocket not in known_pockets:
                args.series = args.suite
                args.pocket = "Release"
        else:
            args.series = args.suite
            args.pocket = "Release"
        args.series = args.distribution.getSeries(name_or_version=args.series)
    else:
        args.series = args.distribution.current_series
        args.pocket = default_pocket
        if args.pocket == "Release":
            args.suite = args.series.name
        else:
            args.suite = "%s-%s" % (args.series.name, args.pocket.lower())

    if getattr(args, "architecture", None) is not None:
        args.architectures = [args.series.getDistroArchSeries(
            archtag=args.architecture)]
    elif getattr(args, "architectures", None) is not None:
        args.architectures = sorted(
            [a for a in args.series.architectures
             if a.architecture_tag in args.architectures],
            key=attrgetter("architecture_tag"))
    else:
        args.architectures = sorted(
            args.series.architectures, key=attrgetter("architecture_tag"))


def find_newest_publication(method, version_attr, **kwargs):
    """Hack around being unable to pass status=("Published", "Pending")."""
    published_pubs = method(status="Published", **kwargs)
    pending_pubs = method(status="Pending", **kwargs)
    try:
        newest_published = published_pubs[0]
        newest_published_ver = getattr(newest_published, version_attr)
    except IndexError:
        try:
            return pending_pubs[0]
        except IndexError:
            if kwargs["version"] is not None:
                try:
                    return method(**kwargs)[0]
                except IndexError:
                    return None
            else:
                return None
    try:
        newest_pending = pending_pubs[0]
        newest_pending_ver = getattr(newest_pending, version_attr)
    except IndexError:
        return newest_published
    if debian_support.version_compare(
            newest_published_ver, newest_pending_ver) > 0:
        return newest_published
    else:
        return newest_pending


def find_latest_published_binaries(args, package):
    target_binaries = []
    for architecture in args.architectures:
        binary = find_newest_publication(
            args.archive.getPublishedBinaries, "binary_package_version",
            binary_name=package, version=args.version,
            distro_arch_series=architecture, pocket=args.pocket,
            exact_match=True)
        if binary is not None:
            target_binaries.append(binary)
    if not target_binaries:
        raise PackageMissing(
            "Could not find binaries for '%s/%s' in %s" %
            (package, args.version, args.suite))
    return target_binaries


def find_latest_published_source(args, package):
    source = find_newest_publication(
        args.archive.getPublishedSources, "source_package_version",
        source_name=package, version=args.version,
        distro_series=args.series, pocket=args.pocket, exact_match=True)
    if source is None:
        raise PackageMissing(
            "Could not find source '%s/%s' in %s" %
            (package, args.version, args.suite))
    return source