ubuntu-dev-tools/syncpackage

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2008-2010 Martin Pitt <martin.pitt@canonical.com>,
#               2010      Benjamin Drung <bdrung@ubuntu.com>,
#               2010-2011 Stefano Rivera <stefanor@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.
#
# 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-3 for more details.
#
# ##################################################################

import debian.deb822
import debian.debian_support
import optparse
import os
import shutil
import sys

from devscripts.logger import Logger

from lazr.restfulclient.errors import HTTPError

from ubuntutools.archive import (DebianSourcePackage, UbuntuSourcePackage,
                                 DownloadError)
from ubuntutools.config import UDTConfig, ubu_email
from ubuntutools.requestsync.mail import (getDebianSrcPkg
                                          as requestsync_mail_getDebianSrcPkg)
from ubuntutools.requestsync.lp import getDebianSrcPkg, getUbuntuSrcPkg
from ubuntutools.lp import udtexceptions
from ubuntutools.lp.lpapicache import Distribution, Launchpad
from ubuntutools.misc import split_release_pocket
from ubuntutools.question import YesNoQuestion
from ubuntutools import subprocess


class Version(debian.debian_support.Version):
    def strip_epoch(self):
        '''Removes the epoch from a Debian version string.

        strip_epoch(1:1.52-1) will return "1.52-1" and strip_epoch(1.1.3-1) will
        return "1.1.3-1".'''

        parts = self.full_version.split(':')
        if len(parts) > 1:
            del parts[0]
        version_without_epoch = ':'.join(parts)
        return version_without_epoch

    def get_related_debian_version(self):
        related_debian_version = self.full_version
        uidx = related_debian_version.find('ubuntu')
        if uidx > 0:
            related_debian_version = related_debian_version[:uidx]
        uidx = related_debian_version.find('build')
        if uidx > 0:
            related_debian_version = related_debian_version[:uidx]
        return Version(related_debian_version)

    def is_modified_in_ubuntu(self):
        return 'ubuntu' in self.full_version


def remove_signature(dscname):
    '''Removes the signature from a .dsc file if the .dsc file is signed.'''

    dsc_file = open(dscname)
    if dsc_file.readline().strip() == "-----BEGIN PGP SIGNED MESSAGE-----":
        unsigned_file = []
        # search until begin of body found
        for line in dsc_file:
            if line.strip() == "":
                break

        # search for end of body
        for line in dsc_file:
            if line.strip() == "":
                break
            unsigned_file.append(line)

        dsc_file.close()
        dsc_file = open(dscname, "w")
        dsc_file.writelines(unsigned_file)
        dsc_file.close()

def add_fixed_bugs(changes, bugs):
    '''Add additional Launchpad bugs to the list of fixed bugs in changes
       file.'''

    changes = [l for l in changes.split("\n") if l.strip() != ""]
    # Remove duplicates
    bugs = set(bugs)

    for i in xrange(len(changes)):
        if changes[i].startswith("Launchpad-Bugs-Fixed:"):
            bugs.update(changes[i][22:].strip().split(" "))
            changes[i] = "Launchpad-Bugs-Fixed: %s" % (" ".join(bugs))
            break
        elif i == len(changes) - 1:
            # Launchpad-Bugs-Fixed entry does not exist in changes file
            line = "Launchpad-Bugs-Fixed: %s" % (" ".join(bugs))
            changes.append(line)

    return "\n".join(changes + [""])

