~wgrant/lp-ftbfs-report/production

#!/usr/bin/python3
# -*- coding: UTF-8 -*-

# Copyright © 2007-2010 Michael Bienia <geser@ubuntu.com>
# Authors:
# Michael Bienia <geser@ubuntu.com>
# Andrea Gasparini <gaspa@yattaweb.it>
# License:
# GPLv2 (or later), see /usr/share/common-licenses/GPL

# Rewrite of the old build_status script using LP API

# Requirements:
# - python-launchpadlib
# - python-apt
# - python-jinja2

# Uncomment for tracing LP API calls
#import httplib2
#httplib2.debuglevel = 1

import functools
import os
import requests
import sys
import time
import apt_pkg
import json
from datetime import datetime
from jinja2 import (Environment, FileSystemLoader)
from launchpadlib.errors import HTTPError
from launchpadlib.launchpad import Launchpad
from operator import (attrgetter, methodcaller)
from optparse import OptionParser

lp_service = 'production'
api_version = 'devel'
default_arch_list = []
find_tagged_bugs = 'ftbfs'
apt_pkg.init_system()

# copied from ubuntu-dev-tools, libsupport.py:
def translate_api_web(self_url):
    if self_url is None:
        return ''
    else:
        return self_url.replace('api.', '').replace('%s/' % api_version, '')

class PersonTeam(object):
    _cache = dict()

    def __new__(cls, personteam_link):
        try:
            return cls._cache[personteam_link]
        except KeyError:
            try:
                personteam = super(PersonTeam, cls).__new__(cls)

                # fill the new PersonTeam object with data
                lp_object = launchpad.load(personteam_link)
                personteam.display_name = lp_object.display_name
                personteam.name = lp_object.name

            except KeyError:
                return None
            except HTTPError as e:
                if e.response.status == 410:
                    personteam = None
                else:
                    raise

            # add to cache
            cls._cache[personteam_link] = personteam

            return personteam

    @classmethod
    def clear(cls):
        cls._cache.clear()

    def __str__(self):
        return '%s (%s)' % (self.display_name, self.name)

class SourcePackage(object):
    _cache = dict()

    class VersionList(list):
        def append(self, item):
            super(SourcePackage.VersionList, self).append(item)
            self.sort(key = functools.cmp_to_key(lambda a, b: apt_pkg.version_compare(a.version, b.version)))

    def __new__(cls, spph):
        try:
            return cls._cache[spph.source_package_name]
        except KeyError:
            srcpkg = super(SourcePackage, cls).__new__(cls)

            # fill the new SourcePackage object with data
            srcpkg.name = spph.source_package_name
            srcpkg.url = 'https://launchpad.net/ubuntu/+source/%s' % srcpkg.name
            srcpkg.versions = cls.VersionList()
            if find_tagged_bugs is None:
                srcpkg.tagged_bugs = []
            else:
                ts = ubuntu.getSourcePackage(name=srcpkg.name).searchTasks(tags=find_tagged_bugs)
                srcpkg.tagged_bugs = [t.bug for t in ts]
            srcpkg.packagesets = set([ps for (ps, srcpkglist) in packagesets.items() if spph.source_package_name in srcpkglist])
            components[spph.component_name].append(srcpkg)
            for ps in srcpkg.packagesets:
                packagesets_ftbfs[ps].append(srcpkg)

            srcpkg.teams = set([team for (team, srcpkglist) in teams.items() if spph.source_package_name in srcpkglist and spph.component_name == "main"])
            for team in srcpkg.teams:
                teams_ftbfs[team].append(srcpkg)

            # add to cache
            cls._cache[spph.source_package_name] = srcpkg

            return srcpkg

    @classmethod
    def clear(cls):
        cls._cache.clear()

    def __lt__(self, other):
        return self.name < other.name

    def isFTBFS(self, arch_list = default_arch_list, current = True):
        ''' Returns True if at least one FTBFS exists. '''
        for ver in self.versions:
            if ver.current != current:
                continue
            for arch in arch_list:
                log = ver.getArch(arch)
                if log is not None:
                    return True
        return False

    def getCount(self, arch, state):
        count = 0
        for ver in self.versions:
            if arch in ver.logs and ver.logs[arch].buildstate == state:
                count += 1
        return count

    def getPackagesets(self, name=None):
        '''Return the list of packagesets without the packageset `name`.'''
        if name is None:
            return list(self.packagesets)
        else:
            return list(self.packagesets.difference((name,)))

