openSUSE-release-tools/staging-report.py

#!/usr/bin/python3

import argparse
from datetime import datetime, timedelta
from collections import defaultdict
import json

from osclib.comments import CommentAPI
from osclib.conf import Config
from osclib.stagingapi import StagingAPI
from lxml import etree as ET
import osc

MARGIN_HOURS = 4
MAX_LINES = 6
MARKER = 'StagingReport'


class StagingReport(object):
    def __init__(self, api):
        self.api = api
        self.comment = CommentAPI(api.apiurl)

    def _package_url(self, package):
        link = '/package/live_build_log/%s/%s/%s/%s'
        link = link % (package.get('project'),
                       package.get('package'),
                       package.get('repository'),
                       package.get('arch'))
        text = '[%s](%s)' % (package.get('arch'), link)
        return text

    def old_enough(self, _date):
        time_delta = datetime.utcnow() - _date
        safe_margin = timedelta(hours=MARGIN_HOURS)
        return safe_margin <= time_delta

    def update_status_comment(self, project, report, force=False, only_replace=False):
        report = self.comment.add_marker(report, MARKER)
        comments = self.comment.get_comments(project_name=project)
        comment, _ = self.comment.comment_find(comments, MARKER)
        if comment:
            write_comment = (report != comment['comment'] and self.old_enough(comment['when']))
        else:
            write_comment = not only_replace

        if write_comment or force:
            if osc.conf.config['debug']:
                print('Updating comment')
            if comment:
                self.comment.delete(comment['id'])
            self.comment.add_comment(project_name=project, comment=report)

    def _report_broken_packages(self, info):
        # Group packages by name
        groups = defaultdict(list)
        for package in info.findall('broken_packages/package'):
            groups[package.get('package')].append(package)

        failing_lines = [
            '* Build failed %s (%s)' % (key, ', '.join(self._package_url(p) for p in value))
            for key, value in groups.items()
        ]

        report = '\n'.join(failing_lines[:MAX_LINES])
        if len(failing_lines) > MAX_LINES:
            report += '* and more (%s) ...' % (len(failing_lines) - MAX_LINES)
        return report

    def report_checks(self, info):
        links_state = {}
        for check in info.findall('checks/check'):
            state = check.find('state').text
            links_state.setdefault(state, [])
            links_state[state].append('[{}]({})'.format(check.get('name'), check.find('url').text))

        lines = []
        failure = False
        for state, links in links_state.items():
            if len(links) > MAX_LINES:
                extra = len(links) - MAX_LINES
                links = links[:MAX_LINES]
                links.append('and {} more...'.format(extra))

            lines.append('- {}'.format(state))
            if state != 'success':
                lines.extend(['  - {}'.format(link) for link in links])
                failure = True
            else:
                lines[-1] += ': {}'.format(', '.join(links))

        return '\n'.join(lines).strip(), failure

    def report(self, project, force=False):
        info = self.api.project_status(project)

        # Do not attempt to process projects without staging info, or projects
        # in a pending state that will change before settling. This avoids
        # intermediate notifications that may end up being spammy and for
        # long-lived stagings where checks may be re-triggered multiple times
        # and thus enter pending state (not seen on first run) which is not
        # useful to report.
        if info is None or not self.api.project_status_final(info):
            return

        report_broken_packages = self._report_broken_packages(info)
        report_checks, check_failure = self.report_checks(info)

        if report_broken_packages or check_failure:
            if report_broken_packages:
                report_broken_packages = 'Broken:\n\n' + report_broken_packages
            if report_checks:
                report_checks = 'Checks:\n\n' + report_checks
            report = '\n\n'.join((report_broken_packages, report_checks))
            report = report.strip()
            only_replace = False
        else:
            report = 'Congratulations! All fine now.'
            only_replace = True

        report = self.cc_list(project, info) + report
        self.update_status_comment(project, report, force=force, only_replace=only_replace)

        if osc.conf.config['debug']:
            print(project)
            print('-' * len(project))
            print(report)

    def cc_list(self, project, info):
        if not self.api.is_adi_project(project):
            return ""
        ccs = set()
        for req in info.findall('staged_requests/request'):
            ccs.add("@" + req.get('creator'))
        str = "Submitters: " + " ".join(sorted(list(ccs))) + "\n\n"
        return str


if __name__ == '__main__':
    parser = argparse.ArgumentParser(
        description='Publish report on staging status as comment on staging project')
    parser.add_argument('-s', '--staging', type=str, default=None,
                        help='staging project')
    parser.add_argument('-f', '--force', action='store_true', default=False,
                        help='force a comment to be written')
    parser.add_argument('-p', '--project', type=str, default='openSUSE:Factory',
                        help='project to check (ex. openSUSE:Factory, openSUSE:Leap:15.1)')
    parser.add_argument('-d', '--debug', action='store_true', default=False,
                        help='enable debug information')
    parser.add_argument('-A', '--apiurl', metavar='URL', help='API URL')

    args = parser.parse_args()

    osc.conf.get_config(override_apiurl=args.apiurl)
    osc.conf.config['debug'] = args.debug

    apiurl = osc.conf.config['apiurl']
    Config(apiurl, args.project)
    api = StagingAPI(apiurl, args.project)
    staging_report = StagingReport(api)

    if args.staging:
        staging_report.report(api.prj_from_letter(args.staging), args.force)
    else:
        for staging in api.get_staging_projects():
            staging_report.report(staging, args.force)