ubuntu-dev-tools/pull-uca-source

#!/usr/bin/python3
#
# pull-uca-source -- pull a source package from Ubuntu Cloud Archive
# Basic usage: pull-uca-source <source package> <openstack release> [version]
#
# Copyright (C) 2008,      Iain Lane <iain@orangesquash.org.uk>,
#               2010-2011, Stefano Rivera <stefanor@ubuntu.com>
#               2016,      Corey Bryant <corey.bryant@ubuntu.com>
#               2016,      Dan Streetman <dan.streetman@canonical.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; either version 3
# 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.
#
# See file /usr/share/common-licenses/GPL for more details.
#
# ##################################################################


import re
import sys
from optparse import OptionParser

from ubuntutools.archive import UbuntuCloudArchiveSourcePackage, DownloadError
from ubuntutools.config import UDTConfig
from ubuntutools.lp.lpapicache import Launchpad
from ubuntutools.lp.udtexceptions import PocketDoesNotExistError
from ubuntutools.logger import Logger
from ubuntutools.misc import split_release_pocket

from lazr.restfulclient.errors import NotFound

from launchpadlib.launchpad import Launchpad as LP


def showOpenstackReleases(uca):
    releases = []
    for p in uca.ppas:
        if re.match(r"\w*-staging", p.name):
            releases.append(re.sub("-staging", "", p.name))
    Logger.error("Openstack releases are:\n\t%s", ", ".join(releases))


def getSPPH(lp, archive, package, version=None, series=None, pocket=None, try_binary=True):
    params = {'exact_match': True, 'order_by_date': True}
    if pocket:
        params['pocket'] = pocket
    if series:
        params['distro_series'] = series()
    elif version:
        params['version'] = version
    Logger.normal("checking %s version %s pocket %s", package, version, pocket)
    spphs = archive.getPublishedSources(source_name=package, **params)
    if spphs:
        return spphs[0]
    if not try_binary:
        return None

    # Didn't find any, maybe the package is a binary package name
    if series:
        del params['distro_series']
        archs = lp.load(series().architectures_collection_link).entries
        params['distro_arch_series'] = archs[0]['self_link']
    bpphs = archive.getPublishedBinaries(binary_name=package, **params)
    if bpphs:
        bpph_build = lp.load(bpphs[0].build_link)
        source_package = bpph_build.source_package_name
        return getSPPH(lp, archive, source_package, version, series, pocket,
                       try_binary=False)

    return None


def main():
    usage = "Usage: %prog <package> <openstack release> [version]"
    opt_parser = OptionParser(usage)
    opt_parser.add_option('-d', '--download-only',
                          dest='download_only', default=False,
                          action='store_true',
                          help="Do not extract the source package")
    opt_parser.add_option('-m', '--mirror', metavar='OPENSTACK_MIRROR',
                          dest='openstack_mirror',
                          help='Preferred Openstack mirror (default: Launchpad)')
    opt_parser.add_option('--no-conf',
                          dest='no_conf', default=False, action='store_true',
                          help="Don't read config files or environment "
                               "variables")
    (options, args) = opt_parser.parse_args()
    if len(args) < 2:
        opt_parser.error("Must specify package name and openstack release")

    config = UDTConfig(options.no_conf)
    if options.openstack_mirror is None:
        options.openstack_mirror = config.get_value('OPENSTACK_MIRROR')
    mirrors = []
    if options.openstack_mirror:
        mirrors.append(options.openstack_mirror)

    # Login anonymously to LP
    Launchpad.login_anonymously()
    lp = LP.login_anonymously("pull-uca-source", "production")
    uca = lp.people("ubuntu-cloud-archive")

    package = str(args[0]).lower()
    release = str(args[1]).lower()
    version = None
    if len(args) > 2:
        version = str(args[2])

    pocket = None
    try:
        (release, pocket) = split_release_pocket(release, default=None)
    except PocketDoesNotExistError as e:
        pass

    try:
        archive = uca.getPPAByName(name="%s-staging" % release)
    except NotFound as e:
        Logger.error('Archive does not exist for Openstack release: %s',
                     release)
        showOpenstackReleases(uca)
        sys.exit(1)

    spph = getSPPH(lp, archive, package, version, pocket=pocket)
    if not spph:
        Logger.error("Package %s in %s not found.", package, release)
        sys.exit(1)

    package = spph.source_package_name
    version = spph.source_package_version
    component = spph.component_name
    Logger.normal('Downloading %s version %s component %s', package, version, component)
    srcpkg = UbuntuCloudArchiveSourcePackage(release, package, version, component=component,
                                             mirrors=mirrors)

    try:
        srcpkg.pull()
    except DownloadError as e:
        Logger.error('Failed to download: %s', str(e))
        sys.exit(1)
    if not options.download_only:
        srcpkg.unpack()


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        Logger.normal('User abort.')