class MainArchiveBuilds(object):
    _cache = dict()

    def __new__(cls, main_archive, source, version):
        try:
            return cls._cache["%s,%s" % (source, version)]
        except KeyError:
            bfm = super(MainArchiveBuilds, cls).__new__(cls)
            results = {}
            sourcepubs = main_archive.getPublishedSources(
                exact_match=True, source_name=source, version=version)
            for pub in sourcepubs:
                for build in pub.getBuilds():
                    # assumes sourcepubs are sorted latest release to oldest,
                    # so first record wins
                    if build.arch_tag not in results:
                        results[build.arch_tag] = build.buildstate
            bfm.results = results
            # add to cache
            cls._cache["%s,%s" % (source, version)] = bfm

            return bfm

    @classmethod
    def clear(cls):
        cls._cache.clear()

class SPPH(object):
    _cache = dict() # dict with all SPPH objects

    def __new__(cls, spph_link):
        try:
            return cls._cache[spph_link]
        except KeyError:
            spph = super(SPPH, cls).__new__(cls)

            # fill the new SPPH object with data
            lp_object = launchpad.load(spph_link)
            spph._lp = lp_object
            spph.logs = dict()
            spph.version = lp_object.source_package_version
            spph.pocket = lp_object.pocket
            spph.changed_by = PersonTeam(lp_object.package_creator_link)
            #spph.signed_by = spph._lp.package_signer_link and PersonTeam(lp_object.package_signer_link)
            spph.current = None
            SourcePackage(lp_object).versions.append(spph)

            # add to cache
            cls._cache[spph_link] = spph

            return spph

    @classmethod
    def clear(cls):
        cls._cache.clear()

    class BuildLog(object):
        def __init__(self, build):
            buildstates = {
                    'Failed to build': 'FAILEDTOBUILD',
                    'Dependency wait': 'MANUALDEPWAIT',
                    'Chroot problem': 'CHROOTWAIT',
                    'Failed to upload': 'UPLOADFAIL',
                    'Cancelled build': 'CANCELLED',
                    }
            self.buildstate = buildstates[build.buildstate]
            self.url = translate_api_web(build.self_link)

            if self.buildstate == 'UPLOADFAIL':
                self.log = translate_api_web(build.upload_log_url)
            else:
                if build.build_log_url:
                    self.log = translate_api_web(build.build_log_url)
                else:
                    self.log = ''

            if self.buildstate == 'MANUALDEPWAIT':
                self.tooltip = 'waits on %s' % build.dependencies
            elif build.datebuilt is None:
                self.tooltip = 'Broken build'
            else:
                if build.datebuilt:
                    self.tooltip = 'Build finished on %s' % build.datebuilt.strftime('%Y-%m-%d %H:%M:%S UTC')
                else:
                    self.tooltip = 'Build finish unknown'

    def addBuildLog(self, buildlog):
        self.logs[buildlog.arch_tag] = self.BuildLog(buildlog)

    def getArch(self, arch):
        return self.logs.get(arch)

    def getChangedBy(self):
        '''
        Returns a string with the person who changed this package.
        '''
        return 'Changed-By: %s' % (self.changed_by)


def fetch_pkg_list(archive, series, state, last_published, arch_list=default_arch_list, main_archive=None, main_series=None, release_only=False):
    print("Processing '%s'" % state)

    cur_last_published = None
    # XXX wgrant 2009-09-19: This is an awful hack. We should really
    # just let IArchive.getBuildRecords take a series argument.
    if archive.name == 'primary':
        buildlist = series.getBuildRecords(build_state = state)
    else:
        buildlist = archive.getBuildRecords(build_state = state)

    for build in buildlist:
        if (last_published is not None and
            build.datebuilt is not None and
            last_published > build.datebuilt.replace(tzinfo=None)):
                # leave the loop as we're past the last known published build record
                break

        csp_link = build.current_source_publication_link
        if not csp_link:
            # Build log for an older version
            continue

        if build.arch_tag not in arch_list:
            print("  Skipping %s" % build.title)
            continue

        cur_last_published = build.datebuilt

        print("  %s %s" % (build.datebuilt, build.title))

        spph = SPPH(csp_link)

        if spph.current is None:
            # If a main archive is specified, we check if the current source
            # is still published there. If it isn't, then it's out of date.
            # We should make this obvious.
            # The main archive will normally be the primary archive, and
            # probably only makes sense if the target archive is a rebuild.
            if main_archive:
                main_publications = main_archive.getPublishedSources(
                    distro_series=main_series,
                    exact_match=True,
                    source_name=spph._lp.source_package_name,
                    version=spph._lp.source_package_version,
                    status='Published')
                spph.current = len(main_publications[:1]) > 0
            elif release_only:
                release_publications = archive.getPublishedSources(
                    distro_series=series,
                    pocket='Release',
                    exact_match=True,
                    source_name=spph._lp.source_package_name,
                    version=spph._lp.source_package_version,
                    status='Published')
                spph.current = len(release_publications[:1]) > 0
                if not spph.current:
                    release_publications = archive.getPublishedSources(
                        distro_series=series,
                        pocket='Release',
                        exact_match=True,
                        source_name=spph._lp.source_package_name,
                        version=spph._lp.source_package_version,
                        status='Pending')
                    spph.current = len(release_publications[:1]) > 0
            else:
                spph.current = True

        if not spph.current:
            print("    superseded")

        if main_archive:
            # If this build failure is not a regression versus the
            # main archive, do not report it.
            main_builds = MainArchiveBuilds(main_archive,
                                            spph._lp.source_package_name,
                                            spph._lp.source_package_version)
            try:
                if main_builds.results[arch] != 'Successfully built':
                    print("  Skipping %s" % build.title)
                    continue
            except KeyError:
                pass

        SPPH(csp_link).addBuildLog(build)

    return cur_last_published