def sync_dsc(src_pkg, debian_dist, release, name, email, bugs, ubuntu_mirror,
             keyid=None, simulate=False, force=False):
    uploader = name + " <" + email + ">"

    src_pkg.pull_dsc()
    new_ver = Version(src_pkg.dsc["Version"])

    try:
        ubuntu_source = getUbuntuSrcPkg(src_pkg.source, release)
        ubuntu_ver = Version(ubuntu_source.getVersion())
        ubu_pkg = UbuntuSourcePackage(src_pkg.source, ubuntu_ver.full_version,
                                      ubuntu_source.getComponent(),
                                      mirrors=[ubuntu_mirror])
        ubu_pkg.pull_dsc()
        need_orig = ubuntu_ver.upstream_version != new_ver.upstream_version
    except udtexceptions.PackageNotFoundException:
        ubuntu_ver = Version('~')
        ubu_pkg = None
        need_orig = True
        Logger.info('%s does not exist in Ubuntu.', name)

    Logger.debug('Source %s: current version %s, new version %s',
                 src_pkg.source, ubuntu_ver, new_ver)
    Logger.debug('Needs source tarball: %s', str(need_orig))

    cur_ver = ubuntu_ver.get_related_debian_version()
    if ubuntu_ver.is_modified_in_ubuntu():
        if not force:
            Logger.error('--force is required to discard Ubuntu changes.')
            sys.exit(1)

        Logger.warn('Overwriting modified Ubuntu version %s, '
                    'setting current version to %s',
                    ubuntu_ver.full_version, cur_ver.full_version)
    if simulate:
        return

    try:
        src_pkg.pull()
    except DownloadError, e:
        Logger.error('Failed to download: %s', str(e))
        sys.exit(1)
    src_pkg.unpack()

    fakesync = not (need_orig or ubu_pkg.verify_orig())

    if fakesync:
        Logger.warn('The checksums of the Debian and Ubuntu packages mismatch. '
                    'A fake sync is required.')
        # Download Ubuntu files (override Debian source tarballs)
        try:
            ubu_pkg.pull()
        except DownloadError, e:
            Logger.error('Failed to download: %s', str(e))
            sys.exit(1)

    # change into package directory
    directory = src_pkg.source + '-' + new_ver.upstream_version
    Logger.command(('cd', directory))
    os.chdir(directory)

    # read Debian distribution from debian/changelog if not specified
    if debian_dist is None:
        line = open("debian/changelog").readline()
        debian_dist = line.split(" ")[2].strip(";")

    if not fakesync:
        # create the changes file
        changes_filename = "%s_%s_source.changes" % \
                           (src_pkg.source, new_ver.strip_epoch())
        cmd = ["dpkg-genchanges", "-S", "-v" + cur_ver.full_version,
               "-DDistribution=" + release,
               "-DOrigin=debian/" + debian_dist,
               "-e" + uploader]
        if need_orig:
            cmd.append("-sa")
        else:
            cmd.append("-sd")
        if not Logger.verbose:
            cmd += ["-q"]
        Logger.command(cmd + ['>', '../' + changes_filename])
        changes = subprocess.Popen(cmd, stdout=subprocess.PIPE).communicate()[0]

        # Add additional bug numbers
        if len(bugs) > 0:
            changes = add_fixed_bugs(changes, bugs)

        # remove extracted (temporary) files
        Logger.command(('cd', '..'))
        os.chdir('..')
        shutil.rmtree(directory, True)

        # write changes file
        changes_file = open(changes_filename, "w")
        changes_file.writelines(changes)
        changes_file.close()

        # remove signature and sign package
        remove_signature(src_pkg.dsc_name)
        if keyid is not False:
            cmd = ["debsign", changes_filename]
            if not keyid is None:
                cmd.insert(1, "-k" + keyid)
            Logger.command(cmd)
            subprocess.check_call(cmd)
    else:
        # Create fakesync changelog entry
        new_ver = Version(new_ver.full_version + "fakesync1")
        changes_filename = "%s_%s_source.changes" % \
                           (src_pkg.source, new_ver.strip_epoch())
        if len(bugs) > 0:
            message = "Fake sync due to mismatching orig tarball (LP: %s)." % \
                      (", ".join(["#" + str(b) for b in bugs]))
        else:
            message = "Fake sync due to mismatching orig tarball."
        cmd = ['dch', '-v', new_ver.full_version, '--force-distribution',
               '-D', release, message]
        env = {'DEBFULLNAME': name, 'DEBEMAIL': email}
        Logger.command(cmd)
        subprocess.check_call(cmd, env=env)

        # update the Maintainer field
        cmd = ["update-maintainer"]
        if not Logger.verbose:
            cmd.append("-q")
        Logger.command(cmd)
        subprocess.check_call(cmd)

        # Build source package
        cmd = ["debuild", "--no-lintian", "-S", "-v" + cur_ver.full_version]
        if need_orig:
            cmd += ['-sa']
        if keyid:
            cmd += ["-k" + keyid]
        Logger.command(cmd)
        returncode = subprocess.call(cmd)
        if returncode != 0:
            Logger.error('Source-only build with debuild failed. '
                         'Please check build log above.')
            sys.exit(1)

