centos-compose-check.py

#!/usr/bin/python

import re

import click
import koji
import requests


KEYID = '8483c65d'


def get_builds(ctx, param, value):
    if not value and not click.get_text_stream('stdin').isatty():
        return click.get_text_stream('stdin').read().strip().split()
    else:
        return value


@click.command()
@click.option('-p', '--profile', default='centos')
@click.option('--missing', is_flag=True)
@click.option('--bad-builds', is_flag=True)
@click.option('--bad-tasks', is_flag=True)
@click.option('--needs-signing', is_flag=True)
@click.option('--needs-tagging', is_flag=True)
@click.option('--ready-c8', is_flag=True)
@click.option('--ready-c8s', is_flag=True)
@click.argument('builds', callback=get_builds, required=False, nargs=-1)
def main(profile, missing, bad_builds, bad_tasks, needs_signing, needs_tagging, ready_c8, ready_c8s, builds):

    # To enable builds being passed in on standard input, builds are not a required argument.  But
    # we need them either as arguments or as standard input, so check that we have one or the other.
    if not builds:
        raise click.BadArgumentUsage('pass in builds as arguments or to standard input')

    # If no filtering flags are passed, show message for every build.
    everything = not any([missing, bad_builds, bad_tasks, needs_signing, needs_tagging, ready_c8, ready_c8s])

    # Set up koji session.
    centos_koji = koji.get_profile_module(profile)
    session = centos_koji.ClientSession(centos_koji.config.server)

    # Ensure kojifiles is accessible.
    try:
        requests.head(centos_koji.config.topurl, timeout=1).raise_for_status()
    except requests.exceptions.RequestException:
        raise click.ClickException('cannot connect to koji')

    # Set formatting width based on the longest build.
    width = len(max(builds, key=len)) + 10

    # Create our output function helper
    def say(build, msg, **kwargs):
        click.echo(f'{build} : '.rjust(width) + click.style(msg, **kwargs))

    def signed(build_info):
        srpm_url = (
            f'{centos_koji.config.topurl}/packages/{build_info["name"]}/{build_info["version"]}/{build_info["release"]}/'
            f'data/signed/{KEYID}/src/{build_info["name"]}-{build_info["version"]}-{build_info["release"]}.src.rpm'
        )
        r = requests.head(srpm_url, verify=centos_koji.config.serverca)
        return r.ok

    # main loop
    for build in builds:
        # Skip right away if the input doesn't look like an NVR.
        try:
            assert re.match('^[\w.+-]+$', build, flags=re.ASCII)
            _, _, _ = build.rsplit('-', 2)
        except (AssertionError, ValueError):
            say(build, f'invalid build name', bg='red')
            continue

        # Get the build information.
        build_info = session.getBuild(build)

        # Verify build exists.
        if not build_info:
            if everything:
                say(build, f'does not exist', fg='bright_black')
            elif missing:
                click.echo(build)
            continue

        # Check build status.
        if build_info['state'] != 1:
            if everything:
                say(build, f'bad build status', fg='red')
            elif bad_builds:
                click.echo(build)
            continue

        # Check task status, if there is one.  Module builds and imported builds won't have one.
        task_info = session.getTaskInfo(build_info['task_id'])
        if task_info:
            if task_info['state'] != 2:
                if everything:
                    say(build, f'bad task state', fg='red')
                elif bad_tasks:
                    click.echo(build)
                continue

        try:
            content_tag = build_info['extra']['typeinfo']['module']['content_koji_tag']
        except (KeyError, TypeError):
            # This is a regular build.

            # Check for a signed SRPM.  The compose should catch if anything is unsigned, but we
            # will key off the SRPM to guess if signing has even been attempted.
            if not signed(build_info):
                if everything:
                    say(build, 'not signed', fg='bright_yellow')
                elif needs_signing:
                    click.echo(build)
                continue

            # Check that the build is tagged appropriately.
            tags = {tag['name'] for tag in session.listTags(build_info['id'])}
            c8s_compose_tags = {'dist-c8-stream', 'dist-c8-stream-compose'}
            c8_compose_tags = c8s_compose_tags.union({'dist-c8-updates', 'dist-c8-compose'})

            if c8_compose_tags.issubset(tags):
                if everything:
                    say(build, 'tagged for c8 compose', fg='bright_blue')
                elif ready_c8:
                    click.echo(build)

            elif c8s_compose_tags.issubset(tags):
                if everything:
                    say(build, 'tagged for c8s compose', fg='bright_cyan')
                elif ready_c8s:
                    click.echo(build)

            else:
                if everything:
                    say(build, 'ready to be tagged for compose', fg='green')
                elif needs_tagging:
                    click.echo(build)

        else:
            # This is a module build.

            # Gather component builds from module content tag.
            component_builds = [
                build_info['nvr']
                for build_info in session.listTagged(content_tag)
            ]

            for component_build in component_builds:
                # Get build info.
                component_build_info = session.getBuild(component_build)

                # Check component build status.
                if component_build_info['state'] != 1:
                    if everything:
                        say(component_build, f'bad module component build status', fg='red')
                    elif bad_builds:
                        click.echo(component_build)
                    continue

                # Check task status, if there is one.
                component_task_info = session.getTaskInfo(component_build_info['task_id'])
                if component_task_info:
                    if component_task_info['state'] != 2:
                        if everything:
                            say(component_build, f'bad module component task state', fg='red')
                        elif bad_tasks:
                            click.echo(component_build)
                        continue

                # Check for a signed SRPM.  The compose should catch if anything is unsigned, but
                # we will key off the SRPM to guess if signing has even been attempted.
                if not signed(component_build_info):
                    if everything:
                        say(component_build, 'module component not signed', fg='bright_yellow')
                    elif needs_signing:
                        click.echo(component_build)
                    continue

            # Check that the module build and it's related devel module are tagged appropriately.
            devel_build = f'{build_info["name"]}-devel-{build_info["version"]}-{build_info["release"]}' 
            for module_build in [build, devel_build]:
                tags = {tag['name'] for tag in session.listTags(module_build)}
                c8s_module_compose_tags = {'modular-updates-candidate', 'dist-c8-stream-module-compose'}
                c8_module_compose_tags = c8s_module_compose_tags.union({'dist-c8-module-compose'})

                if c8_module_compose_tags.issubset(tags):
                    if everything:
                        say(module_build, 'tagged for c8 module compose', fg='bright_blue')
                    elif ready_c8:
                        click.echo(module_build)

                elif c8s_module_compose_tags.issubset(tags):
                    if everything:
                        say(module_build, 'tagged for c8s module compose', fg='bright_cyan')
                    elif ready_c8s:
                        click.echo(module_build)

                else:
                    if everything:
                        say(module_build, 'ready to be tagged for module compose', fg='green')
                    elif needs_tagging:
                        click.echo(module_build)


if __name__ == '__main__':
    main()