def generate_page(name, archive, series, archs_by_archive, main_archive, template = 'build_status.html', arch_list = default_arch_list, notice=None, release_only=False):
    # sort the package lists
    filter_ftbfs = lambda pkglist, current: list(filter(methodcaller('isFTBFS', arch_list, current), sorted(pkglist)))
    data = {}
    for comp in ('main', 'restricted', 'universe', 'multiverse'):
        data[comp] = filter_ftbfs(components[comp], True)
        data['%s_superseded' % comp] = filter_ftbfs(components[comp], False) if not release_only else []
    for pkgset, pkglist in packagesets_ftbfs.items():
        packagesets_ftbfs[pkgset] = filter_ftbfs(pkglist, True)
    for team, pkglist in teams_ftbfs.items():
        teams_ftbfs[team] = filter_ftbfs(pkglist, True)

    # container object to hold the counts and the tooltip
    class StatData(object):
        def __init__(self, cnt, cnt_superseded, tooltip):
            self.cnt = cnt
            self.cnt_superseded = cnt_superseded
            self.tooltip = tooltip

    # compute some statistics (number of packages for each build failure type)
    stats = {}
    for state in ('FAILEDTOBUILD', 'MANUALDEPWAIT', 'CHROOTWAIT', 'UPLOADFAIL', 'CANCELLED'):
        stats[state] = {}
        for arch in arch_list:
            tooltip = []
            cnt = 0
            cnt_sup = 0
            for comp in ('main', 'restricted', 'universe', 'multiverse'):
                s = sum([pkg.getCount(arch, state) for pkg in data[comp]])
                s_sup = sum([pkg.getCount(arch, state) for pkg in data['%s_superseded' % comp]])
                if s or s_sup:
                    cnt += s
                    cnt_sup += s_sup
                    tooltip.append('<td>%s:</td><td style="text-align:right;">%i (%i superseded)</td>' % (comp, s, s_sup))
            if cnt:
                tooltiphtml = '<table><tr>'
                tooltiphtml += '</tr><tr>'.join(tooltip)
                tooltiphtml += '</tr></table>'
                stats[state][arch] = StatData(cnt, cnt_sup, tooltiphtml)
            else:
                stats[state][arch] = StatData(None, None, None)

    data['stats'] = stats
    data['archive'] = archive
    data['main_archive'] = main_archive
    data['series'] = series
    data['arch_list'] = arch_list
    data['archs_by_archive'] = archs_by_archive
    data['lastupdate'] = time.strftime('%F %T %z')
    data['packagesets'] = packagesets_ftbfs
    data['teams'] = teams_ftbfs
    data['notice'] = notice
    data['abbrs'] = {
        'FAILEDTOBUILD': 'F',
        'CANCELLED': 'X',
        'MANUALDEPWAIT': 'M',
        'CHROOTWAIT': 'C',
        'UPLOADFAIL': 'U',
        }

    env = Environment(loader=FileSystemLoader('.'))
    template = env.get_template('build_status.html')
    stream = template.render(**data)

    fn = '../%s.html' % name
    out = open('%s.new' % fn, 'w')
    out.write(stream)
    out.close()
    os.rename('%s.new' % fn, fn)

def generate_csvfile(name, arch_list = default_arch_list):
    csvout = open('../%s.csv' % name, 'w')
    linetemplate = '%(name)s,%(link)s,%(explain)s\n'
    for comp in components.values():
        for pkg in comp:
            for ver in pkg.versions:
                for state in ('FAILEDTOBUILD', 'MANUALDEPWAIT', 'CHROOTWAIT', 'UPLOADFAIL', 'CANCELLED'):
                    archs = [ arch for (arch, log) in ver.logs.items() if log.buildstate == state ]
                    if archs:
                        log = ver.logs[archs[0]].log
                        csvout.write(linetemplate  % {'name': pkg.name, 'link': log,
                            'explain':"[%s] %s" %(', '.join(archs), state)})

