~ubuntu-archive/ubuntu-archive-scripts/trunk

#!/usr/bin/python3

# Request re-runs of autopkgtests for packages

from datetime import datetime

import os

import sys

import argparse

import json

import urllib.parse

import amqplib.client_0_8 as amqp

#my_dir = os.path.dirname(os.path.realpath(sys.argv[0]))

my_dir = os.path.expanduser('~ubuntu-archive/proposed-migration/code/b2')

def parse_args():

    '''Parse command line arguments'''

    parser = argparse.ArgumentParser()

    parser.add_argument('-c', '--config',

                        help='britney config file (default: britney.conf.ubuntu.<series>)')

    parser.add_argument('-s', '--series', required=True,

                        help='Distro series name (required).')

    parser.add_argument('-a', '--architecture', action='append', default=[],

                        help='Only run test(s) on given architecture name(s). '

                        'Can be specified multiple times (default: all).')

    parser.add_argument('--trigger', action='append', default=[],

                        metavar='SOURCE/VERSION',

                        help='Add triggering package to request. '

                        'Can be specified multiple times.')

    parser.add_argument('--ppa', metavar='LPUSER/PPANAME', action='append',

                        default=[],

                        help='Enable PPA for requested test(s). '

                        'Can be specified multiple times.')

    parser.add_argument('--env', metavar='KEY=VALUE', action='append',

                        default=[],

                        help='List of VAR=value strings. '

                        'This can be used to influence a test\'s behaviour '

                        'from a test request. '

                        'Can be specified multiple times.')

    parser.add_argument('--test-git',

                        metavar='URL [branchname]',

                        help='A single URL or URL branchname. '

                        'The test will be git cloned from that URL and ran '

                        'from the checkout. This will not build binary '

                        'packages from the branch and run tests against '

                        'those, the test dependencies will be taken from the '

                        'archive, or PPA if given. In this case the '

                        'srcpkgname will only be used for the result path in '

                        'swift and be irrelevant for the actual test.')

    parser.add_argument('--build-git',

                        metavar='URL [branchname]',

                        help='A single URL or URL branchname. '

                        'Like --test-git`, but will first build binary '

                        'packages from the branch and run tests against those.')

    parser.add_argument('--test-bzr',

                        help='A single URL. '

                        'The test will be checked out with bzr from that URL. '

                        'Otherwise this has the same behaviour as test-git.')

    parser.add_argument('--all-proposed', action='store_true',

                        help='Disable apt pinning and use all of -proposed')

    parser.add_argument('--bulk', action='store_true',

                        help='Mark this as a bulk (low priority) test where possible')

    parser.add_argument('package', nargs='+',

                        help='Source package name(s) whose tests to run.')

    args = parser.parse_args()

    if not args.trigger and not args.ppa:

        parser.error('One of --trigger or --ppa must be given')

    # verify syntax of triggers

    for t in args.trigger:

        try:

            (src, ver) = t.split('/')

        except ValueError:

            parser.error('Invalid trigger format "%s", must be "sourcepkg/version"' % t)

    # verify syntax of PPAs

    for t in args.ppa:

        try:

            (user, name) = t.split('/')

        except ValueError:

            parser.error('Invalid ppa format "%s", must be "lpuser/ppaname"' % t)

    return args

def parse_config(config_file):

    '''Parse config file (like britney.py)'''

    config = argparse.Namespace()

    with open(config_file) as f:

        for k, v in [r.split('=', 1) for r in f if '=' in r and not r.strip().startswith('#')]:

            k = k.strip()

            if not getattr(config, k.lower(), None):

                setattr(config, k.lower(), v.strip())

    return config

if __name__ == '__main__':

    args = parse_args()

    britney_conf = os.path.join(my_dir, 'britney.conf')

    if args.config:

        config = parse_config(args.config)

    elif os.path.exists(britney_conf + '.ubuntu.' + args.series):

        config = parse_config(britney_conf + '.ubuntu.' + args.series)

    else:

        config = parse_config(britney_conf)

    if not args.architecture:

        args.architecture = config.adt_arches.split()

    context = ''

    params = {}

    if args.bulk:

        context = 'huge-'

    if args.trigger:

        params['triggers'] = args.trigger

    if args.ppa:

        params['ppas'] = args.ppa

        context = 'ppa-'

    if args.env:

        params['env'] = args.env

    if args.test_git:

        params['test-git'] = args.test_git

        context = 'upstream-'

    elif args.build_git:

        params['build-git'] = args.build_git

        context = 'upstream-'

    if args.test_bzr:

        params['test-bzr'] = args.test_bzr

        context = 'upstream-'

    if args.all_proposed:

        params['all-proposed'] = True

    try:

        params['requester'] = os.environ['SUDO_USER']

    except KeyError:

        pass

    params['submit-time'] = datetime.strftime(datetime.utcnow(), '%Y-%m-%d %H:%M:%S%z')

    params = '\n' + json.dumps(params)

    creds = urllib.parse.urlsplit(config.adt_amqp, allow_fragments=False)

    assert creds.scheme == 'amqp'

    with amqp.Connection(creds.hostname, userid=creds.username,

                         password=creds.password) as amqp_con:

        with amqp_con.channel() as ch:

            for arch in args.architecture:

                queue = 'debci-%s%s-%s' % (context, args.series, arch)

                for pkg in args.package:

                    ch.basic_publish(amqp.Message(pkg + params,

                                                  delivery_mode=2),  # persistent

                                     routing_key=queue)