openqa-trigger-bisect-jobs

#!/usr/bin/env python3

import argparse
import logging
import re
import requests
import subprocess
import sys
from urllib.parse import urljoin, urlparse

logging.basicConfig()
log = logging.getLogger(sys.argv[0] if __name__ == '__main__' else __name__)


class CustomFormatter(argparse.ArgumentDefaultsHelpFormatter, argparse.RawDescriptionHelpFormatter):
    """Preserve multi-line __doc__ and provide default arguments in help strings."""

    pass


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=CustomFormatter)
    parser.add_argument('-v', '--verbose',
                        help='Increase verbosity level, specify multiple times to increase verbosity', action='count', default=1)
    parser.add_argument('--url', required=True,
                        help='The openQA test URL for which to trigger bisection investigation jobs')
    parser.add_argument('--dry-run', action='store_true',
                        help='Do not do any action on openQA')
    parser.add_argument('--include-depends', action='store_true', default=False,
                        help='Restart also the jobs on which the investigated one depends')
    args = parser.parse_args()
    verbose_to_log = {
        0: logging.CRITICAL,
        1: logging.ERROR,
        2: logging.WARN,
        3: logging.INFO,
        4: logging.DEBUG
    }
    logging_level = logging.DEBUG if args.verbose > 4 else verbose_to_log[args.verbose]
    log.setLevel(logging_level)
    return args


def call(cmds, dry_run=False):
    log.debug('call: %s' % cmds)
    return subprocess.call((['echo', 'Simulating: '] if dry_run else []) + cmds)


def openqa_clone(cmds, dry_run, include_depends, default_opts=['--skip-chained-deps', '--within-instance'], default_cmds=['_GROUP=0']):
    if include_depends:
      default_opts = ['--within-instance','--parental-inheritance']
    return call(['openqa-clone-job'] + default_opts + cmds + default_cmds, dry_run)


def main():
    args = parse_args()
    investigation_url = urljoin(args.url + '/', 'investigation_ajax')
    log.debug('Retrieving investigation info from %s' % investigation_url)
    out = requests.get(investigation_url)
    log.debug('Received investigation info: %s' % out)
    investigation = out.json()
    os_test_issues = [line for line in investigation['diff_to_last_good'].split('\n') if 'OS_TEST_ISSUES' in line]
    good, bad = [set(re.compile(': "([^"]*)",').search(line).group(1).split(',')) for line in os_test_issues]
    removed, added = list(good - bad), list(bad - good)
    log.debug('removed: %s, added: %s' % (removed, added))
    parsed_url = urlparse(args.url)
    test_url = urljoin(parsed_url.geturl(), '/api/v1/jobs/' + parsed_url.path.lstrip('/tests/'))
    log.debug('Retrieving additional job data from %s' % test_url)
    test_data = requests.get(test_url)
    log.debug('Received job data: %s' % test_data)
    test = test_data.json()['job']['settings']['TEST']
    log.debug("Found test name '%s'" % test)
    for issue in added:
        log.info("Triggering one bisection job without issue '%s'" % issue)
        new = ','.join(bad - set([issue]))
        log.debug("New set of OS_TEST_ISSUES='%s'" % new)
        test_name = test + ':investigate:bisect_without_%s' % issue
        openqa_clone([args.url, 'OS_TEST_ISSUES=' + new, 'TEST=' + test_name, 'OPENQA_INVESTIGATE_ORIGIN=' + args.url], args.dry_run, args.include_depends)


if __name__ == '__main__':
    main()