ubuntu-dev-tools/requestsync

#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# (C) 2007 Canonical Ltd., Steve Kowalik
# Authors:
#  Martin Pitt <martin.pitt@ubuntu.com>
#  Steve Kowalik <stevenk@ubuntu.com>
#  Michael Bienia <geser@ubuntu.com> (python-launchpad-bugs support)
#  Daniel Hahler <ubuntu@thequod.de>
#  Iain Lane <iain@orangesquash.org.uk>
#  Jonathan Davies <jpds@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 2.
# 
# 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-2 for more details.
#
# ##################################################################

import os
import subprocess
import sys
import urllib
import urllib2
from debian_bundle.changelog import Version
from optparse import OptionParser
from time import sleep

# Use functions from ubuntu-dev-tools to create Launchpad cookie file.
sys.path.append('/usr/share/ubuntu-dev-tools/')
import common
    
launchpad_cookiefile = common.prepareLaunchpadCookie()

def checkNeedsSponsorship(component):
    """
        Check that the user has the appropriate permissions by checking what
        Launchpad returns while authenticating with their cookie.

        If they are an indirect or direct member of the ~ubuntu-dev team on
        Launchpad - sponsorship is not required if the package is in the
        universe / multiverse component.

        If they are in the ~ubuntu-core-dev team, no sponsorship required.

        The prepareLaunchpadCookie function above shall ensure that a cookie
        file exists first.
    """
    urlopener = common.setupLaunchpadUrlOpener(launchpad_cookiefile)

    # Check where the package is and assign the appropriate variables.
    if component in ['main', 'restricted']:
        team = "ubuntu-core-dev"
        sponsor = "ubuntu-main-sponsors"
    else:
        team = "ubuntu-dev"
        sponsor = "ubuntu-universe-sponsors"

    # Check if they are a member of the team.
    teamMember = common.isLPTeamMember(team)
    
    if not teamMember:
        print "You are not a member (direct or indirect) of the '%s' " \
            "team on Launchpad." % team
        print "Your sync request shall require an approval by a member of " \
            "the '%s'\nteam, who shall be subscribed to this bug report." % sponsor
        print "This must be done before it can be processed by a member of " \
            "the Ubuntu Archive team."
        print "If the above is correct please press Enter, otherwise please " \
            "press Ctrl-C to stop this script now\nand check the cookie file " \
            "at:", launchpad_cookiefile
        print "Logging into Launchpad, deleting the above and rerunning this " \
            "script should be enough to generate the cookie."
        raw_input_exit_on_ctrlc() # Abort if necessary.
        return True # Sponsorship required.

    # Is a team member, no sponsorship required.
    return False

def checkExistingReports(package, isNewPackage):
    """ Check existing bug reports on Launchpad for a possible sync request.

        If found ask for confirmation on filing a request.
    """
    # Determine if the package is new or not.
    if isNewPackage:
        # Package not in Ubuntu, check Ubuntu Archive Team's bug page for
        # possible duplicate reports.
        bugListUrl = "https://bugs.launchpad.net/~ubuntu-archive"
    else:
        # Package in Ubuntu, check the package's bug list for duplicate reports.
        bugListUrl = "https://bugs.launchpad.net/ubuntu/+source/%s" % package

    try:
        import launchpadbugs.connector as Connector
    except:
        # Failed to import launchpadbugs - skip check.
        print >> sys.stderr, "Unable to import launchpadbugs. Is " \
            "python-launchpad-bugs installed?"
        print >> sys.stderr, "Skipping existing report check, you should "\
            "manually check at:"
        print "-", bugListUrl
        return

    # Connect to the bug list.
    bugList = Connector.ConnectBugList()

    # Fetch data from Launchpad.
    pkgBugList = bugList(bugListUrl)

    if len(pkgBugList) == 0:
        return # No bugs found.

    # Search bug list for other sync requests.
    matchingBugs = [bug for bug in pkgBugList if "Please sync %s" % 
                        package in bug.summary]

    if len(matchingBugs) == 0:
        return # No sync bugs found.

    print "The following bugs could possibly be duplicate sync request(s) on Launchpad:"

    for bug in matchingBugs:
        print " *", bug.summary
        print "   -", bug.url
    
    print "Please check the above URLs to verify this before filing a " \
        "possible duplicate report."
    print "Press Ctrl-C to stop filing the bug report now, otherwise " \
         "please press enter."
    raw_input_exit_on_ctrlc()
            