def fetch_source_pkg(package, dist, version, component, ubuntu_release, mirror):
    """Download the specified source package.
    dist, version, component, mirror can all be None.
    """
    if mirror is None:
        mirrors = []
    else:
        mirrors = [mirror]

    if package.endswith('.dsc'):
        return DebianSourcePackage(dscfile=package, mirrors=mirrors)

    if dist is None:
        dist = "unstable"
    requested_version = version
    if type(version) == str:
        version = Version(version)

    if version is None or component is None:
        try:
            debian_srcpkg = getDebianSrcPkg(package, dist)
        except (udtexceptions.PackageNotFoundException,
                udtexceptions.SeriesNotFoundException), e:
            Logger.error(str(e))
            sys.exit(1)
        if version is None:
            version = Version(debian_srcpkg.getVersion())
        try:
            ubuntu_srcpkg = getUbuntuSrcPkg(package, ubuntu_release)
            ubuntu_version = Version(ubuntu_srcpkg.getVersion())
        except udtexceptions.PackageNotFoundException:
            ubuntu_version = Version('~')
        except udtexceptions.SeriesNotFoundException, e:
            Logger.error(str(e))
            sys.exit(1)
        if ubuntu_version >= version:
            # The LP importer is maybe out of date
            debian_srcpkg = requestsync_mail_getDebianSrcPkg(package, dist)
            if requested_version is None:
                version = Version(debian_srcpkg.getVersion())
            if ubuntu_version >= version:
                Logger.error("Version in Debian %s (%s) isn't newer than "
                             "Ubuntu %s (%s)",
                             version, dist, ubuntu_version, ubuntu_release)
                sys.exit(1)
        if component is None:
            component = debian_srcpkg.getComponent()

    assert component in ('main', 'contrib', 'non-free')

    return DebianSourcePackage(package, version.full_version, component,
                               mirrors=mirrors)

def copy(src_pkg, release, simulate=False, force=False):
    """Copy a source package from Debian to Ubuntu using the Launchpad API."""
    ubuntu = Distribution('ubuntu')
    debian_archive = Distribution('debian').getArchive()
    ubuntu_archive = ubuntu.getArchive()
    if release is None:
        ubuntu_series = ubuntu.getDevelopmentSeries().name
        ubuntu_pocket = 'Release'
    else:
        ubuntu_series, ubuntu_pocket = split_release_pocket(release)

    # Ensure that the provided Debian version actually exists.
    try:
        debian_archive.getPublishedSources(
            source_name=src_pkg.source,
            version=src_pkg.version.full_version,
            exact_match=True)[0]
    except IndexError:
        Logger.error('Debian version %s does not exist!', src_pkg.version)
        sys.exit(1)

    try:
        ubuntu_spph = getUbuntuSrcPkg(src_pkg.source,
                                      ubuntu_series, ubuntu_pocket)
        ubuntu_pkg = UbuntuSourcePackage(src_pkg.source,
                                         ubuntu_spph.getVersion(),
                                         ubuntu_spph.getComponent(),
                                         mirrors=[])

        Logger.normal('Source %s -> %s/%s: current version %s, new version %s',
                      src_pkg.source, ubuntu_series, ubuntu_pocket,
                      ubuntu_pkg.version, src_pkg.version)

        ubuntu_version = Version(ubuntu_pkg.version.full_version)
        if not force and ubuntu_version.is_modified_in_ubuntu():
            Logger.error('--force is required to discard Ubuntu changes.')
            sys.exit(1)

        # Check whether a fakesync would be required.
        src_pkg.pull_dsc()
        ubuntu_pkg.pull_dsc()
        if not src_pkg.dsc.compare_dsc(ubuntu_pkg.dsc):
            Logger.error('The checksums of the Debian and Ubuntu packages '
                         'mismatch. A fake sync using --no-lp is required.')
            sys.exit(1)
    except udtexceptions.PackageNotFoundException:
        Logger.normal('Source %s -> %s/%s: not in Ubuntu, new version %s',
                      src_pkg.source, ubuntu_series, ubuntu_pocket,
                      src_pkg.version)
    if simulate:
        return

    answer = YesNoQuestion().ask("Sync this package", "no")
    if answer != "yes":
        return

    try:
        ubuntu_archive.copyPackage(
            source_name=src_pkg.source,
            version=src_pkg.version.full_version,
            from_archive=debian_archive,
            to_series=ubuntu_series,
            to_pocket=ubuntu_pocket,
            include_binaries=False)
    except HTTPError, error:
        Logger.error("HTTP Error %s: %s", error.response.status,
                     error.response.reason)
        Logger.error(error.content)
        sys.exit(1)

    Logger.normal('Request succeeded; you should get an e-mail once it is '
                  'processed.')