def load_timestamps(name):
    '''Load the saved timestamps about the last still published FTBFS build record.'''
    try:
        timestamp_file = open('%s.json' % name, 'r')
        tmp = json.load(timestamp_file)
        timestamps = {}
        for state, timestamp in tmp.items():
            try:
                timestamps[state] = datetime.utcfromtimestamp(int(timestamp))
            except TypeError:
                timestamps[state] = None
        return timestamps
    except (IOError):
        return {
            'Failed to build': None,
            'Dependency wait': None,
            'Chroot problem': None,
            'Failed to upload': None,
            'Cancelled build': None,
        }

def save_timestamps(name, timestamps):
    '''Save the timestamps of the last still published FTBFS build record into a JSON file.'''
    timestamp_file = open('%s.json' % name, 'w')
    tmp = {}
    for state, timestamp in timestamps.items():
        if timestamp is not None:
            tmp[state] = timestamp.strftime('%s')
        else:
            tmp[state] = None
    json.dump(tmp, timestamp_file)
    timestamp_file.close()

if __name__ == '__main__':
    # login anonymously to LP
    launchpad = Launchpad.login_anonymously('qa-ftbfs', lp_service, version=api_version)

    ubuntu = launchpad.distributions['ubuntu']

    usage = "usage: %prog [options] <archive> <series> <arch> [<arch> ...]"
    parser = OptionParser(usage=usage)
    parser.add_option(
        "-f", "--filename", dest="name",
        help="File name prefix for the result.")
    parser.add_option(
        "-n", "--notice", dest="notice_file",
        help="HTML notice file to include in the page header.")
    parser.add_option(
        "--release-only", dest="release_only", action="store_true",
        help="Only include sources currently published in the release pocket.")
    (options, args) = parser.parse_args()
    if len(args) < 3:
        parser.error("Need at least 4 arguments.")

    try:
        archive = ubuntu.getArchive(name=args[0])
    except HTTPError:
        print('Error: %s is not a valid archive.' % args[0])
    try:
        series = ubuntu.getSeries(name_or_version=args[1])
    except HTTPError:
            print('Error: %s is not a valid series.' % args[1])

    if options.name is None:
        options.name = '%s-%s' % (archive.name, series.name)

    if archive.name != 'primary':
        main_archive = ubuntu.main_archive
        main_series = series
    else:
        main_archive = main_series = None

    archs_by_archive = dict(main=[], ports=[])
    for arch in args[2:]:
        das = series.getDistroArchSeries(archtag=arch)
        archs_by_archive[das.official and 'main' or 'ports'].append(arch)
    default_arch_list.extend(archs_by_archive['main'])
    default_arch_list.extend(archs_by_archive['ports'])

    for (archive, series) in [(archive, series)]:
        print("Generating FTBFS for %s" % series.fullseriesname)

        # clear all caches
        PersonTeam.clear()
        SourcePackage.clear()
        SPPH.clear()
        last_published = load_timestamps(options.name)

        # list of SourcePackages for each component
        components = {
                'main': [],
                'restricted': [],
                'universe': [],
                'multiverse': [],
                'partner': [],
                }

        # packagesets for this series
        packagesets = dict()
        packagesets_ftbfs = dict()
        for ps in launchpad.packagesets:
            if ps.distroseries_link == series.self_link:
                packagesets[ps.name] = ps.getSourcesIncluded(direct_inclusion=False)
                packagesets_ftbfs[ps.name] = [] # empty list to add FTBFS for each package set later

        teams = requests.get('https://people.canonical.com/~ubuntu-archive/package-team-mapping.json').json()

        # Per team list of FTBFS
        teams_ftbfs = {team: [] for team in teams}

        for state in ('Failed to build', 'Dependency wait', 'Chroot problem', 'Failed to upload', 'Cancelled build'):
            last_published[state] = fetch_pkg_list(archive, series, state, last_published[state], default_arch_list, main_archive, main_series, options.release_only)

        save_timestamps(options.name, last_published)

        if options.notice_file:
            notice = open(options.notice_file).read()
        else:
            notice = None

        print("Generating HTML page...")
        generate_page(options.name, archive, series, archs_by_archive, main_archive, notice=notice, release_only=options.release_only)
        print("Generating CSV file...")
        generate_csvfile(options.name)