def cur_version_component(sourcepkg, release):
    '''Determine current package version in ubuntu.'''
    madison = subprocess.Popen(['rmadison', '-u', 'ubuntu', '-a', 'source', \
                                '-s', release, sourcepkg], stdout=subprocess.PIPE)
    out = madison.communicate()[0]
    try:
        assert (madison.returncode == 0)
    except AssertionError:
        print out
        sys.exit(1)
        
    for l in out.splitlines():
        (pkg, version, rel, builds) = l.split('|')
        component = 'main'
        if rel.find('/') != -1:
            component = rel.split('/')[1]
        return (version.strip(), component.strip())
           
    print "%s doesn't appear to exist in %s, specify -n for a package not in Ubuntu." % (sourcepkg, release)
    sys.exit(1)

def cur_deb_version(sourcepkg, distro):
    '''Return the current debian version of a package in a Debian distro.'''
    madison = subprocess.Popen(['rmadison', '-u', 'debian', '-a', 'source', \
                                '-s', distro, sourcepkg], \
                               stdout=subprocess.PIPE)
    out = madison.communicate()[0]
    assert (madison.returncode == 0)

    try:
        assert out
    except AssertionError:
        print "%s doesn't appear to exist in Debian." % sourcepkg
        sys.exit(1)

    # Work-around for a bug in Debians madison.php script not returning
    # only the source line
    for line in out.splitlines():
        if line.find('source') > 0:
            out = line
   
    return out.split('|')[1].rstrip('[]''').strip()

def debian_changelog(sourcepkg, component, version):
    '''Return the Debian changelog from the latest up to the given version
    (exclusive).'''

    base_version = Version(version)

    ch = ''
    subdir = sourcepkg[0]

    if sourcepkg.startswith('lib'):
        subdir = 'lib%s' % sourcepkg[3]

    # Get the debian/changelog file from packages.debian.org.
    try:
        debianChangelogPage = urllib2.urlopen('http://packages.debian.org/changelogs/pool/%s/%s/%s/current/changelog.txt' % (component, subdir, sourcepkg))
    except urllib2.HTTPError, error:
        print >> sys.stderr, "Unable to connect to packages.debian.org. " \
            "Received a %s." % error.code
        sys.exit(1)

    for l in debianChangelogPage:
        if l.startswith(sourcepkg):
            ch_version = l[ l.find("(")+1 : l.find(")") ]
            if Version(ch_version) <= base_version:
                break
        ch += l

    return ch

def debian_component(sourcepkg, distro):
    '''Return the Debian component for the source package.'''
    madison = subprocess.Popen(['rmadison', '-u', 'debian', '-a', 'source', '-s', distro, \
                                sourcepkg], stdout=subprocess.PIPE)
    out = madison.communicate()[0]
    assert (madison.returncode == 0)

    try:
        assert out
    except AssertionError:
        print "%s doesn't appear to exist in Debian." % sourcepkg
        sys.exit(1)
    raw_comp = out.split('|')[2].split('/')
    component = 'main'
    if len(raw_comp) == 2:
        component = raw_comp[1].strip()
    return component

def raw_input_exit_on_ctrlc(*args, **kwargs):
    """A wrapper around raw_input() to exit with a normalized message on Control-C"""
    try:
        return raw_input(*args, **kwargs)
    except KeyboardInterrupt:
        print 'Abort requested. No sync request filed.'
        sys.exit(1)

def get_email_address():
    '''Get the DEBEMAIL environment variable or give an error.'''
    myemailaddr = os.getenv('DEBEMAIL')
    if not myemailaddr:
        print >> sys.stderr, 'The environment variable DEBEMAIL needs to be ' +\
            ' set to make use of this script, unless you use option --lp.'
    return myemailaddr