def main():
    usage = "%prog [options] <.dsc URL/path or package name>"
    epilog = "See %s(1) for more info." % os.path.basename(sys.argv[0])
    parser = optparse.OptionParser(usage=usage, epilog=epilog)

    parser.add_option("-d", "--distribution",
                      dest="dist", default=None,
                      help="Debian distribution to sync from.")
    parser.add_option("-r", "--release",
                      dest="release", default=None,
                      help="Specify target Ubuntu release.")
    parser.add_option("-V", "--debian-version",
                      dest="debversion", default=None,
                      help="Specify the version to sync from.")
    parser.add_option("-c", "--component",
                      dest="component", default=None,
                      help="Specify the Debian component to sync from.")
    parser.add_option("-v", "--verbose",
                      dest="verbose", action="store_true", default=False,
                      help="Display more progress information.")
    parser.add_option("--no-lp",
                      dest="lp", action="store_false", default=True,
                      help="Construct sync locally rather than letting "
                           "Launchpad copy the package directly (not "
                           "recommended).")
    parser.add_option('-l', '--lpinstance', metavar='INSTANCE',
                      dest='lpinstance', default=None,
                      help='Launchpad instance to connect to '
                           '(default: production).')
    parser.add_option("-n", "--uploader-name",
                      dest="uploader_name", default=None,
                      help="Use UPLOADER_NAME as the name of the maintainer "
                           "for this upload.")
    parser.add_option("-e", "--uploader-email",
                      dest="uploader_email", default=None,
                      help="Use UPLOADER_EMAIL as email address of the "
                           "maintainer for this upload.")
    parser.add_option("-k", "--key",
                      dest="keyid", default=None,
                      help="Specify the key ID to be used for signing.")
    parser.add_option('--dont-sign',
                      dest='keyid', action='store_false',
                      help='Do not sign the upload.')
    parser.add_option("-b", "--bug", metavar="BUG",
                      dest="bugs", action="append", default=list(),
                      help="Mark Launchpad bug BUG as being fixed by this "
                           "upload.")
    parser.add_option("-f", "--force",
                      dest="force", action="store_true", default=False,
                      help="Force sync over the top of Ubuntu changes.")
    parser.add_option('-D', '--debian-mirror', metavar='DEBIAN_MIRROR',
                      dest='debian_mirror',
                      help='Preferred Debian mirror '
                           '(default: %s)'
                           % UDTConfig.defaults['DEBIAN_MIRROR'])
    parser.add_option('-U', '--ubuntu-mirror', metavar='UBUNTU_MIRROR',
                      dest='ubuntu_mirror',
                      help='Preferred Ubuntu mirror '
                           '(default: %s)'
                           % UDTConfig.defaults['UBUNTU_MIRROR'])
    parser.add_option('--no-conf',
                      dest='no_conf', default=False, action='store_true',
                      help="Don't read config files or environment variables.")
    parser.add_option('--simulate',
                      dest='simulate', default=False, action='store_true',
                      help="Show what would be done, but don't actually do "
                           "it.")

    (options, args) = parser.parse_args()

    if len(args) == 0:
        parser.error('No .dsc URL/path or package name specified.')
    if len(args) > 1:
        parser.error('Multiple .dsc URLs/paths or package names specified: '
                     + ', '.join(args))

    invalid_bug_numbers = [bug for bug in options.bugs if not bug.isdigit()]
    if len(invalid_bug_numbers) > 0:
        parser.error('Invalid bug number(s) specified: '
                     + ', '.join(invalid_bug_numbers))

    if options.component not in (None, "main", "contrib", "non-free"):
        parser.error('%s is not a valid Debian component. '
                     'It should be one of main, contrib, or non-free.'
                     % options.component)

    if options.lp and options.uploader_name:
        parser.error('Uploader name can only be overridden using --no-lp.')
    if options.lp and options.uploader_email:
        parser.error('Uploader email address can only be overridden using '
                     '--no-lp.')
    # --key, --dont-sign, --debian-mirror, and --ubuntu-mirror are just
    # ignored with options.lp, and do not require warnings.

    Logger.verbose = options.verbose
    config = UDTConfig(options.no_conf)
    if options.debian_mirror is None:
        options.debian_mirror = config.get_value('DEBIAN_MIRROR')
    if options.ubuntu_mirror is None:
        options.ubuntu_mirror = config.get_value('UBUNTU_MIRROR')
    if options.uploader_name is None:
        options.uploader_name = ubu_email(export=False)[0]
    if options.uploader_email is None:
        options.uploader_email = ubu_email(export=False)[1]

    if options.lp:
        if args[0].endswith('.dsc'):
            parser.error('.dsc files can only be synced using --no-lp.')

        if options.lpinstance is None:
            options.lpinstance = config.get_value('LPINSTANCE')

        try:
            Launchpad.login(service=options.lpinstance, api_version='devel')
        except IOError:
            sys.exit(1)

        src_pkg = fetch_source_pkg(args[0], options.dist, options.debversion,
                                   options.component, options.release, None)

        copy(src_pkg, options.release, options.simulate, options.force)
    else:
        Launchpad.login_anonymously()
        if options.release is None:
            ubuntu = Launchpad.distributions["ubuntu"]
            options.release = ubuntu.current_series.name

        os.environ['DEB_VENDOR'] = 'Ubuntu'

        src_pkg = fetch_source_pkg(args[0], options.dist, options.debversion,
                                   options.component, options.release,
                                   options.debian_mirror)

        sync_dsc(src_pkg, options.dist, options.release, options.uploader_name,
                 options.uploader_email, options.bugs, options.ubuntu_mirror,
                 options.keyid, options.simulate, options.force)

if __name__ == "__main__":
    main()