def mail_bug(source_package, subscribe, status, bugtitle, bugtext, keyid = None):
    '''Submit the sync request per email.
    Return True if email successfully send, otherwise False.'''
    
    import smtplib
    import socket

    to = 'new@bugs.launchpad.net'

    myemailaddr = get_email_address()
    if not myemailaddr:
        return False

    # generate initial mailbody
    mailbody = ''
    if source_package:
        mailbody += ' affects ubuntu/%s\n' % source_package
    else:
        mailbody += ' affects ubuntu\n'
    mailbody = mailbody + ' status %s\n importance wishlist\n subscribe %s\n\n%s' % (status, subscribe, bugtext)

    # prepare sign_command
    sign_command = 'gpg'
    for cmd in ('gpg2', 'gnome-gpg'):
        if os.access('/usr/bin/%s' % cmd, os.X_OK):
            sign_command = cmd

    gpg_command = [sign_command, '--clearsign']
    if keyid:
        gpg_command.extend(('-u', keyid))

    in_confirm_loop = True
    while in_confirm_loop:
        # sign it
        gpg = subprocess.Popen(gpg_command, stdin = subprocess.PIPE, stdout = subprocess.PIPE)
        signed_report = gpg.communicate(mailbody)[0]
        assert gpg.returncode == 0

        # generate email
        mail = 'From: %s\nTo: %s\nSubject: %s\n\n%s' % (myemailaddr, to, bugtitle, signed_report)

        # ask for confirmation and allow to edit:
        print mail
        print 'Do you want to edit the report before sending [y/N]? Press Control-C to abort.'
        while 1:
            val = raw_input_exit_on_ctrlc()
            if val.lower() in ('y', 'yes'):
                (bugtitle, mailbody) = edit_report(bugtitle, mailbody)
                break
            elif val.lower() in ('n', 'no', ''):
                in_confirm_loop = False
                break
            else:
                print "Invalid answer"

    # get server address
    mailserver = os.getenv('DEBSMTP')
    if mailserver:
        print 'Using custom SMTP server:', mailserver
    else:
        mailserver = 'fiordland.ubuntu.com'

    # get server port
    mailserver_port = os.getenv('DEBSMTP_PORT')
    if mailserver_port:
        print 'Using custom SMTP port:', mailserver_port
    else:
        mailserver_port = 25

    # connect to the server
    try:
        s = smtplib.SMTP(mailserver, mailserver_port)
    except socket.error, s:
        print >> sys.stderr, "Could not connect to mailserver %s at port %s: %s (%i)" % \
            (mailserver, mailserver_port, s[1], s[0])
        print "The port %s may be firewalled. Please try using requestsync with" \
            % mailserver_port
        print "the '--lp' flag to file a sync request with the launchpadbugs " \
            "module."
        return False

    # authenticate to the server
    mailserver_user = os.getenv('DEBSMTP_USER')
    mailserver_pass = os.getenv('DEBSMTP_PASS')
    if mailserver_user and mailserver_pass:
        try:
            s.login(mailserver_user, mailserver_pass)
        except smtplib.SMTPAuthenticationError:
            print 'Error authenticating to the server: invalid username and password.'
            s.quit()
            return False
        except:
            print 'Unknown SMTP error.'
            s.quit()
            return False

    s.sendmail(myemailaddr, to, mail)
    s.quit()
    print 'Sync request mailed.'

    return True

def post_bug(source_package, subscribe, status, bugtitle, bugtext):
    '''Use python-launchpad-bugs to submit the sync request.
    Return True if successfully posted, otherwise False.'''

    import glob, os.path

    try:
        import launchpadbugs.connector
        from launchpadbugs.lpconstants import HTTPCONNECTION
    except ImportError:
        print >> sys.stderr, 'Importing launchpadbugs failed. Is python-launchpad-bugs installed?'
        return False

    if source_package:
        product = {'name': source_package, 'target': 'ubuntu'}
    else:
        # new source package
        product = {'name': 'ubuntu'}

    in_confirm_loop = True
    while in_confirm_loop:
        print 'Summary:\n%s\n\nDescription:\n%s' % (bugtitle, bugtext)

        # ask for confirmation and allow to edit:
        print 'Do you want to edit the report before sending [y/N]? Press Control-C to abort.'
        while 1:
            val = raw_input_exit_on_ctrlc()
            if val.lower() in ('y', 'yes'):
                (bugtitle, bugtext) = edit_report(bugtitle, bugtext)
                break
            elif val.lower() in ('n', 'no', ''):
                in_confirm_loop = False
                break
            else:
                print "Invalid answer."

    # Create bug
    Bug = launchpadbugs.connector.ConnectBug()
    # Force the usage of stable Launchpad.
    Bug.set_connection_mode(HTTPCONNECTION.MODE.STABLE)

    # Use our cookie file for authentication.
    Bug.authentication = launchpad_cookiefile
    print "Using LP cookie at: %s for authentication." % launchpad_cookiefile 

    # Submit bug report.
    bug = Bug.New(product = product, summary = bugtitle, description = bugtext)
    sleep(2) # Wait in case of slow Launchpad.

    try:
        bug.importance = 'Wishlist'
    except IOError, s:
        print "Warning: setting importance failed: %s" % s
    
    bug.status = status
    bug.subscriptions.add(subscribe)
    sleep(1) # Wait.

    bug.commit()
    sleep(1) # Wait.

    print 'Sync request filed as bug #%i: https://launchpad.net/bugs/%i' % (bug.bugnumber, bug.bugnumber)

    return True

def edit_report(subject, body, changes_required=False):
    """Edit a report (consisting of subject and body) in sensible-editor.
    
    subject and body get decorated, before they are written to the temporary
    file and undecorated after editing again.
    If changes_required is True and the file has not been edited
    (according to its mtime), an error is written to STDERR and the
    program exits.
    Returns (new_subject, new_body).
    """
    import re
    import string

    report = "Summary (one line):\n%s\n\nDescription:\n%s" % (subject, body)

    # Create tempfile and remember mtime
    import tempfile
    report_file = tempfile.NamedTemporaryFile( prefix='requestsync_' )
    report_file.file.write(report)
    report_file.file.flush()
    mtime_before = os.stat( report_file.name ).st_mtime

    # Launch editor
    try:
        editor = subprocess.check_call( ['sensible-editor', report_file.name] )
    except subprocess.CalledProcessError, e:
        print >> sys.stderr, 'Error calling sensible-editor: %s\nAborting.' % (e,)
        sys.exit(1)

    # Check if the tempfile has been changed
    if changes_required:
        report_file_info = os.stat( report_file.name )
        if mtime_before == os.stat( report_file.name ).st_mtime:
            print >> sys.stderr, 'The temporary file %s has not been changed, but you have\nto explain why the Ubuntu changes can be dropped. Aborting. [Press ENTER]' % (report_file.name,)
            raw_input()
            sys.exit(1)

    report_file.file.seek(0)
    report = report_file.file.read()
    report_file.file.close()

    # Undecorate report again:
    (new_subject, new_body) = report.split("\nDescription:\n", 1)
    # Remove prefix and whitespace for subject:
    new_subject = string.rstrip( re.sub("\n", " ", re.sub("^Summary \(one line\):\s*", "", new_subject, 1)) )

    return (new_subject, new_body)


#
# entry point
#

if __name__ == '__main__':
    # Our usage options.
    usage = "Usage: %prog [-d distro] [-k keyid] [-n] [--lp] [-s] <source "
    usage += "package> <target release> [base version]"
    optParser = OptionParser(usage)

    optParser.add_option("-d", type = "string",
        dest = "dist", default = "unstable",
        help = "Debian distribution to sync from.")
    optParser.add_option("-k", type = "string",
        dest = "keyid", default = None,
        help = "GnuPG key ID to use for signing report.")
    optParser.add_option("-n", action = "store_true",
        dest = "newpkg", default = False,
        help = "Whether package to sync is a new package in Ubuntu.")
    optParser.add_option("--lp", action = "store_true",
        dest = "lpbugs", default = False,
        help = "Specify whether to use the launchpadbugs module for filing " \
            "report.")
    optParser.add_option("-s", action = "store_true",
        dest = "sponsor", default = False,
        help = "Force sponsorship requirement (shall be autodetected if not " \
            "specified).")

    (options, args) = optParser.parse_args()

    newsource = options.newpkg
    sponsorship = options.sponsor
    keyid = options.keyid
    use_lp_bugs = options.lpbugs
    need_interaction = False
    distro = options.dist

    if len(args) not in (2, 3):
        optParser.error("Source package / target release missing - please " \
            "specify.")

    if not use_lp_bugs and not get_email_address():
        sys.exit(1)

    srcpkg = args[0]
    release = args[1]
    force_base_ver = None

    # Base version specified.
    if len(args) == 3: force_base_ver = args[2]

    (cur_ver, component) = ('0', 'universe') # Let's assume universe
    
    # Find Ubuntu release's package version.
    if not newsource: (cur_ver, component) = cur_version_component(srcpkg, release)

    debiancomponent = debian_component(srcpkg, distro)
    # Find Debian release's package version.
    deb_version = cur_deb_version(srcpkg, distro)

    # Debian and Ubuntu versions are the same - stop.
    if deb_version == cur_ver:
        print 'The versions in Debian and Ubuntu are the same already (%s). Aborting.' % (deb_version,)
        sys.exit(1)

    # -s flag not specified - check if we do need sponsorship.
    if not sponsorship: sponsorship = checkNeedsSponsorship(component)

    # Check for existing sync requests.
    checkExistingReports(srcpkg, newsource)

    # Generate bug report.
    subscribe = 'ubuntu-archive'
    status = 'confirmed'
    if sponsorship:
        status = 'new'
        if component in ['main', 'restricted']:
            subscribe = 'ubuntu-main-sponsors'
        else:
            subscribe = 'ubuntu-universe-sponsors'

    report = 'Please sync %s %s (%s) from Debian %s (%s).\n\n' % (srcpkg, deb_version, component, distro, debiancomponent)
    title = report[:-2]
    
    base_ver = cur_ver
    uidx = base_ver.find('ubuntu')
    if uidx > 0:
        base_ver = base_ver[:uidx]
        need_interaction = True

        print 'Changes have been made to the package in Ubuntu.'
        print 'Please edit the report and give an explanation.'
        print 'Press ENTER to start your editor. Press Control-C to abort now.'
        print 'Not saving the report file will abort the request, too.'
        raw_input_exit_on_ctrlc()
        report += 'Explanation of the Ubuntu delta and why it can be dropped:\n' + \
                  '>>> ENTER_EXPLANATION_HERE <<<\n\n'

    uidx = base_ver.find('build')
    if uidx > 0:
        base_ver = base_ver[:uidx]

    if force_base_ver:
        base_ver = force_base_ver

    report += 'Changelog since current %s version %s:\n\n' % (release, cur_ver)
    report += debian_changelog(srcpkg, debiancomponent, base_ver) + '\n'

    if need_interaction:
        (_, report) = edit_report(title, report, changes_required=True)

    # Post sync request using Launchpad interface:
    srcpkg = not newsource and srcpkg or None
    if use_lp_bugs:
        # Map status to the values expected by lp-bugs
        mapping = {'new': 'New', 'confirmed': 'Confirmed'}
        if post_bug(srcpkg, subscribe, mapping[status], title, report):
            sys.exit(0)
        # Abort on error:
        print 'Something went wrong. No sync request filed.'
        sys.exit(1)

    # Mail sync request:
    if mail_bug(srcpkg, subscribe, status, title, report, keyid):
        sys.exit(0)

    print 'Something went wrong. No sync request filed.'
    sys.exit(1)